twister: Refactor and extend quarantine implementation in twister

Implementation ported from TwisterV2.
- quarantine handled by separate module
- multiple yaml allowed from args: --quarantine-list
- scenarios, platforms, architectures keywords in
quarantine yaml are optional, if not given - means take it all

Signed-off-by: Grzegorz Chwierut <[email protected]>

Author: gchwier

scripts/pylib/twister/twisterlib/environment.py

@@ -475,6 +475,7 @@ def add_parse_arguments(parser = None):
 
     parser.add_argument(
         "--quarantine-list",
+        action="append",
         metavar="FILENAME",
         help="Load list of test scenarios under quarantine. The entries in "
              "the file need to correspond to the test scenarios names as in "

scripts/pylib/twister/twisterlib/quarantine.py

@@ -0,0 +1,105 @@
+# Copyright (c) 2022 Nordic Semiconductor ASA
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import logging
+
+from pathlib import Path
+from yaml import safe_load
+from dataclasses import dataclass, field
+
+
+logger = logging.getLogger(__name__)
+
+
+class QuarantineException(Exception):
+    pass
+
+
+class Quarantine:
+    """Handle tests under quarantine."""
+
+    def __init__(self, quarantine_list=[]) -> None:
+        self.quarantine = QuarantineData()
+        for quarantine_file in quarantine_list:
+            self.quarantine.extend(QuarantineData.load_data_from_yaml(quarantine_file))
+
+    def get_matched_quarantine(self, testname, platform, architecture):
+        qelem = self.quarantine.get_matched_quarantine(testname, platform, architecture)
+        if qelem:
+            return qelem.comment
+        return None
+
+
+@dataclass
+class QuarantineElement:
+    scenarios: list[str] = field(default_factory=list)
+    platforms: list[str] = field(default_factory=list)
+    architectures: list[str] = field(default_factory=list)
+    comment: str = 'under quarantine'
+
+    def __post_init__(self):
+        if 'all' in self.scenarios:
+            self.scenarios = []
+        if 'all' in self.platforms:
+            self.platforms = []
+        if 'all' in self.architectures:
+            self.architectures = []
+        if not any([self.scenarios, self.platforms, self.architectures]):
+            raise QuarantineException("At least one of filters ('scenarios', 'platforms', "
+                                      "'architectures') must be specified")
+
+
+@dataclass
+class QuarantineData:
+    qlist: list[QuarantineElement] = field(default_factory=list)
+
+    def __post_init__(self):
+        qelements = []
+        for qelem in self.qlist:
+            qelements.append(QuarantineElement(**qelem))
+        self.qlist = qelements
+
+    @classmethod
+    def load_data_from_yaml(cls, filename: str | Path) -> QuarantineData:
+        """Load quarantine from yaml file."""
+        with open(filename, 'r', encoding='UTF-8') as yaml_fd:
+            qlist: list(dict) = safe_load(yaml_fd)
+        try:
+            return cls(qlist)
+
+        except Exception as e:
+            logger.error(f'When loading {filename} received error: {e}')
+            raise QuarantineException('Cannot load Quarantine data') from e
+
+    def extend(self, qdata: QuarantineData) -> list[QuarantineElement]:
+        self.qlist.extend(qdata.qlist)
+
+    def get_matched_quarantine(self, scenario: str, platform: str,
+                               architecture: str) -> QuarantineElement | None:
+        """Return quarantine element if test is matched to quarantine rules"""
+        for qelem in self.qlist:
+            matched: bool = False
+            if qelem.scenarios:
+                if scenario in qelem.scenarios:
+                    matched = True
+                else:
+                    matched = False
+                    continue
+            if qelem.platforms:
+                if platform in qelem.platforms:
+                    matched = True
+                else:
+                    matched = False
+                    continue
+            if qelem.architectures:
+                if architecture in qelem.architectures:
+                    matched = True
+                else:
+                    matched = False
+                    continue
+            if matched:
+                return qelem
+
+        return None

scripts/pylib/twister/twisterlib/testplan.py

@@ -29,6 +29,7 @@
 from twisterlib.platform import Platform
 from twisterlib.config_parser import TwisterConfigParser
 from twisterlib.testinstance import TestInstance
+from twisterlib.quarantine import Quarantine
 
 
 from zephyr_module import parse_modules
@@ -79,7 +80,7 @@ def __init__(self, env=None):
 
         # Keep track of which test cases we've filtered out and why
         self.testsuites = {}
-        self.quarantine = {}
+        self.quarantine = None
         self.platforms = []
         self.platform_names = []
         self.selected_platforms = []
@@ -128,15 +129,12 @@ def discover(self):
 
         # handle quarantine
         ql = self.options.quarantine_list
-        if ql:
-            self.load_quarantine(ql)
-
         qv = self.options.quarantine_verify
-        if qv:
-            if not ql:
-                logger.error("No quarantine list given to be verified")
-                raise TwisterRuntimeError("No quarantine list given to be verified")
-
+        if qv and not ql:
+            logger.error("No quarantine list given to be verified")
+            raise TwisterRuntimeError("No quarantine list given to be verified")
+        if ql:
+            self.quarantine = Quarantine(ql)
 
     def load(self):
 
@@ -463,35 +461,6 @@ def get_platform(self, name):
                 break
         return selected_platform
 
-    def load_quarantine(self, file):
-        """
-        Loads quarantine list from the given yaml file. Creates a dictionary
-        of all tests configurations (platform + scenario: comment) that shall be
-        skipped due to quarantine
-        """
-
-        # Load yaml into quarantine_yaml
-        quarantine_yaml = scl.yaml_load_verify(file, self.quarantine_schema)
-
-        # Create quarantine_list with a product of the listed
-        # platforms and scenarios for each entry in quarantine yaml
-        quarantine_list = []
-        for quar_dict in quarantine_yaml:
-            if quar_dict['platforms'][0] == "all":
-                plat = self.platform_names
-            else:
-                plat = quar_dict['platforms']
-                self.verify_platforms_existence(plat, "quarantine-list")
-            comment = quar_dict.get('comment', "NA")
-            quarantine_list.append([{".".join([p, s]): comment}
-                                   for p in plat for s in quar_dict['scenarios']])
-
-        # Flatten the quarantine_list
-        quarantine_list = [it for sublist in quarantine_list for it in sublist]
-        # Change quarantine_list into a dictionary
-        for d in quarantine_list:
-            self.quarantine.update(d)
-
     def load_from_file(self, file, filter_platform=[]):
         with open(file, "r") as json_test_plan:
             jtp = json.load(json_test_plan)
@@ -775,14 +744,15 @@ def apply_filters(self, **kwargs):
                     else:
                         instance.add_filter(f"Excluded platform missing key fields demanded by test {key_fields}", Filters.PLATFORM)
 
-                test_configuration = ".".join([instance.platform.name,
-                                               instance.testsuite.id])
-                # skip quarantined tests
-                if test_configuration in self.quarantine and not self.options.quarantine_verify:
-                    instance.add_filter(f"Quarantine: {self.quarantine[test_configuration]}", Filters.QUARENTINE)
-                # run only quarantined test to verify their statuses (skip everything else)
-                if self.options.quarantine_verify and test_configuration not in self.quarantine:
-                    instance.add_filter("Not under quarantine", Filters.QUARENTINE)
+                # handle quarantined tests
+                if self.quarantine:
+                    matched_quarantine = self.quarantine.get_matched_quarantine(
+                        instance.testsuite.id, plat.name, plat.arch
+                    )
+                    if matched_quarantine and not self.options.quarantine_verify:
+                        instance.add_filter(matched_quarantine, Filters.QUARENTINE)
+                    if not matched_quarantine and self.options.quarantine_verify:
+                        instance.add_filter("Not under quarantine", Filters.QUARENTINE)
 
                 # if nothing stopped us until now, it means this configuration
                 # needs to be added.

scripts/schemas/twister/quarantine-schema.yaml

@@ -16,12 +16,18 @@ sequence:
     mapping:
       "scenarios":
         type: seq
-        required: true
+        required: false
         sequence:
           - type: str
           - unique: true
       "platforms":
-        required: true
+        required: false
+        type: seq
+        sequence:
+          - type: str
+          - unique: True
+      "architectures":
+        required: false
         type: seq
         sequence:
           - type: str