ETR to wait for environment (#55)

src/etos_test_runner/etr.py

@@ -16,21 +16,20 @@
 # limitations under the License.
 # -*- coding: utf-8 -*-
 """ETOS test runner module."""
-import argparse
 import sys
 import logging
 import os
 import signal
 import importlib
 import pkgutil
+from typing import Optional, Union
 from pprint import pprint
 from collections import OrderedDict
 
 from etos_lib import ETOS
 from etos_lib.logging.logger import FORMAT_CONFIG
 from jsontas.jsontas import JsonTas
 
-from etos_test_runner import VERSION
 from etos_test_runner.lib.testrunner import TestRunner
 from etos_test_runner.lib.iut import Iut
 from etos_test_runner.lib.custom_dataset import CustomDataset
@@ -43,28 +42,12 @@
 _LOGGER = logging.getLogger(__name__)
 
 
-def parse_args(args):
-    """Parse command line parameters.
-
-    :param args: command line parameters as list of strings
-    :return: command line parameters as :obj:`airgparse.Namespace`
-    """
-    parser = argparse.ArgumentParser(description="ETOS test runner")
-    parser.add_argument(
-        "-v",
-        "--version",
-        action="version",
-        version=f"etos_test_runner {VERSION}",
-    )
-    return parser.parse_args(args)
-
-
 class ETR:
     """ETOS Test Runner."""
 
     context = None
 
-    def __init__(self):
+    def __init__(self) -> None:
         """Initialize ETOS library and start eiffel publisher."""
         self.etos = ETOS("ETOS Test Runner", os.getenv("HOSTNAME"), "ETOS Test Runner")
         if os.getenv("ETOS_ENCRYPTION_KEY"):
@@ -85,15 +68,14 @@ def __init__(self):
         signal.signal(signal.SIGTERM, self.graceful_shutdown)
 
     @staticmethod
-    def graceful_shutdown(*args):
+    def graceful_shutdown(*_) -> None:
         """Catch sigterm."""
         raise Exception("ETR has been terminated.")  # pylint:disable=broad-exception-raised
 
-    def download_and_load(self, sub_suite_url):
+    def download_and_load(self, sub_suite_url: str) -> None:
         """Download and load test json.
 
         :param sub_suite_url: URL to where the sub suite information exists.
-        :type sub_suite_url: str
         """
         response = self.etos.http.get(sub_suite_url)
         json_config = response.json(object_pairs_hook=OrderedDict)
@@ -112,17 +94,16 @@ def download_and_load(self, sub_suite_url):
         self.etos.config.set("main_suite_id", config.get("test_suite_started_id"))
         self.etos.config.set("suite_id", config.get("suite_id"))
 
-    def _run_tests(self):
+    def _run_tests(self) -> Union[int, dict]:
         """Run tests in ETOS test runner.
 
         :return: Results of test runner execution.
-        :rtype: bool
         """
         iut = Iut(self.etos.config.get("test_config").get("iut"))
         test_runner = TestRunner(iut, self.etos)
         return test_runner.execute()
 
-    def load_plugins(self):
+    def load_plugins(self) -> None:
         """Load plugins from environment using the name etr_."""
         disable_plugins = os.getenv("DISABLE_PLUGINS")
         disabled_plugins = []
@@ -142,13 +123,11 @@ def load_plugins(self):
             plugins.append(module.ETRPlugin(self.etos))
         self.etos.config.set("plugins", plugins)
 
-    def get_sub_suite_url(self, environment_id):
+    def get_sub_suite_url(self, environment_id: str) -> Optional[str]:
         """Get sub suite from ETOS environment defined event.
 
         :param environment_id: ID of th environment defined event.
-        :type environment_id: str
         :return: URL for sub suite.
-        :rtype: str
         """
         query = (
             """
@@ -166,6 +145,8 @@ def get_sub_suite_url(self, environment_id):
         """
             % environment_id
         )
+        # Timeout can be configured using ETOS_DEFAULT_WAIT_TIMEOUT environment variable
+        # Default timeout is 60s.
         wait_generator = self.etos.utils.wait(self.etos.graphql.execute, query=query)
         for response in wait_generator:
             if response:
@@ -174,15 +155,14 @@ def get_sub_suite_url(self, environment_id):
                         self.etos.graphql.search_for_nodes(response, "environmentDefined")
                     )
                 except StopIteration:
-                    return None
+                    continue
                 return environment_defined["data"]["uri"]
         return None
 
-    def run_etr(self):
+    def run_etr(self) -> Union[int, dict]:
         """Send activity events and run ETR.
 
         :return: Result of testrunner execution.
-        :rtype: bool
         """
         _LOGGER.info("Starting ETR.")
         sub_suite_url = self.get_sub_suite_url(self.environment_id)
@@ -208,10 +188,8 @@ def run_etr(self):
         return result
 
 
-def main(args):
+def main() -> None:
     """Start ETR."""
-    args = parse_args(args)
-
     etr = ETR()
     result = etr.run_etr()
     if isinstance(result, dict):
@@ -224,7 +202,7 @@ def main(args):
 def run():
     """Entry point to ETR."""
     logging.basicConfig(level=logging.INFO, stream=sys.stdout)
-    main(sys.argv[1:])
+    main()
 
 
 if __name__ == "__main__":