[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/main.py

@@ -79,6 +79,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="Regex used to filter specific test scenario. If not set, all scenario are ran. Mainly useful to quickly debug a specifc scenario.",
+    )
+
     return parser.parse_args()
 
 
@@ -109,10 +115,15 @@ def sign(self, whole_config: USSQualifierConfiguration) -> None:
 
 
 def execute_test_run(
-    whole_config: USSQualifierConfiguration, description: TestDefinitionDescription
+    whole_config: USSQualifierConfiguration,
+    description: TestDefinitionDescription,
+    scenarios_filter: str | None,
 ):
     config = whole_config.v1.test_run
 
+    if not config:
+        raise Exception("Config not found")
+
     logger.info("Instantiating resources")
     stop_when_not_created = (
         "execution" in config
@@ -127,7 +138,9 @@ def execute_test_run(
     )
 
     logger.info("Instantiating top-level test suite action")
-    context = ExecutionContext(config.execution if "execution" in config else None)
+    context = ExecutionContext(
+        config.execution if "execution" in config else None, scenarios_filter
+    )
     action = TestSuiteAction(config.action, resources)
     logger.info("Running top-level test suite action")
     report = action.run(context)
@@ -172,6 +185,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)
 
@@ -228,7 +242,7 @@ def run_config(
         )
 
     logger.info("Executing test run")
-    report = execute_test_run(whole_config, description)
+    report = execute_test_run(whole_config, description, scenarios_filter)
 
     if runtime_metadata is not None:
         report.runtime_metadata = runtime_metadata
@@ -257,6 +271,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 +300,7 @@ def main() -> int:
             output_path,
             runtime_metadata,
             disallow_unredacted,
+            scenarios_filter,
         )
         if exit_code != os.EX_OK:
             return exit_code

monitoring/uss_qualifier/scenarios/interuss/unit_test.py

@@ -25,7 +25,7 @@ def run(self, context: ExecutionContext):
         self.end_test_scenario()
 
     def execute_unit_test(self):
-        context = ExecutionContext(None)
+        context = ExecutionContext(None, None)
         self.run(context)
         self.cleanup()
         return self

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)
@@ -412,12 +413,16 @@ class ExecutionContext:
     config: ExecutionConfiguration | None
     top_frame: ActionStackFrame | None
     current_frame: ActionStackFrame | None
+    scenarios_filter: str | None
 
-    def __init__(self, config: ExecutionConfiguration | None):
+    def __init__(
+        self, config: ExecutionConfiguration | None, scenarios_filter: str | None
+    ):
         self.config = config
         self.top_frame = None
         self.current_frame = None
         self.start_time = arrow.utcnow().datetime
+        self.scenarios_filter = scenarios_filter
 
     def sibling_queries(self) -> Iterator[Query]:
         if self.current_frame is None or self.current_frame.parent is None:
@@ -630,6 +635,17 @@ def evaluate_skip(self) -> SkippedActionReport | None:
                         declaration=self.current_frame.action.declaration,
                     )
 
+        if self.scenarios_filter and self.current_frame.action.test_scenario:
+            if not re.match(
+                self.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: