Refactor scenario configuration to support suites and enhance CLI

VerificationFlowDiagram.md

@@ -21,7 +21,7 @@ graph TD
     end
 
     subgraph "Core Execution"
-        H --> I{"For each scenario_id in config.scenarios"};
+        H --> I{"For each suite/scenario in config.suites"};
         I -- Yes --> J["Registry Lookup: SCENARIO_REGISTRY[scenario_id]"];
         J --> K{"OpenSky Required?"};
         K -- Yes --> L["OpenSky Settings: create_opensky_settings()"];

config/default.yaml

@@ -35,22 +35,25 @@ data_files:
   flight_declaration: "config/bern/flight_declaration.json" # Path to flight declarations JSON file
   # geo_fence: "config/geo_fences.json" # Path to geo-fences
 
-# List of test scenario IDs to execute
-scenarios:
-  F1_happy_path:
-    trajectory: "config/bern/trajectory_f1.json"
-  F2_contingent_path:
-    trajectory: "config/bern/trajectory_f2.json"
-  F3_non_conforming_path:
-    trajectory: "config/bern/trajectory_f3.json"
-#  F5_non_conforming_path:
-#    trajectory: "config/bern/trajectory_f5.json"
-  opensky_live_data:
-  add_flight_declaration:
-  geo_fence_upload:
-  # sdsp_track:
-  # sdsp_heartbeat:
-  openutm_sim_air_traffic_data:
+suites:
+  basic_conformance:
+    scenarios:
+    - name: F1_happy_path
+      trajectory: "config/bern/trajectory_f1.json"
+    - name: F2_contingent_path
+      trajectory: "config/bern/trajectory_f2.json"
+  extra:
+    scenarios:
+    - name: F3_non_conforming_path
+      trajectory: "config/bern/trajectory_f3.json"
+#     - name: F5_non_conforming_path
+#       trajectory: "config/bern/trajectory_f5.json"
+    - name: opensky_live_data
+    - name: add_flight_declaration
+    - name: geo_fence_upload
+    - name: openutm_sim_air_traffic_data
+#     - name: sdsp_track
+#     - name: sdsp_heartbeat
 
 # Reporting configuration
 reporting:

src/openutm_verification/cli/__init__.py

@@ -31,6 +31,13 @@ def main():
     project_root = config_path.parent.parent  # config/ -> project root
     config.resolve_paths(project_root)
 
+    if args.suite:
+        for suite_name in args.suite:
+            if suite_name not in config.suites:
+                print(f"Error: Suite '{suite_name}' not found in configuration.")
+                sys.exit(1)
+        config.target_suites = args.suite
+
     ConfigProxy.initialize(config)
 
     # Setup logging

src/openutm_verification/cli/parser.py

@@ -20,4 +20,11 @@ def create_parser() -> argparse.ArgumentParser:
         action="store_true",
         help="Enable debug logging on the console.",
     )
+    parser.add_argument(
+        "-s",
+        "--suite",
+        type=str,
+        action="append",
+        help="Name of the test suite to run (overrides individual scenarios). Can be specified multiple times.",
+    )
     return parser

src/openutm_verification/core/execution/config_models.py

@@ -99,6 +99,22 @@ def resolve_and_validate_path(path_str: str, field_name: str) -> str:
             self.geo_fence = resolve_and_validate_path(self.geo_fence, "Geo-fence")
 
 
+class SuiteScenario(DataFiles):
+    """A scenario within a suite, allowing overrides."""
+
+    name: str
+
+
+class SuiteConfig(StrictBaseModel):
+    """Configuration for a test suite."""
+
+    scenarios: List[SuiteScenario]
+
+    def resolve_paths(self, base_path: Path) -> None:
+        for scenario in self.scenarios:
+            scenario.resolve_paths(base_path)
+
+
 class AppConfig(StrictBaseModel):
     """Root model for the application configuration."""
 
@@ -108,23 +124,27 @@ class AppConfig(StrictBaseModel):
     opensky: OpenSkyConfig
     air_traffic_simulator_settings: AirTrafficSimulatorSettings
     data_files: DataFiles
-    scenarios: Dict[str, DataFiles | None] = Field(default_factory=dict)
+    suites: Dict[str, SuiteConfig] = Field(default_factory=dict)
     reporting: ReportingConfig
 
