ESR now listens to the environment defined events from EP (#25)

* ESR now listens to the environment defined events from EP

There are a few TODOs in the code. These will be fixed when
we continue with this path. Some requirements in different parts
of ETOS that need to be implemented to get there.

* Change FAILED to FAILURE to align with celery

* ESR will pass its test suite started events to environment provider

* More robust waiting for environments and sub suites

* Make the log filter respect args

* Using f-strings instead

* Add documentation strings for newly added parameters

* Fix a few documentation errors and an uneccesary if statement

* Clarify that run_suites starts multiple suites

docs/conf.py

@@ -92,8 +92,8 @@
 master_doc = "index"
 
 # General information about the project.
-project = u"etos_suite_runner"
-copyright = u"2020, Axis Communications AB"
+project = "etos_suite_runner"
+copyright = "Axis Communications AB"
 
 # The version info for the project you're documenting, acts as replacement for
 # |version| and |release|, also used in various other places throughout the
@@ -241,7 +241,7 @@
 # Grouping the document tree into LaTeX files. List of tuples
 # (source start file, target name, title, author, documentclass [howto/manual]).
 latex_documents = [
-    ("index", "user_guide.tex", u"etos_suite_runner Documentation", u"", "manual"),
+    ("index", "user_guide.tex", "etos_suite_runner Documentation", "", "manual"),
 ]
 
 # The name of an image file (relative to this directory) to place at the top of

pylintrc

@@ -1,3 +1,3 @@
 [messages control]
 disable=
-    duplicate-code
+    duplicate-code,fixme

src/etos_suite_runner/__main__.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python
-# Copyright 2020-2021 Axis Communications AB.
+# Copyright 2020-2022 Axis Communications AB.
 #
 # For a full list of individual contributors, please see the commit history.
 #
@@ -20,12 +20,15 @@
 import logging
 import traceback
 import signal
+import threading
+from uuid import uuid4
 
 from etos_lib import ETOS
 from etos_lib.logging.logger import FORMAT_CONFIG
 
 from etos_suite_runner.lib.runner import SuiteRunner
 from etos_suite_runner.lib.esr_parameters import ESRParameters
+from etos_suite_runner.lib.exceptions import EnvironmentProviderException
 
 # Remove spam from pika.
 logging.getLogger("pika").setLevel(logging.WARNING)
@@ -34,15 +37,6 @@
 BASE_DIR = os.path.dirname(os.path.relpath(__file__))
 
 
-class EnvironmentProviderException(Exception):
-    """Exception from EnvironmentProvider."""
-
-    def __init__(self, msg, task_id):
-        """Initialize with task_id."""
-        self.task_id = task_id
-        super().__init__(msg)
-
-
 class ESR:  # pylint:disable=too-many-instance-attributes
     """Suite runner for ETOS main program.
 
@@ -67,13 +61,19 @@ def __init__(self):
             int(os.getenv("ESR_WAIT_FOR_ENVIRONMENT_TIMEOUT")),
         )
 
-    def _request_environment(self):
+    def _request_environment(self, ids):
         """Request an environment from the environment provider.
 
+        :param ids: Generated suite runner IDs used to correlate environments and the suite
+                    runners.
+        :type ids: list
         :return: Task ID and an error message.
         :rtype: tuple
         """
-        params = {"suite_id": self.params.tercc.meta.event_id}
+        params = {
+            "suite_id": self.params.tercc.meta.event_id,
+            "suite_runner_ids": ",".join(ids),
+        }
         wait_generator = self.etos.http.retry(
             "POST", self.etos.debug.environment_provider, json=params
         )
@@ -92,13 +92,11 @@ def _request_environment(self):
             return None, str(exception)
         return task_id, ""
 
-    def _wait_for_environment(self, task_id):
+    def _get_environment_status(self, task_id):
         """Wait for an environment being provided.
 
         :param task_id: Task ID to wait for.
         :type task_id: str
-        :return: Environment and an error message.
-        :rtype: tuple
         """
         timeout = self.etos.config.get("WAIT_FOR_ENVIRONMENT_TIMEOUT")
         wait_generator = self.etos.utils.wait(
@@ -107,30 +105,26 @@ def _wait_for_environment(self, task_id):
             timeout=timeout,
             params={"id": task_id},
         )
-        environment = None
         result = {}
         response = None
         for generator in wait_generator:
             for response in generator:
-                result = response.get("result", {})
-                if response and result and result.get("error") is None:
-                    environment = response
+                result = (
+                    response.get("result", {})
+                    if response.get("result") is not None
+                    else {}
+                )
+                self.params.set_status(response.get("status"), result.get("error"))
+                if response and result:
                     break
-                if result and result.get("error"):
-                    return None, result.get("error")
-            if environment is not None:
+            if response and result:
                 break
         else:
-            if result and result.get("error"):
-                return None, result.get("error")
-            return (
-                None,
-                (
-                    "Unknown Error: Did not receive an environment "
-                    f"within {self.etos.debug.default_http_timeout}s"
-                ),
+            self.params.set_status(
+                "FAILURE",
+                "Unknown Error: Did not receive an environment "
+                f"within {self.etos.debug.default_http_timeout}s",
             )
-        return environment, ""
 
     def _release_environment(self, task_id):
         """Release an environment from the environment provider.
@@ -145,21 +139,23 @@ def _release_environment(self, task_id):
             if response:
                 break
 
-    def _reserve_workers(self):
-        """Reserve workers for test."""
+    def _reserve_workers(self, ids):
+        """Reserve workers for test.
+
+        :param ids: Generated suite runner IDs used to correlate environments and the suite
+                    runners.
+        :type ids: list
+        :return: The environment provider task ID
+        :rtype: str
+        """
         LOGGER.info("Request environment from environment provider")
-        task_id, msg = self._request_environment()
+        task_id, msg = self._request_environment(ids)
         if task_id is None:
             raise EnvironmentProviderException(msg, task_id)
+        return task_id
 
-        LOGGER.info("Wait for environment to become ready.")
-        environment, msg = self._wait_for_environment(task_id)
-        if environment is None:
-            raise EnvironmentProviderException(msg, task_id)
-        return environment, task_id
-
-    def run_suite(self, triggered):
-        """Trigger an activity and starts the actual test runner.
+    def run_suites(self, triggered):
+        """Start up a suite runner handling multiple suites that execute within test runners.
 
         Will only start the test activity if there's a 'slot' available.
 
@@ -173,22 +169,30 @@ def run_suite(self, triggered):
         )
         runner = SuiteRunner(self.params, self.etos, context)
 
