Run the environment provider locally (#51)

* Run the environment provider locally

projects/etos_suite_runner/requirements.txt

@@ -16,6 +16,7 @@
 # scipy==1.0
 #
 PyScaffold==3.2.3
-packageurl-python==0.9.1
+packageurl-python~=0.11
 cryptography~=41.0
 etos_lib==4.0.0
+environment_provider@https://github.com/eiffel-community/etos-environment-provider/releases/download/3.0.0/environment_provider-3.0.0-py2.py3-none-any.whl

projects/etos_suite_runner/setup.cfg

@@ -26,9 +26,10 @@ setup_requires = pyscaffold>=3.2a0,<3.3a0
 
 install_requires =
     PyScaffold==3.2.3
-    packageurl-python==0.9.1
+    packageurl-python~=0.11
     cryptography~=41.0
     etos_lib==4.0.0
+    environment_provider@https://github.com/eiffel-community/etos-environment-provider/releases/download/3.0.0/environment_provider-3.0.0-py2.py3-none-any.whl
 
 python_requires = >=3.4
 

projects/etos_suite_runner/src/etos_suite_runner/__init__.py

@@ -15,7 +15,8 @@
 # limitations under the License.
 """ETOS suite runner module."""
 import os
-from importlib.metadata import version, PackageNotFoundError
+from importlib.metadata import PackageNotFoundError, version
+
 from etos_lib.logging.logger import setup_logging
 
 try:
@@ -27,4 +28,5 @@
 BASE_DIR = os.path.dirname(os.path.relpath(__file__))
 DEV = os.getenv("DEV", "false").lower() == "true"
 ENVIRONMENT = "development" if DEV else "production"
+os.environ["ENVIRONMENT_PROVIDER_DISABLE_LOGGING"] = "true"
 setup_logging("ETOS Suite Runner", VERSION, ENVIRONMENT)

projects/etos_suite_runner/src/etos_suite_runner/esr.py

@@ -19,16 +19,15 @@
 import os
 import signal
 import threading
-import time
 import traceback
-from json import JSONDecodeError
 from uuid import uuid4
 
 from eiffellib.events import EiffelActivityTriggeredEvent
+from environment_provider.environment_provider import EnvironmentProvider
+from environment_provider_api.backend.environment import release_full_environment
 from etos_lib import ETOS
 from etos_lib.logging.logger import FORMAT_CONFIG
-from requests.exceptions import ConnectionError as RequestsConnectionError
-from requests.exceptions import HTTPError
+from jsontas.jsontas import JsonTas
 
 from .lib.esr_parameters import ESRParameters
 from .lib.exceptions import EnvironmentProviderException
@@ -62,109 +61,54 @@ def __init__(self) -> None:
             int(os.getenv("ESR_WAIT_FOR_ENVIRONMENT_TIMEOUT")),
         )
 
-    def _request_environment(self, ids: list[str]) -> tuple[str, str]:
+    def _request_environment(self, ids: list[str]) -> None:
         """Request an environment from the environment provider.
 
         :param ids: Generated suite runner IDs used to correlate environments and the suite
                     runners.
-        :return: Task ID and an error message.
         """
-        params = {
-            "suite_id": self.params.tercc.meta.event_id,
-            "suite_runner_ids": ",".join(ids),
-        }
         try:
