ETOS must understand exit codes from the test command

src/etos_test_runner/lib/executor.py

@@ -99,6 +99,7 @@ def __init__(self, test, iut, etos):
         self.context = self.etos.config.get("context")
         self.plugins = self.etos.config.get("plugins")
         self.result = True
+        self.returncode = None
 
     def load_regex(self):
         """Attempt to load regex file from environment variables.
@@ -404,11 +405,21 @@ def _execute(self, workspace):
 
             self.logger.info("Wait for test to finish.")
             # We must consume the iterator here, even if we do not parse the lines.
-            for _, line in iterator:
+            proc = None
+            line = ""
+            for proc, line in iterator:
                 if self.test_regex:
                     self.parse(line)
             self.result = line
-            self.logger.info("Finished with result %r.", self.result)
+            if proc is not None:
+                self.returncode = proc.returncode
+                self.logger.info(
+                    "Finished with result %r, exit code: %d",
+                    self.result,
+                    self.returncode,
+                )
+            else:
+                self.logger.info("Finished with result %r", self.result)
 
     def execute(self, workspace, retries=3):
         """Retry execution of test cases.

src/etos_test_runner/lib/testrunner.py

@@ -14,20 +14,25 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """ETR test runner module."""
+import json
 import time
 import os
 import logging
 from pprint import pprint
+from typing import Union
 
 from etos_test_runner.lib.iut_monitoring import IutMonitoring
 from etos_test_runner.lib.executor import Executor
 from etos_test_runner.lib.workspace import Workspace
 from etos_test_runner.lib.log_area import LogArea
+from etos_test_runner.lib.verdict import CustomVerdictMatcher
 
 
 class TestRunner:
     """Test runner for ETOS."""
 
+    # pylint: disable=too-many-instance-attributes
+
     logger = logging.getLogger("ETR")
 
     def __init__(self, iut, etos):
@@ -48,6 +53,15 @@ def __init__(self, iut, etos):
         self.etos.config.set("iut", self.iut)
         self.plugins = self.etos.config.get("plugins")
 
+        verdict_rule_file = os.getenv("VERDICT_RULE_FILE")
+        if verdict_rule_file is not None:
+            with open(verdict_rule_file, "r", encoding="utf-8") as inp:
+                rules = json.load(inp)
+        else:
+            rules = []
+
+        self.verdict_matcher = CustomVerdictMatcher(rules)
+
     def test_suite_started(self):
         """Publish a test suite started event.
 
@@ -91,7 +105,7 @@ def environment(self, context):
                 host={"name": os.getenv("EXECUTION_SPACE_URL"), "user": "etos"},
             )
 
-    def run_tests(self, workspace):
+    def run_tests(self, workspace: Workspace) -> tuple[bool, list[Union[int, None]]]:
         """Execute test recipes within a test executor.
 
         :param workspace: Which workspace to execute test suite within.
@@ -101,18 +115,29 @@ def run_tests(self, workspace):
         """
         recipes = self.config.get("recipes")
         result = True
+        test_framework_exit_codes = []
         for num, test in enumerate(recipes):
             self.logger.info("Executing test %s/%s", num + 1, len(recipes))
             with Executor(test, self.iut, self.etos) as executor:
                 self.logger.info("Starting test '%s'", executor.test_name)
                 executor.execute(workspace)
-
                 if not executor.result:
                     result = executor.result
-                self.logger.info("Test finished. Result: %s.", executor.result)
-        return result
+                self.logger.info(
+                    "Test finished. Result: %s. Test framework exit code: %d",
+                    executor.result,
+                    executor.returncode,
+                )
+                test_framework_exit_codes.append(executor.returncode)
+        return result, test_framework_exit_codes
 
-    def outcome(self, result, executed, description):
+    def outcome(
+        self,
+        result: bool,
+        executed: bool,
+        description: str,
+        test_framework_exit_codes: list[Union[int, None]],
+    ) -> dict:
         """Get outcome from test execution.
 
         :param result: Result of execution.
