[uss_qualifier] Add filter option when running tests

.basedpyright/baseline.json

@@ -5191,46 +5191,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportOperatorIssue",
-                "range": {
-                    "startColumn": 8,
-                    "endColumn": 29,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 19,
-                    "endColumn": 28,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 55,
-                    "endColumn": 64,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 19,
-                    "endColumn": 28,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 15,
-                    "endColumn": 24,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportArgumentType",
                 "range": {
@@ -5239,30 +5199,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 38,
-                    "endColumn": 47,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportOperatorIssue",
-                "range": {
-                    "startColumn": 51,
-                    "endColumn": 72,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 36,
-                    "endColumn": 42,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportOptionalSubscript",
                 "range": {

monitoring/uss_qualifier/configurations/configuration.py

@@ -132,6 +132,9 @@ class ExecutionConfiguration(ImplicitDict):
     stop_when_resource_not_created: bool | None = False
     """If true, stop test execution if one of the resources cannot be created.  Otherwise, resources that cannot be created due to missing prerequisites are simply treated as omitted."""
 
+    scenarios_filter: str | None
+    """Filter test scenarios by class name using a regex. When empty, all scenarios are executed. Useful for targeted debugging. Overridden by --filter"""
+
 
 class TestConfiguration(ImplicitDict):
     action: TestSuiteActionDeclaration

monitoring/uss_qualifier/main.py

@@ -12,6 +12,8 @@
 from monitoring.monitorlib.dicts import get_element_or_default, remove_elements
 from monitoring.monitorlib.versioning import get_code_version, get_commit_hash
 from monitoring.uss_qualifier.configurations.configuration import (
+    ExecutionConfiguration,
+    TestConfiguration,
     USSQualifierConfiguration,
     USSQualifierConfigurationV1,
 )
@@ -79,6 +81,12 @@ def parseArgs() -> argparse.Namespace:
         help="When true, do not run a test configuration which would produce unredacted sensitive information in its artifacts",
     )
 
+    parser.add_argument(
+        "--filter",
+        default=None,
+        help="Filter test scenarios by class name using a regex. When empty, all scenarios are executed. Useful for targeted debugging.",
+    )
+
     return parser.parse_args()
 
 
@@ -101,18 +109,20 @@ def sign(self, whole_config: USSQualifierConfiguration) -> None:
             baseline = whole_config
             environment = []
         self.baseline_signature = compute_baseline_signature(
-            self.codebase_version,
-            self.commit_hash,
-            compute_signature(baseline),
+            self.codebase_version, self.commit_hash, compute_signature(baseline)
         )
         self.environment_signature = compute_signature(environment)
 
 
 def execute_test_run(
-    whole_config: USSQualifierConfiguration, description: TestDefinitionDescription
+    whole_config: USSQualifierConfiguration,
+    description: TestDefinitionDescription,
 ):
     config = whole_config.v1.test_run
 
+    if not config:
+        raise ValueError("v1.test_run not defined in configuration")
+
     logger.info("Instantiating resources")
     stop_when_not_created = (
         "execution" in config
@@ -172,6 +182,7 @@ def run_config(
     output_path: str | None,
     runtime_metadata: dict | None,
     disallow_unredacted: bool,
+    scenarios_filter: str | None,
 ):
     config_src = load_dict_with_references(config_name)
 
@@ -187,6 +198,21 @@ def run_config(
 
     whole_config = ImplicitDict.parse(config_src, USSQualifierConfiguration)
 
+    if scenarios_filter:
+        # We set the scenario filter in the test run execution's object
+        # As parameters are optional, we ensure they do exist first
+
+        if "v1" not in whole_config or not whole_config.v1:
+            whole_config.v1 = USSQualifierConfigurationV1()
+        if "test_run" not in whole_config.v1 or not whole_config.v1.test_run:
+            whole_config.v1.test_run = TestConfiguration()
+        if (
+            "execution" not in whole_config.v1.test_run
+            or not whole_config.v1.test_run.execution
+        ):
+            whole_config.v1.test_run.execution = ExecutionConfiguration()
+        whole_config.v1.test_run.execution.scenarios_filter = scenarios_filter
+
     if config_output:
         logger.info("Writing flattened configuration to {}", config_output)
         if config_output.lower().endswith(".json"):
@@ -257,6 +283,7 @@ def main() -> int:
         raise ValueError("--runtime-metadata must specify a JSON dictionary")
 
     disallow_unredacted = args.disallow_unredacted
+    scenarios_filter = args.filter
 
     config_names = str(args.config).split(",")
 
@@ -285,6 +312,7 @@ def main() -> int:
             output_path,
             runtime_metadata,
             disallow_unredacted,
+            scenarios_filter,
         )
         if exit_code != os.EX_OK:
             return exit_code

monitoring/uss_qualifier/suites/suite.py

@@ -158,6 +158,7 @@ def _run_test_scenario(self, context: ExecutionContext) -> TestScenarioReport:
         scenario.time_context[TimeDuringTest.StartOfScenario] = Time(
             arrow.utcnow().datetime
         )
+
         try:
             try:
                 scenario.run(context)
@@ -630,6 +631,21 @@ def evaluate_skip(self) -> SkippedActionReport | None:
                         declaration=self.current_frame.action.declaration,
                     )
 
+        if (
+            "scenarios_filter" in self.config
+            and self.config.scenarios_filter
+            and self.current_frame.action.test_scenario
+        ):
+            if not re.match(
+                self.config.scenarios_filter,
+                self.current_frame.action.test_scenario.__class__.__name__,
+            ):
+                return SkippedActionReport(
+                    timestamp=StringBasedDateTime(arrow.utcnow()),
+                    reason="Filtered scenario",
+                    declaration=self.current_frame.action.declaration,
+                )
+
         return None
 
     def begin_action(self, action: TestSuiteAction) -> None:

schemas/monitoring/uss_qualifier/configurations/configuration/ExecutionConfiguration.json

@@ -17,6 +17,13 @@
         "null"
       ]
     },
+    "scenarios_filter": {
+      "description": "Filter test scenarios by class name using a regex. When empty, all scenarios are executed. Useful for targeted debugging. Overridden by --filter",
+      "type": [
+        "string",
+        "null"
+      ]
+    },
     "skip_action_when": {
       "description": "If specified, do not execute test actions if they are selected by ANY of these conditions.",
       "items": {