+    # Runtime only
+    target_suites: List[str] = Field(default_factory=list)
+
     def resolve_paths(self, config_file_path: Path) -> None:
         """Resolve all relative paths in the configuration to absolute paths."""
         base_path = config_file_path.parent
         self.data_files.resolve_paths(base_path)
-        for scenario_data in self.scenarios.values():
-            if scenario_data:
-                scenario_data.resolve_paths(base_path)
+        for suite in self.suites.values():
+            suite.resolve_paths(base_path)
 
 
 ScenarioId = Annotated[str, "The unique identifier for a scenario"]
 
 
 class RunContext(TypedDict):
     scenario_id: str
+    suite_scenario: Optional[SuiteScenario]
+    suite_name: Optional[str]
 
 
 class ConfigMeta(type):

src/openutm_verification/core/execution/dependencies.py

@@ -1,4 +1,4 @@
-from typing import Callable, Generator, Iterable, TypeVar
+from typing import Callable, Generator, Iterable, TypeVar, cast
 
 from loguru import logger
 
@@ -22,14 +22,42 @@ def scenarios() -> Iterable[tuple[str, Callable[..., ScenarioResult]]]:
     Returns:
         An iterable of tuples containing (scenario_id, scenario_function).
     """
-    scenarios_to_run = get_settings().scenarios
-    logger.info(f"Found {len(scenarios_to_run)} scenarios to run.")
-    for scenario_id in scenarios_to_run:
-        CONTEXT.set({"scenario_id": scenario_id})
+    config = get_settings()
+
+    scenarios_to_iterate = []
+
+    # Determine which suites to run
+    target_suite_names = config.target_suites if config.target_suites else config.suites.keys()
+
+    if not target_suite_names:
+        logger.warning("No suites defined in configuration.")
+
+    for suite_name in target_suite_names:
+        if suite_name not in config.suites:
+            logger.error(f"Target suite '{suite_name}' not found in configuration.")
+            continue
+
+        suite = config.suites[suite_name]
+        logger.info(f"Adding suite: {suite_name} with {len(suite.scenarios)} scenarios.")
+        for suite_scenario in suite.scenarios:
+            scenarios_to_iterate.append((suite_name, suite_scenario))
+
+    for suite_name, item in scenarios_to_iterate:
+        scenario_id = item.name
+        suite_scenario = item
+
         if scenario_id in SCENARIO_REGISTRY:
             logger.info("=" * 100)
             logger.info(f"Running scenario: {scenario_id}")
+
             scenario_func = SCENARIO_REGISTRY[scenario_id]
+
+            CONTEXT.set({
+                "scenario_id": scenario_id,
+                "suite_scenario": suite_scenario,
+                "suite_name": suite_name,
+                "docs": None
+            })
             yield scenario_id, scenario_func
         else:
             logger.warning(f"Scenario {scenario_id} not found in registry.")
@@ -54,11 +82,25 @@ def data_files(scenario_id: ScenarioId) -> Generator[DataFiles, None, None]:
         An instance of DataFiles.
     """
     config = get_settings()
-    scenario_config = config.scenarios.get(scenario_id) or config.data_files
+
+    # Check for suite override
+    suite_scenario = CONTEXT.get().get("suite_scenario")
+
+    if suite_scenario:
+        # Merge suite overrides with base config
+        trajectory = suite_scenario.trajectory or config.data_files.trajectory
+        flight_declaration = suite_scenario.flight_declaration or config.data_files.flight_declaration
+        geo_fence = suite_scenario.geo_fence or config.data_files.geo_fence
+    else:
+        # Use base config
+        trajectory = config.data_files.trajectory
+        flight_declaration = config.data_files.flight_declaration
+        geo_fence = config.data_files.geo_fence
+
     data = DataFiles(
-        trajectory=scenario_config.trajectory or config.data_files.trajectory,
-        flight_declaration=scenario_config.flight_declaration or config.data_files.flight_declaration,
-        geo_fence=scenario_config.geo_fence or config.data_files.geo_fence,
+        trajectory=trajectory,
+        flight_declaration=flight_declaration,
+        geo_fence=geo_fence,
     )
     yield data
 