@@ -124,7 +149,16 @@ def outcome(self, result, executed, description):
         :return: Outcome of test execution.
         :rtype: dict
         """
-        if executed:
+        test_framework_output = {
+            "test_framework_exit_codes": test_framework_exit_codes,
+        }
+        custom_verdict = self.verdict_matcher.evaluate(test_framework_output)
+        if custom_verdict is not None:
+            conclusion = custom_verdict["conclusion"]
+            verdict = custom_verdict["verdict"]
+            description = custom_verdict["description"]
+            self.logger.info("Verdict matches testrunner verdict rule: %s", custom_verdict)
+        elif executed:
             conclusion = "SUCCESSFUL"
             verdict = "PASSED" if result else "FAILED"
             self.logger.info(
@@ -205,12 +239,13 @@ def execute(self):  # pylint:disable=too-many-branches,disable=too-many-statemen
         result = True
         description = None
         executed = False
+        test_framework_exit_codes = []
         try:
             with Workspace(self.log_area) as workspace:
                 self.logger.info("Start IUT monitoring.")
                 self.iut_monitoring.start_monitoring()
                 self.logger.info("Starting test executor.")
-                result = self.run_tests(workspace)
+                result, test_framework_exit_codes = self.run_tests(workspace)
                 executed = True
                 self.logger.info("Stop IUT monitoring.")
                 self.iut_monitoring.stop_monitoring()
@@ -224,7 +259,7 @@ def execute(self):  # pylint:disable=too-many-branches,disable=too-many-statemen
                 self.logger.info("Stop IUT monitoring.")
                 self.iut_monitoring.stop_monitoring()
             self.logger.info("Figure out test outcome.")
-            outcome = self.outcome(result, executed, description)
+            outcome = self.outcome(result, executed, description, test_framework_exit_codes)
             pprint(outcome)
 
             self.logger.info("Send test suite finished event.")

src/etos_test_runner/lib/verdict.py

@@ -0,0 +1,87 @@
+# 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.
+"""Verdict module."""
+from typing import Union
+
+
+class CustomVerdictMatcher:
+    # pylint: disable=too-few-public-methods
+    """Match testframework output against user-defined verdict rules.
+
+    Example rule definition:
+
+    rules = [
+        {
+            "description": "Test collection error, no artifacts created",
+            "condition": {
+                "test_framework_exit_code": 4,
+            },
+            "conclusion": "FAILED",
+            "verdict": "FAILED",
+        }
+    ]
+
+    Condition keywords:
+    - test_framework_exit_code: allows set custom verdict if the given exit code is
+        found in the list exit codes produced by the test framework.
+    """
+
+    REQUIRED_RULE_KEYWORDS = {
+        "description",
+        "condition",
+        "conclusion",
+        "verdict",
+    }
+    SUPPORTED_CONDITION_KEYWORDS = {
+        "test_framework_exit_code",
+    }
+
+    def __init__(self, rules: list) -> None:
+        """Create new instance."""
+        self.rules = rules
+        for rule in self.rules:
+            if set(rule.keys()) != self.REQUIRED_RULE_KEYWORDS:
+                raise ValueError(
+                    f"Unsupported rule definition: {rule}. "
+                    f"Required keywords: {self.REQUIRED_RULE_KEYWORDS}"
+                )
+            for key in rule["condition"].keys():
+                if key not in self.SUPPORTED_CONDITION_KEYWORDS:
+                    raise ValueError(
+                        f"Unsupported condition keyword for test outcome rules: {key}! "
+                        f"Supported keywords: {self.SUPPORTED_CONDITION_KEYWORDS}."
+                    )
+
+    def _evaluate_rule(self, rule: dict, test_framework_output: dict) -> bool:
+        """Evaluate conditions within the given rule."""
+        for kw, expected_value in rule["condition"].items():
+            # If the condition has multiple expressions, they are implicitly
+            # joined using logical AND: i. e. all shall evaluate to True
+            # in order for the condition to be True.
+            # False is returned as soon as a false statement is encountered.
+            if kw == "test_framework_exit_code":
+                # If the exit code given by the condition is found in
+                # the list of produced exit codes, the rule will evaluate as True.
+                if expected_value not in test_framework_output.get("test_framework_exit_codes"):
+                    return False
+        return True
+
+    def evaluate(self, test_framework_output: dict) -> Union[dict, None]:
+        """Evaluate the list of given rules and return the first match."""
+        for rule in self.rules:
+            if self._evaluate_rule(rule, test_framework_output):
+                return rule
+        return None