+        ids = []
+        for suite in self.params.test_suite:
+            suite["test_suite_started_id"] = str(uuid4())
+            ids.append(suite["test_suite_started_id"])
+
         task_id = None
         try:
             LOGGER.info("Wait for test environment.")
-            environment, task_id = self._reserve_workers()
+            task_id = self._reserve_workers(ids)
+            self.etos.config.set("task_id", task_id)
+            threading.Thread(
+                target=self._get_environment_status, args=(task_id,), daemon=True
+            ).start()
 
             self.etos.events.send_activity_started(triggered, {"CONTEXT": context})
 
             LOGGER.info("Starting ESR.")
-            runner.run(environment.get("result"))
+            runner.start_suites_and_wait()
         except EnvironmentProviderException as exception:
             task_id = exception.task_id
-            raise
-        finally:
             LOGGER.info("Release test environment.")
             if task_id is not None:
                 self._release_environment(task_id)
+            raise
 
     @staticmethod
     def verify_input():
@@ -239,7 +243,7 @@ def run(self):
             raise
 
         try:
-            self.run_suite(triggered)
+            self.run_suites(triggered)
             self.etos.events.send_activity_finished(
                 triggered, {"conclusion": "SUCCESSFUL"}, {"CONTEXT": context}
             )

src/etos_suite_runner/lib/esr_parameters.py

@@ -1,4 +1,4 @@
-# Copyright 2020 Axis Communications AB.
+# Copyright 2020-2022 Axis Communications AB.
 #
 # For a full list of individual contributors, please see the commit history.
 #
@@ -17,6 +17,7 @@
 import os
 import json
 import logging
+from threading import Lock
 
 from packageurl import PackageURL
 from eiffellib.events import EiffelTestExecutionRecipeCollectionCreatedEvent
@@ -44,11 +45,20 @@ class ESRParameters:
     """Parameters required for ESR."""
 
     logger = logging.getLogger("ESRParameters")
+    lock = Lock()
+    __test_suite = None
 
     def __init__(self, etos):
         """ESR parameters instance."""
         self.etos = etos
         self.issuer = {"name": "ETOS Suite Runner"}
+        self.environment_status = {"status": "NOT_STARTED", "error": None}
+
+    def set_status(self, status, error):
+        """Set environment provider status."""
+        with self.lock:
+            self.environment_status["status"] = status
+            self.environment_status["error"] = error
 
     def get_node(self, response):
         """Get a single node from a GraphQL response.
@@ -106,6 +116,28 @@ def tercc(self):
             self.etos.config.set("tercc", tercc)
         return self.etos.config.get("tercc")
 
+    @property
+    def test_suite(self):
+        """Download and return test batches.
+
+        :return: Batches.
+        :rtype: list
+        """
+        with self.lock:
+            if self.__test_suite is None:
+                tercc = self.tercc.json
+                batch_uri = tercc.get("data", {}).get("batchesUri")
+                json_header = {"Accept": "application/json"}
+                json_response = self.etos.http.wait_for_request(
+                    batch_uri,
+                    headers=json_header,
+                )
+                response = {}
+                for response in json_response:
+                    break
+                self.__test_suite = response
+        return self.__test_suite if self.__test_suite else []
+
     @property
     def product(self):
         """Product name from artifact created event.

src/etos_suite_runner/lib/exceptions.py

@@ -0,0 +1,25 @@
+# Copyright 2022 Axis Communications AB.
+#
+# For a full list of individual contributors, please see the commit history.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""ESR exceptions."""
+
+
+class EnvironmentProviderException(Exception):
+    """Exception from EnvironmentProvider."""
+
+    def __init__(self, msg, task_id):
+        """Initialize with task_id."""
+        self.task_id = task_id
+        super().__init__(msg)

src/etos_suite_runner/lib/executor.py

@@ -1,4 +1,4 @@
-# Copyright 2020 Axis Communications AB.
+# Copyright 2020-2022 Axis Communications AB.
 #
 # For a full list of individual contributors, please see the commit history.
 #
@@ -52,8 +52,8 @@ def __auth(username, password, type="basic"):  # pylint:disable=redefined-builti
     def run_tests(self, test_suite):
         """Run tests in jenkins.
 
-        :param test_file: Tests to execute.
-        :type test_file: dict
+        :param test_suite: Tests to execute.
+        :type test_suite: dict
         """
         executor = test_suite.get("executor")
         request = executor.get("request")