@@ -70,7 +112,7 @@ def app_config() -> Generator[AppConfig, None, None]:
     Returns:
         An instance of AppConfig.
     """
-    yield get_settings()
+    yield cast(AppConfig, get_settings())
 
 
 @dependency(FlightBlenderClient)

src/openutm_verification/core/execution/dependency_resolution.py

@@ -7,12 +7,22 @@
 
 T = TypeVar("T")
 
-DEPENDENCIES: dict[Type, Callable[..., ContextManager[object]]] = {}
-CONTEXT: ContextVar[RunContext] = ContextVar("context", default={"scenario_id": ""})
-
-
-def dependency(type: Type) -> Callable:
-    def wrapper(func: Callable[..., object]) -> Callable[..., object]:
+DEPENDENCIES: dict[object, Callable[..., ContextManager[object]]] = {}
+CONTEXT: ContextVar[RunContext] = ContextVar(
+    "context",
+    default=cast(
+        RunContext,
+        {
+            "scenario_id": "",
+            "suite_scenario": None,
+            "suite_name": None,
+        },
+    ),
+)
+
+
+def dependency(type: object) -> Callable:
+    def wrapper(func: Callable[..., Generator]) -> Callable[..., Generator]:
         DEPENDENCIES[type] = contextmanager(func)
         return func
 
@@ -37,9 +47,9 @@ class DependencyResolver:
 
     def __init__(self, stack: ExitStack):
         self.stack = stack
-        self._cache: dict[Type, object] = {}
+        self._cache: dict[object, object] = {}
 
-    def resolve(self, type_: Type) -> object:
+    def resolve(self, type_: object) -> object:
         """Resolve a dependency of a specific type."""
         if type_ in self._cache:
             return self._cache[type_]
@@ -63,7 +73,7 @@ def resolve(self, type_: Type) -> object:
 
 
 @contextmanager
-def provide(*types: Type[T]) -> Generator[tuple[T, ...], None, None]:
+def provide(*types: object) -> Generator[tuple[object, ...], None, None]:
     """Context manager to provide dependencies for the given types.
 
     This function recursively resolves dependencies, meaning if a dependency
@@ -77,5 +87,5 @@ def provide(*types: Type[T]) -> Generator[tuple[T, ...], None, None]:
     """
     with ExitStack() as stack:
         resolver = DependencyResolver(stack)
-        instances = [cast(T, resolver.resolve(t)) for t in types]
+        instances = [resolver.resolve(t) for t in types]
         yield tuple(instances)

src/openutm_verification/core/execution/execution.py

@@ -19,7 +19,7 @@
 )
 from openutm_verification.core.execution.config_models import AppConfig
 from openutm_verification.core.execution.dependencies import scenarios
-from openutm_verification.core.execution.dependency_resolution import call_with_dependencies
+from openutm_verification.core.execution.dependency_resolution import CONTEXT, call_with_dependencies
 from openutm_verification.core.reporting.reporting import generate_reports
 from openutm_verification.core.reporting.reporting_models import (
     ReportData,
@@ -80,6 +80,11 @@ def run_verification_scenarios(config: AppConfig, config_path: Path):
                 steps=[],
                 error_message=str(e),
             )
+
+        # Enrich result with context data
+        context_data = CONTEXT.get()
+        result.suite_name = context_data.get("suite_name")
+
         scenario_results.append(result)
         logger.info(f"Scenario {scenario_id} finished with status: {result.status}")
 

src/openutm_verification/core/reporting/reporting_models.py

@@ -34,6 +34,7 @@ class ScenarioResult(BaseModel):
     """Data model for the result of a single scenario."""
 
     name: str
+    suite_name: Optional[str] = None
     status: Status
     duration_seconds: float
     steps: List[StepResult[Any]]

src/openutm_verification/core/templates/report_template.html

@@ -64,7 +64,12 @@ <h2>Scenario Results ({{ report_data.results|length }} executed)</h2>
     <div class="scenario" id="scenario-{{ loop.index }}">
         <details>
             <summary class="scenario-summary">
-                <span>{{ result.name }}</span>
+                <span>
+                    {% if result.suite_name %}
+                        <strong>[{{ result.suite_name }}]</strong>
+                    {% endif %}
+                    {{ result.name }}
+                </span>
                 <span class="status status-{{ result.status }}">{{ result.status }}</span>
             </summary>
             <div class="scenario-details">