Add ssev2 client

cli/src/etos_client/etos/v0/etos.py

@@ -14,8 +14,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """ETOS v0."""
+
 import os
 import logging
+import time
 import shutil
 from pathlib import Path
 from json import JSONDecodeError
@@ -132,7 +134,7 @@ def __wait(
             events = test_run.track(
                 self.sse_client,
                 response,
-                24 * 60 * 60,  # 24 hours
+                time.time() + 24 * 60 * 60,  # 24 hours
             )
         except SystemExit as exit:
             clear_queue = False

cli/src/etos_client/etos/v0/test_run/test_run.py

@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """ETOS test run handler."""
+
 import logging
 import sys
 import time
@@ -77,9 +78,8 @@ def __log_debug_information(self, response: ResponseSchema) -> None:
         self.logger.info("Purl: %s", response.artifact_identity)
         self.logger.info("Event repository: %r", response.event_repository)
 
-    def track(self, sse_client: SSEClient, response: ResponseSchema, timeout: int) -> Events:
+    def track(self, sse_client: SSEClient, response: ResponseSchema, end: float) -> Events:
         """Track, and wait for, an ETOS test run."""
-        end = time.time() + timeout
 
         self.__log_debug_information(response)
         last_log = time.time()

cli/src/etos_client/etos/v1alpha/etos.py

@@ -14,27 +14,31 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """ETOS v1alpha."""
+
 import os
 import logging
+import time
 import shutil
 from pathlib import Path
 from json import JSONDecodeError
-from typing import Optional
+from typing import Optional, Union
 
 from requests.exceptions import HTTPError
 from etos_lib.lib.http import Http
-from urllib3.util import Retry
 from etos_lib import ETOS as ETOSLibrary
+from urllib3.util import Retry
 
 from etos_client.types.result import Result, Verdict, Conclusion
 from etos_client.shared.downloader import Downloader
 from etos_client.shared.utilities import directories
-from etos_client.sse.v1.client import SSEClient
+from etos_client.sse.v2alpha.client import SSEClient as SSEV2AlphaClient, TokenExpired
+from etos_client.sse.v1.client import SSEClient as SSEV1Client
 
+from etos_client.etos.v0.test_run import TestRun as V0TestRun
+from etos_client.etos.v0.event_repository import graphql
 from etos_client.etos.v0.events.collector import Collector
 from etos_client.etos.v0.test_results import TestResults
-from etos_client.etos.v0.event_repository import graphql
-from etos_client.etos.v0.test_run import TestRun
+from etos_client.etos.v1alpha.test_run import TestRun as V1AlphaTestRun
 from etos_client.etos.v1alpha.schema.response import ResponseSchema
 from etos_client.etos.v1alpha.schema.request import RequestSchema
 
@@ -61,17 +65,37 @@ class Etos:
 
     version = "v1alpha"
     logger = logging.getLogger(__name__)
+    __apikey = None
     start_response = ResponseSchema
     start_request = RequestSchema
 
-    def __init__(self, args: dict, sse_client: SSEClient):
+    def __init__(self, args: dict, sse_client: Union[SSEV1Client, SSEV2AlphaClient]):
         """Set up sse client and cluster variables."""
         self.args = args
         self.cluster = args.get("<cluster>")
         assert self.cluster is not None
         self.sse_client = sse_client
         self.logger.info("Running ETOS version %s", self.version)
 
+    @property
+    def apikey(self) -> str:
+        """Generate and return an API key."""
+        if self.__apikey is None:
+            http = Http(retry=HTTP_RETRY_PARAMETERS, timeout=10)
+            url = f"{self.cluster}/keys/v1alpha/generate"
+            response = http.post(
+                url,
+                json={"identity": "etos-client", "scope": "post-testrun delete-testrun get-sse"},
+            )
+            try:
+                response.raise_for_status()
+                response_json = response.json()
+            except HTTPError:
+                self.logger.exception("Failed to generate an API key for ETOS.")
+                response_json = {}
+            self.__apikey = response_json.get("token")
+        return self.__apikey or ""
+
     def run(self) -> Result:
         """Run ETOS v1alpha."""
         error = self.__check()
@@ -81,20 +105,7 @@ def run(self) -> Result:
         if error is not None:
             return Result(verdict=Verdict.INCONCLUSIVE, conclusion=Conclusion.FAILED, reason=error)
         assert response is not None
-        (success, msg), error = self.__wait(response)
-        if error is not None:
-            return Result(verdict=Verdict.INCONCLUSIVE, conclusion=Conclusion.FAILED, reason=error)
-        if success is None or msg is None:
-            return Result(
-                verdict=Verdict.INCONCLUSIVE,
-                conclusion=Conclusion.FAILED,
-                reason="No test result received from ETOS testrun",
-            )
-        return Result(
-            verdict=Verdict.PASSED if success else Verdict.FAILED,
-            conclusion=Conclusion.SUCCESSFUL,
-            reason=msg,
-        )
+        return self.__wait(response)
 
     def __start(self) -> tuple[Optional[ResponseSchema], Optional[str]]:
         """Trigger ETOS, retrying on non-client errors until successful or timeout."""
@@ -104,7 +115,9 @@ def __start(self) -> tuple[Optional[ResponseSchema], Optional[str]]:
 
         response_json = {}
         http = Http(retry=HTTP_RETRY_PARAMETERS, timeout=10)
-        response = http.post(url, json=request.model_dump())
+        response = http.post(
+            url, json=request.model_dump(), headers={"Authorization": f"Bearer {self.apikey}"}
+        )
         try:
             response.raise_for_status()
             response_json = response.json()
@@ -120,30 +133,40 @@ def __start(self) -> tuple[Optional[ResponseSchema], Optional[str]]:
             )
         return self.start_response.from_response(response_json), None
 