-            response = self.etos.http.post(self.etos.debug.environment_provider, json=params)
-            response.raise_for_status()
-        except (RequestsConnectionError, HTTPError) as exception:
-            return None, str(exception)
-
-        try:
-            json_response = response.json()
-        except JSONDecodeError:
-            return None, "Could not parse JSON from the environment provider"
-
-        result = json_response.get("result", "")
-        if result.lower() != "success":
-            return (
-                None,
-                "Could not retrieve an environment from the environment provider",
-            )
-        task_id = json_response.get("data", {}).get("id")
-        if task_id is None:
-            return None, "Did not retrieve an environment"
-        return task_id, ""
-
-    def _get_environment_status(self, task_id: str, identifier: str) -> None:
-        """Wait for an environment being provided.
-
-        :param task_id: Task ID to wait for.
-        :param identifier: An identifier to use for logging.
-        """
-        FORMAT_CONFIG.identifier = identifier
-        timeout = self.etos.config.get("WAIT_FOR_ENVIRONMENT_TIMEOUT")
-        end = time.time() + timeout
-        while time.time() < end:
-            response = self.etos.http.get(
-                url=self.etos.debug.environment_provider,
-                timeout=timeout,
-                params={"id": task_id},
-            )
-            response.raise_for_status()
-            json_response = response.json()
-            # dict.get() does not work here as it only sets None if the key does not exist
-            # and sometimes the key exists, but the value is None.
-            result = json_response.get("result") or {}
-            self.params.set_status(json_response.get("status"), result.get("error"))
-            if json_response and result:
-                break
-            time.sleep(5)
-        else:
-            self.params.set_status(
-                "FAILURE",
-                "Unknown Error: Did not receive an environment " f"within {timeout}s",
+            provider = EnvironmentProvider(self.params.tercc.meta.event_id, ids, copy=False)
+            result = provider.run()
+        except Exception:
+            self.params.set_status("FAILURE", "Failed to run environment provider")
+            self.logger.error(
+                "Environment provider has failed in creating an environment for test.",
+                extra={"user_log": True},
             )
-        if self.params.get_status().get("error") is not None:
+            raise
+        if result.get("error") is not None:
+            self.params.set_status("FAILURE", result.get("error"))
             self.logger.error(
                 "Environment provider has failed in creating an environment for test.",
                 extra={"user_log": True},
             )
         else:
+            self.params.set_status("SUCCESS", result.get("error"))
             self.logger.info(
                 "Environment provider has finished creating an environment for test.",
                 extra={"user_log": True},
             )
 
-    def _release_environment(self, task_id: str) -> None:
-        """Release an environment from the environment provider.
-
-        :param task_id: Task ID to release.
-        """
-        response = self.etos.http.get(
-            self.etos.debug.environment_provider, params={"release": task_id}
+    def _release_environment(self) -> None:
+        """Release an environment from the environment provider."""
+        # TODO: We should remove jsontas as a requirement for this function.
+        # Passing variables as keyword argument to make it easier to transition to a function where
+        # jsontas is not required.
+        jsontas = JsonTas()
+        status, message = release_full_environment(
+            etos=self.etos, jsontas=jsontas, suite_id=self.params.tercc.meta.event_id
         )
-        response.raise_for_status()
-
-    def _reserve_workers(self, ids: list[str]) -> str:
-        """Reserve workers for test.
+        if not status:
+            self.logger.error(message)
 
-        :param ids: Generated suite runner IDs used to correlate environments and the suite
-                    runners.
-        :return: The environment provider task ID
-        """
-        self.logger.info("Request environment from environment provider", extra={"user_log": True})
-        task_id, msg = self._request_environment(ids)
-        if task_id is None:
-            raise EnvironmentProviderException(msg, task_id)
-        return task_id
-
-    def run_suites(self, triggered: EiffelActivityTriggeredEvent, tercc_id: str) -> None:
+    def run_suites(self, triggered: EiffelActivityTriggeredEvent) -> list[str]:
         """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.
 
         :param triggered: Activity triggered.
-        :param tercc_id: The ID of the tercc that is going to be executed.
+        :return: List of main suite IDs
         """
         context = triggered.meta.event_id
         self.etos.config.set("context", context)
@@ -180,26 +124,20 @@ def run_suites(self, triggered: EiffelActivityTriggeredEvent, tercc_id: str) ->
             ids.append(suite["test_suite_started_id"])
         self.logger.info("Number of test suites to run: %d", len(ids), extra={"user_log": True})
 
-        task_id = None
         try:
-            self.logger.info("Wait for test environment.")
-            task_id = self._reserve_workers(ids)
-            self.etos.config.set("task_id", task_id)
+            self.logger.info("Get test environment.")
             threading.Thread(
-                target=self._get_environment_status,
-                args=(task_id, tercc_id),
-                daemon=True,
+                target=self._request_environment, args=(ids.copy(),), daemon=True
             ).start()
 
             self.etos.events.send_activity_started(triggered, {"CONTEXT": context})
 
             self.logger.info("Starting ESR.")
             runner.start_suites_and_wait()
-        except EnvironmentProviderException as exception:
-            task_id = exception.task_id
+            return ids
+        except EnvironmentProviderException:
             self.logger.info("Release test environment.")
-            if task_id is not None:
-                self._release_environment(task_id)
+            self._release_environment()
             raise
 
     @staticmethod
@@ -209,8 +147,11 @@ def verify_input() -> None:
         assert os.getenv("SOURCE_HOST"), "SOURCE_HOST environment variable not provided."
         assert os.getenv("TERCC"), "TERCC environment variable not provided."
 
-    def run(self) -> None:
-        """Run the ESR main loop."""
+    def run(self) -> list[str]:
+        """Run the ESR main loop.
+
+        :return: List of test suites (main suites) that were started.
+        """
         tercc_id = None
         try:
             tercc_id = self.params.tercc.meta.event_id
@@ -252,10 +193,11 @@ def run(self) -> None:
             raise
 
         try:
-            self.run_suites(triggered, tercc_id)
+            ids = self.run_suites(triggered)
             self.etos.events.send_activity_finished(
                 triggered, {"conclusion": "SUCCESSFUL"}, {"CONTEXT": context}
             )
+            return ids
         except Exception as exception:  # pylint:disable=broad-except
             reason = str(exception)
             self.logger.exception(

projects/etos_suite_runner/src/etos_suite_runner/lib/runner.py

@@ -17,7 +17,9 @@
 import logging
 from multiprocessing.pool import ThreadPool
 
+from environment_provider_api.backend.environment import release_full_environment
 from etos_lib.logging.logger import FORMAT_CONFIG
+from jsontas.jsontas import JsonTas
 
 from .exceptions import EnvironmentProviderException
 from .suite import TestSuite
@@ -43,16 +45,17 @@ def __init__(self, params, etos):
         self.params = params
         self.etos = etos
 
-    def _release_environment(self, task_id):
-        """Release an environment from the environment provider.
-
-        :param task_id: Task ID to release.
-        :type task_id: str
-        """
-        response = self.etos.http.get(
-            self.etos.debug.environment_provider, params={"release": task_id}, timeout=60
+    def _release_environment(self):
+        """Release an environment from the environment provider."""
+        # TODO: We should remove jsontas as a requirement for this function.
+        # Passing variables as keyword argument to make it easier to transition to a function where
+        # jsontas is not required.
+        jsontas = JsonTas()
+        status, message = release_full_environment(
+            etos=self.etos, jsontas=jsontas, suite_id=self.params.tercc.meta.event_id
         )
-        response.raise_for_status()
+        if not status:
+            self.logger.error(message)
 
     def start_suites_and_wait(self):
         """Get environments and start all test suites."""
@@ -66,10 +69,8 @@ def start_suites_and_wait(self):
             if status.get("error") is not None:
                 raise EnvironmentProviderException(status["error"], self.etos.config.get("task_id"))
         finally:
-            task_id = self.etos.config.get("task_id")
             self.logger.info("Release the full test environment.")
-            if task_id is not None:
-                self._release_environment(task_id)
+            self._release_environment()
 
     def run(self, test_suite):
         """Run test suite runner.

projects/etos_suite_runner/src/etos_suite_runner/lib/suite.py

@@ -20,9 +20,11 @@
 from typing import Iterator
 
 from eiffellib.events import EiffelTestSuiteStartedEvent
+from environment_provider.lib.registry import ProviderRegistry
+from environment_provider_api.backend.environment import release_environment
 from etos_lib import ETOS
 from etos_lib.logging.logger import FORMAT_CONFIG
-from requests.exceptions import HTTPError
+from jsontas.jsontas import JsonTas
 
 from .esr_parameters import ESRParameters
 from .exceptions import EnvironmentProviderException
@@ -120,25 +122,28 @@ def start(self, identifier: str) -> None:
                     self.logger.info("ETOS test runner has finished", extra={"user_log": True})
                     break
         finally:
-            self.release()
+            self.release(identifier)
 
-    def release(self) -> None:
+    def release(self, testrun_id) -> None:
         """Release this sub suite."""
+        # TODO: This whole method is now a bit of a hack that needs to be cleaned up.
+        # Most cleanup is required in the environment provider so this method will stay until an
+        # update has been made there.
         self.logger.info(
             "Check in test environment %r", self.environment["id"], extra={"user_log": True}
         )
-        response = self.etos.http.get(
-            self.etos.debug.environment_provider,
-            params={"single_release": self.environment["id"]},
-            timeout=60,
+        jsontas = JsonTas()
+        registry = ProviderRegistry(etos=self.etos, jsontas=jsontas, suite_id=testrun_id)
+
+        self.logger.info(self.environment)
+        success = release_environment(
+            etos=self.etos, jsontas=jsontas, provider_registry=registry, sub_suite=self.environment
         )
-        try:
-            response.raise_for_status()
-        except HTTPError:
+        if not success:
             self.logger.exception(
                 "Failed to check in %r", self.environment["id"], extra={"user_log": True}
             )
-            raise
+            return
         self.logger.info("Checked in %r", self.environment["id"], extra={"user_log": True})
         self.released = True
 
@@ -179,6 +184,11 @@ def sub_suite_environments(self) -> Iterator[dict]:
                 self.suite["test_suite_started_id"]
             )
             if activity_triggered is None:
+                status = self.params.get_status()
+                if status.get("status") == "FAILURE":
+                    raise EnvironmentProviderException(
+                        status.get("error"), self.etos.config.get("task_id")
+                    )
                 continue
             activity_finished = self.__environment_activity_finished(
                 activity_triggered["meta"]["id"]
@@ -304,6 +314,10 @@ def start(self) -> None:
                     "Environment received. Starting up a sub suite", extra={"user_log": True}
                 )
                 sub_suite_definition = self._download_sub_suite(sub_suite_environment)
+                if sub_suite_definition is None:
+                    raise EnvironmentProviderException(
+                        "URL to sub suite is missing", self.etos.config.get("task_id")
+                    )
                 sub_suite_definition["id"] = sub_suite_environment["meta"]["id"]
                 sub_suite = SubSuite(
                     self.etos, sub_suite_definition, self.suite["test_suite_started_id"]
@@ -344,7 +358,7 @@ def release_all(self) -> None:
         self.logger.info("Releasing all sub suite environments")
         for sub_suite in self.sub_suites:
             if not sub_suite.released:
-                sub_suite.release()
+                sub_suite.release(self.params.tercc.meta.event_id)
         self.logger.info("All sub suite environments are released")
 
     def finish(self, verdict: str, conclusion: str, description: str) -> None:

projects/etos_suite_runner/test-requirements.txt

@@ -1,3 +1,4 @@
 mock
 pytest
 pytest-cov
+mongomock