twister: quarantine: Added simulations field and regex to filter fields

Regex are allowed on every field, not only scenarios.
Signed-off-by: Grzegorz Chwierut <[email protected]>

Author: gchwier

scripts/pylib/twister/twisterlib/quarantine.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import logging
+import re
 
 from pathlib import Path
 from yaml import safe_load
@@ -25,9 +26,10 @@ def __init__(self, quarantine_list=[]) -> None:
         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)
+    def get_matched_quarantine(self, testname, platform, architecture, simulation):
+        qelem = self.quarantine.get_matched_quarantine(testname, platform, architecture, simulation)
         if qelem:
+            logger.debug('%s quarantined with reason: %s' % (testname, qelem.comment))
             return qelem.comment
         return None
 
@@ -37,18 +39,26 @@ class QuarantineElement:
     scenarios: list[str] = field(default_factory=list)
     platforms: list[str] = field(default_factory=list)
     architectures: list[str] = field(default_factory=list)
+    simulations: list[str] = field(default_factory=list)
     comment: str = 'under quarantine'
 
     def __post_init__(self):
+        # If there is no entry in filters then take all possible values.
+        # To keep backward compatibility, 'all' keyword might be still used.
         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")
+        if 'all' in self.simulations:
+            self.simulations = []
+        # However, at least one of the filters ('scenarios', platforms' ...)
+        # must be given (there is no sense to put all possible configuration
+        # into quarantine)
+        if not any([self.scenarios, self.platforms, self.architectures, self.simulations]):
+            raise QuarantineException("At least one of filters ('scenarios', 'platforms' ...) "
+                                      "must be specified")
 
 
 @dataclass
@@ -58,48 +68,56 @@ class QuarantineData:
     def __post_init__(self):
         qelements = []
         for qelem in self.qlist:
-            qelements.append(QuarantineElement(**qelem))
+            if isinstance(qelem, QuarantineElement):
+                qelements.append(qelem)
+            else:
+                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)
+            qlist_raw_data: list[dict] = safe_load(yaml_fd)
         try:
-            return cls(qlist)
+            return cls(qlist_raw_data)
 
         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]:
+    def extend(self, qdata: QuarantineData) -> None:
         self.qlist.extend(qdata.qlist)
 
-    def get_matched_quarantine(self, scenario: str, platform: str,
-                               architecture: str) -> QuarantineElement | None:
+    def get_matched_quarantine(self,
+                               scenario: str,
+                               platform: str,
+                               architecture: str,
+                               simulation: 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 (qelem.scenarios
+                    and (matched := _is_element_matched(scenario, qelem.scenarios)) is False):
+                continue
+            if (qelem.platforms
+                    and (matched := _is_element_matched(platform, qelem.platforms)) is False):
+                continue
+            if (qelem.architectures
+                    and (matched := _is_element_matched(architecture, qelem.architectures)) is False):
+                continue
+            if (qelem.simulations
+                    and (matched := _is_element_matched(simulation, qelem.simulations)) is False):
+                continue
+
             if matched:
                 return qelem
-
         return None
+
+
+def _is_element_matched(element: str, list_of_elements: list) -> bool:
+    """Return True if given element is matching to any of elements from the list"""
+    for pattern in list_of_elements:
+        if re.fullmatch(pattern, element):
+            return True
+    return False

scripts/pylib/twister/twisterlib/testplan.py

@@ -747,7 +747,7 @@ def apply_filters(self, **kwargs):
                 # handle quarantined tests
                 if self.quarantine:
                     matched_quarantine = self.quarantine.get_matched_quarantine(
-                        instance.testsuite.id, plat.name, plat.arch
+                        instance.testsuite.id, plat.name, plat.arch, plat.simulation
                     )
                     if matched_quarantine and not self.options.quarantine_verify:
                         instance.add_filter(matched_quarantine, Filters.QUARENTINE)

scripts/schemas/twister/quarantine-schema.yaml

@@ -31,7 +31,13 @@ sequence:
         type: seq
         sequence:
           - type: str
-          - unique: true
+          - unique: True
+      "simulations":
+        required: false
+        type: seq
+        sequence:
+          - type: str
+          - unique: True
       "comment":
         type: str
         required: false