-    def __wait(
-        self, response: ResponseSchema
-    ) -> tuple[tuple[Optional[bool], Optional[str]], Optional[str]]:
+    def __wait(self, response: ResponseSchema) -> Result:
         """Wait for ETOS to finish."""
-        etos_library = ETOSLibrary("ETOS Client", os.getenv("HOSTNAME"), "ETOS Client")
-        os.environ["ETOS_GRAPHQL_SERVER"] = response.event_repository
-
         report_dir, artifact_dir = directories(self.args)
 
-        collector = Collector(etos_library, graphql)
         log_downloader = Downloader()
         clear_queue = True
         log_downloader.start()
+
+        end = time.time() + 24 * 60 * 60  # 24 hours
+
+        if isinstance(self.sse_client, SSEV2AlphaClient):
+            test_run = V1AlphaTestRun(log_downloader, report_dir, artifact_dir)
+        else:
+            etos_library = ETOSLibrary("ETOS Client", os.getenv("HOSTNAME"), "ETOS Client")
+            os.environ["ETOS_GRAPHQL_SERVER"] = response.event_repository
+            collector = Collector(etos_library, graphql)
+            test_run = V0TestRun(collector, log_downloader, report_dir, artifact_dir)
+        test_run.setup_logging(self.args["-v"])
+        result = None
         try:
-            test_run = TestRun(collector, log_downloader, report_dir, artifact_dir)
-            test_run.setup_logging(self.args["-v"])
-            events = test_run.track(
-                self.sse_client,
-                response,
-                24 * 60 * 60,  # 24 hours
-            )
-        except SystemExit as exit:
-            clear_queue = False
-            return (False, None), str(exit)
+            while time.time() < end:
+                try:
+                    if isinstance(self.sse_client, SSEV2AlphaClient):
+                        result = self.__track(test_run, response, end)
+                    else:
+                        result = self.__track_v0(test_run, response, end)
+                except TokenExpired:
+                    self.__apikey = None
+                    continue
+                except SystemExit as exit:
+                    clear_queue = False
+                    result = Result(
+                        verdict=Verdict.INCONCLUSIVE, conclusion=Conclusion.FAILED, reason=str(exit)
+                    )
         finally:
             log_downloader.stop(clear_queue)
             log_downloader.join()
