Use the event repository from new ETOS API. (#13)

Author: t-persson

src/etos_client/__main__.py

@@ -63,6 +63,10 @@ def parse_args(args):
     parser = argparse.ArgumentParser(
         description="Client for executing test automation suites in ETOS"
     )
+    parser.add_argument(
+        "cluster",
+        help="Cluster is the URL to the ETOS API.",
+    )
     parser.add_argument(
         "-i",
         "--identity",
@@ -124,11 +128,6 @@ def parse_args(args):
         ),
     )
 
-    parser.add_argument(
-        "--cluster",
-        default=Debug().etos_api,
-        help="Cluster should be in the form of URL to tester-api.",
-    )
     parser.add_argument(
         "--version",
         action="version",
@@ -165,7 +164,7 @@ def check_etos_connectivity(url):
     :type url: str
     """
     try:
-        response = requests.head(url, timeout=5)
+        response = requests.get(url, timeout=5)
         response.raise_for_status()
     except Exception as exception:  # pylint:disable=broad-except
         raise Exception(
@@ -273,7 +272,6 @@ def main(args):
     """
     args = parse_args(args)
     etos = ETOS("ETOS Client", os.getenv("HOSTNAME"), "ETOS Client")
-    os.environ["ETOS_TESTER_API"] = args.cluster
 
     setup_logging(args.loglevel)
     info = generate_spinner(args.no_tty)
@@ -285,11 +283,11 @@ def main(args):
     etos.config.set("dataset", json.loads(args.dataset))
 
     with info(text="Checking connectivity to ETOS", spinner="dots") as spinner:
-        spinner.info("Running in cluster: '{}'".format(etos.debug.etos_api))
+        spinner.info("Running in cluster: '{}'".format(args.cluster))
         spinner.info("Configuration:")
         spinner.info("{}".format(etos.config.config))
         try:
-            check_etos_connectivity(etos.debug.etos_api)
+            check_etos_connectivity(f"{args.cluster}/selftest/ping")
         except Exception as exception:  # pylint:disable=broad-except
             spinner.fail(str(exception))
             sys.exit(1)
@@ -298,7 +296,7 @@ def main(args):
         spinner.succeed("Ready to launch ETOS.")
 
         # Start execution
-        etos_client = ETOSClient(etos)
+        etos_client = ETOSClient(etos, args.cluster)
         spinner.start("Triggering ETOS.")
         success = etos_client.start(spinner)
         if not success:
@@ -307,6 +305,8 @@ def main(args):
             sys.exit(not success)
         spinner.info("Suite ID: {}".format(etos_client.test_suite_id))
         etos.config.set("suite_id", etos_client.test_suite_id)
+        os.environ["ETOS_GRAPHQL_SERVER"] = etos_client.event_repository
+        spinner.info("Event repository: '{}'".format(etos.debug.graphql_server))
 
         # Wait for test results
         test_result_handler = ETOSTestResultHandler(etos)

src/etos_client/client.py

@@ -21,14 +21,19 @@
 
 class ETOSClient:
     """Client for starting test suites in ETOS."""
+    event_repository = None
+    test_suite_id = None
 
-    def __init__(self, etos):
+    def __init__(self, etos, cluster):
         """Initialize ETOS client.
 
         :param etos: ETOS Library instance.
         :type etos: :obj:`etos_lib.etos.ETOS`
+        :param cluster: ETOS cluster to start tests in.
+        :type cluster: str
         """
         self.etos = etos
+        self.cluster = cluster
         self.test_execution = {}
 
     @property
@@ -55,24 +60,18 @@ def start(self, spinner):
         """
         spinner.info(str(self.data))
         generator = self.etos.http.retry(
-            "POST", self.etos.debug.etos_api, timeout=30, json=self.data
+            "POST", f"{self.cluster}/etos", timeout=30, json=self.data
         )
         response = None
         try:
             for response in generator:
                 self.test_execution = response
+                self.test_suite_id = response.get("tercc")
+                self.event_repository = response.get("event_repository")
                 break
         except ConnectionError as exception:
-            spinner.warning(str(exception))
+            spinner.warn(str(exception))
             spinner.fail("Failed to trigger ETOS.")
             return False
         spinner.succeed("ETOS triggered.")
         return True
-
-    @property
-    def test_suite_id(self):
-        """Test suite ID on TERCC."""
-        recipe = self.test_execution.get(
-            "EiffelTestExecutionRecipeCollectionCreatedEvent"
-        )
-        return recipe.get("meta", {}).get("id")