@@ -159,8 +182,52 @@ def __wait(
         self.logger.info("Artifacts: %s", artifact_dir)
 
         if log_downloader.failed:
-            return (False, None), "ETOS logs did not download successfully"
-        return TestResults().get_results(events), None
+            return Result(
+                verdict=Verdict.INCONCLUSIVE,
+                conclusion=Conclusion.FAILED,
+                reason="ETOS logs did not download succesfully",
+            )
+        if result is not None:
+            return result
+        return Result(
+            verdict=Verdict.INCONCLUSIVE,
+            conclusion=Conclusion.INCONCLUSIVE,
+            reason="Got no result from ETOS so could not determine test result.",
+        )
+
+    def __track(self, test_run: V1AlphaTestRun, response: ResponseSchema, end: float) -> Result:
+        """Track a testrun."""
+        shutdown = test_run.track(
+            self.sse_client,
+            self.apikey,
+            response,
+            end,
+        )
+        return Result(
+            verdict=Verdict(shutdown.data.verdict.upper()),
+            conclusion=Conclusion(shutdown.data.conclusion.upper()),
+            reason=shutdown.data.description,
+        )
+
+    def __track_v0(self, test_run: V0TestRun, response: ResponseSchema, end: float) -> Result:
+        """Track a testrun using the v0 testrun handler."""
+        events = test_run.track(
+            self.sse_client,
+            response,
+            end,
+        )
+        success, msg = TestResults().get_results(events)
+        if success is None or msg is None:
+            return Result(
+                verdict=Verdict.INCONCLUSIVE,
+                conclusion=Conclusion.FAILED,
+                reason="No test result received from ETOS testrun",
+            )
+        return Result(
+            verdict=Verdict.PASSED if success else Verdict.FAILED,
+            conclusion=Conclusion.SUCCESSFUL,
+            reason=msg,
+        )
 
     def __check(self) -> Optional[str]:
         """Check connection to ETOS."""

cli/src/etos_client/etos/v1alpha/subcommands/start.py

@@ -15,14 +15,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Command line for starting ETOSv1alpha testruns."""
+
 import sys
 import os
 import logging
 import warnings
 
 from etos_client.types.result import Conclusion, Verdict
 from etos_client.etos.v1alpha.etos import Etos
-from etos_client.sse.v1.client import SSEClient
+from etos_client.sse.v2alpha.client import SSEClient as SSEV2AlphaClient
+from etos_client.sse.v1.client import SSEClient as SSEV1Client
 
 from etosctl.command import SubCommand
 from etosctl.models import CommandMeta
@@ -46,6 +48,7 @@ class Start(SubCommand):
         --log-area-provider LOG_AREA_PROVIDER                     Which log area provider to use.
         --dataset DATASET                                         Additional dataset information to the environment provider.
                                                                   Check with your provider which information can be supplied.
+        --ssev2alpha                                              Use the v2alpha version of sse.
         --version                                                 Show program's version number and exit
     """
 
@@ -69,7 +72,19 @@ def run(self, args: dict) -> None:
         warnings.warn("This is an alpha version of ETOS! Don't expect it to work properly")
         self.logger.info("Running in cluster: %r", args["<cluster>"])
 
-        etos = Etos(args, SSEClient(args["<cluster>"]))
+        filter = [
+            "message.info",
+            "message.warning",
+            "message.error",
+            "message.critical",
+            "report.*",
+            "artifact.*",
+            "shutdown.*",
+        ]
+        if args["--ssev2alpha"]:
+            etos = Etos(args, SSEV2AlphaClient(args["<cluster>"], filter))
+        else:
+            etos = Etos(args, SSEV1Client(args["<cluster>"]))
         result = etos.run()
         if result.conclusion == Conclusion.FAILED:
             sys.exit(result.reason)

cli/src/etos_client/etos/v1alpha/test_run/__init__.py

@@ -0,0 +1,17 @@
+# Copyright 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.
+"""ETOS test run module."""
+from .test_run import TestRun