Simplify scenario building

.github/workflows/main.yml

@@ -2,10 +2,14 @@ name: Integration Test
 
 on:
   pull_request:
-    types: [ready_for_review]
+    types: [opened, synchronize, reopened]
     branches:
     - main
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   docker:
     timeout-minutes: 30
@@ -21,7 +25,7 @@ jobs:
         version: latest
 
     - name: Start Flight Blender and dependencies
-      run: docker compose --env-file .env.tests -f docker-compose.fb.yml up -d
+      run: docker compose --env-file .env.tests -f docker-compose.fb.yml up -d --wait
       working-directory: ./tests
 
     - name: Install uv
@@ -34,6 +38,7 @@ jobs:
       run: uv run openutm-verify --debug --config config/default.yaml
 
     - uses: actions/upload-artifact@v4
+      if: always()
       with:
         name: test-reports
         path: reports/

config/default.yaml

@@ -37,18 +37,19 @@ data_files:
 
 # List of test scenario IDs to execute
 scenarios:
-  # "F1_happy_path":
-  #   telemetry: "config/bern/telemetry_f1.json"
-  # "F2_contingent_path":
-  #   telemetry: "config/bern/telemetry_f2.json"
-  # "F3_non_conforming_path":
-  #   telemetry: "config/bern/telemetry_f3.json"
-  # "F5_non_conforming_path":
+  F1_happy_path:
+    telemetry: "config/bern/telemetry_f1.json"
+  F2_contingent_path:
+    telemetry: "config/bern/telemetry_f2.json"
+  F3_non_conforming_path:
+    telemetry: "config/bern/telemetry_f3.json"
+  # F5_non_conforming_path:
   #   telemetry: "config/bern/telemetry_f5.json"
-  # "opensky_live_data":
-  # "add_flight_declaration":
-  # "geo_fence_upload":
-  # "sdsp_track_heartbeat":
+  opensky_live_data:
+  add_flight_declaration:
+  geo_fence_upload:
+  # sdsp_track:
+  # sdsp_heartbeat:
   openutm_sim_air_traffic_data:
 
 # Reporting configuration

src/openutm_verification/cli/__init__.py

@@ -2,6 +2,7 @@
 Command Line Interface for OpenUTM Verification Tool.
 """
 
+import sys
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -41,12 +42,13 @@ def main():
     log_file = setup_logging(output_dir, base_filename, config.reporting.formats, args.debug)
 
     # Run verification scenarios
-    run_verification_scenarios(config, args.config)
+    failed = run_verification_scenarios(config, args.config)
 
     if log_file:
         from loguru import logger
 
         logger.info(f"Log file saved to: {log_file}")
+    sys.exit(failed)
 
 
 if __name__ == "__main__":

src/openutm_verification/core/clients/flight_blender/flight_blender_client.py

@@ -1,6 +1,7 @@
 import json
 import time
 import uuid
+from contextlib import contextmanager
 from dataclasses import asdict
 from typing import Any, Dict, List, Optional
 
@@ -10,8 +11,12 @@
 from openutm_verification.core.clients.flight_blender.base_client import (
     BaseBlenderAPIClient,
 )
+from openutm_verification.core.execution.config_models import DataFiles
 from openutm_verification.core.execution.scenario_runner import scenario_step
-from openutm_verification.core.reporting.reporting_models import Status, StepResult
+from openutm_verification.core.reporting.reporting_models import (
+    Status,
+    StepResult,
+)
 from openutm_verification.models import (
     FlightBlenderError,
     HeartbeatMessage,
@@ -75,6 +80,8 @@ def __init__(self, base_url: str, credentials: Dict[str, Any], request_timeout:
         self.latest_geo_fence_id: Optional[str] = None
         # Context: store the most recently created flight declaration id for teardown/steps
         self.latest_flight_declaration_id: Optional[str] = None
+        # Context: store the generated telemetry states for the current scenario
+        self.telemetry_states: Optional[List[Dict[str, Any]]] = None
         logger.debug(f"Initialized FlightBlenderClient with base_url={base_url}, request_timeout={request_timeout}")
 
     def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
@@ -89,11 +96,10 @@ def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
         return super().__exit__(exc_type, exc_val, exc_tb)
 
     @scenario_step("Upload Geo Fence")
-    def upload_geo_fence(self, operation_id: Optional[str] = None, filename: Optional[str] = None) -> Dict[str, Any]:
+    def upload_geo_fence(self, filename: Optional[str] = None) -> Dict[str, Any]:
         """Upload an Area-of-Interest (Geo Fence) to Flight Blender.
 
         Args:
-            operation_id: Not used for geo-fence upload (included for API consistency).
             filename: Path to the GeoJSON file containing the geo-fence definition.
 
         Returns:
@@ -121,12 +127,9 @@ def upload_geo_fence(self, operation_id: Optional[str] = None, filename: Optiona
         return body
 
     @scenario_step("Get Geo Fence")
-    def get_geo_fence(self, operation_id: Optional[str] = None) -> Dict[str, Any]:
+    def get_geo_fence(self) -> Dict[str, Any]:
         """Retrieve the details of the most recently uploaded geo-fence.
 
-        Args:
-            operation_id: Not used for geo-fence retrieval (included for API consistency).
-
         Returns:
             The JSON response from the API containing geo-fence details, or a dict
             indicating skip if no geo-fence ID is available.
@@ -237,13 +240,12 @@ def upload_flight_declaration(self, declaration: str | Any) -> Dict[str, Any]:
         return response_json
 
     @scenario_step("Update Operation State")
-    def update_operation_state(self, operation_id: str, new_state: OperationState, duration_seconds: int = 0) -> Dict[str, Any]:
+    def update_operation_state(self, new_state: OperationState, duration_seconds: int = 0) -> Dict[str, Any]:
         """Update the state of a flight operation.
 
         Posts the new state and optionally waits for the specified duration.
 
         Args:
-            operation_id: The ID of the operation to update.
             new_state: The new OperationState to set.
             duration_seconds: Optional seconds to sleep after update (default 0).
 
@@ -253,12 +255,12 @@ def update_operation_state(self, operation_id: str, new_state: OperationState, d
         Raises:
             FlightBlenderError: If the update request fails.
         """
-        endpoint = f"/flight_declaration_ops/flight_declaration_state/{operation_id}"
-        logger.debug(f"Updating operation {operation_id} to state {new_state.name}")
+        endpoint = f"/flight_declaration_ops/flight_declaration_state/{self.latest_flight_declaration_id}"
+        logger.debug(f"Updating operation {self.latest_flight_declaration_id} to state {new_state.name}")
         payload = {"state": new_state.value, "submitted_by": "hh@auth.com"}
 
         response = self.put(endpoint, json=payload)
-        logger.info(f"Operation state updated for {operation_id} to {new_state.name}")
+        logger.info(f"Operation state updated for {self.latest_flight_declaration_id} to {new_state.name}")
         if duration_seconds > 0:
             logger.debug(f"Sleeping for {duration_seconds} seconds after state update")
             time.sleep(duration_seconds)
@@ -281,11 +283,10 @@ def _load_telemetry_file(self, filename: str) -> List[Dict[str, Any]]:
             rid_json = json.loads(rid_json_file.read())
         return rid_json["current_states"]
 
-    def _submit_telemetry_states_impl(self, operation_id: str, states: List[Dict[str, Any]], duration_seconds: int = 0) -> Optional[Dict[str, Any]]:
+    def _submit_telemetry_states_impl(self, states: List[Dict[str, Any]], duration_seconds: int = 0) -> Optional[Dict[str, Any]]:
         """Internal implementation for submitting telemetry states.
 
         Args:
-            operation_id: The ID of the operation for telemetry submission.
             states: List of telemetry state dictionaries.
             duration_seconds: Optional maximum duration in seconds to submit telemetry (default 0 for unlimited).
 
@@ -296,9 +297,9 @@ def _submit_telemetry_states_impl(self, operation_id: str, states: List[Dict[str
             FlightBlenderError: If maximum waiting time is exceeded due to rate limits.
         """
         endpoint = "/flight_stream/set_telemetry"
-        logger.debug(f"Submitting telemetry for operation {operation_id}")
+        logger.debug(f"Submitting telemetry for operation {self.latest_flight_declaration_id}")
 
-        rid_operator_details = _create_rid_operator_details(operation_id)
+        rid_operator_details = _create_rid_operator_details(self.latest_flight_declaration_id)
 
         last_response = None
         maximum_waiting_time = 10.0
@@ -337,14 +338,13 @@ def _submit_telemetry_states_impl(self, operation_id: str, states: List[Dict[str
         return last_response
 
     @scenario_step("Submit Telemetry (from file)")
-    def submit_telemetry_from_file(self, operation_id: str, filename: str, duration_seconds: int = 0) -> Optional[Dict[str, Any]]:
+    def submit_telemetry_from_file(self, filename: str, duration_seconds: int = 0) -> Optional[Dict[str, Any]]:
         """Submit telemetry data for a flight operation.
 
         Loads telemetry states from file and submits them sequentially, with optional
         duration limiting and error handling for rate limits.
 
         Args:
-            operation_id: The ID of the operation for telemetry submission.
             filename: Path to the JSON file containing telemetry data.
             duration_seconds: Optional maximum duration in seconds to submit telemetry (default 0 for unlimited).
 
@@ -355,7 +355,7 @@ def submit_telemetry_from_file(self, operation_id: str, filename: str, duration_
             FlightBlenderError: If maximum waiting time is exceeded due to rate limits.
         """
         states = self._load_telemetry_file(filename)
-        return self._submit_telemetry_states_impl(operation_id, states, duration_seconds)
+        return self._submit_telemetry_states_impl(states, duration_seconds)
 
     @scenario_step("Wait X seconds")
     def wait_x_seconds(self, wait_time_seconds: int = 5) -> str:
@@ -366,15 +366,14 @@ def wait_x_seconds(self, wait_time_seconds: int = 5) -> str:
         return f"Waited for Flight Blender to process {wait_time_seconds} seconds."
 
     @scenario_step("Submit Telemetry")
-    def submit_telemetry(self, operation_id: str, states: List[Dict[str, Any]], duration_seconds: int = 0) -> Optional[Dict[str, Any]]:
+    def submit_telemetry(self, states: Optional[List[Dict[str, Any]]] = None, duration_seconds: int = 0) -> Optional[Dict[str, Any]]:
         """Submit telemetry data for a flight operation from in-memory states.
 
         Submits telemetry states sequentially from the provided list, with optional
         duration limiting and error handling for rate limits.
 
         Args:
-            operation_id: The ID of the operation for telemetry submission.
-            states: List of telemetry state dictionaries.
+            states: List of telemetry state dictionaries. If None, uses the generated telemetry states from context.
             duration_seconds: Optional maximum duration in seconds to submit telemetry (default 0 for unlimited).
 
         Returns:
@@ -383,12 +382,15 @@ def submit_telemetry(self, operation_id: str, states: List[Dict[str, Any]], dura
         Raises:
             FlightBlenderError: If maximum waiting time is exceeded due to rate limits.
         """
-        return self._submit_telemetry_states_impl(operation_id, states, duration_seconds)
+        telemetry_states = states or self.telemetry_states
+        if telemetry_states is None:
+            raise ValueError("Telemetry states are required and could not be resolved from context.")
+
+        return self._submit_telemetry_states_impl(telemetry_states, duration_seconds)
 
     @scenario_step("Check Operation State")
     def check_operation_state(
         self,
-        operation_id: str,
         expected_state: OperationState,
         duration_seconds: int = 0,
     ) -> str:
@@ -398,30 +400,27 @@ def check_operation_state(
         and returns a success status.
 
         Args:
-            operation_id: The ID of the operation to check.
             expected_state: The expected OperationState.
             duration_seconds: Seconds to wait for processing.
 
         Returns:
             A dictionary with the check result.
         """
-        logger.info(f"Checking operation state for {operation_id} (simulated)...")
+        logger.info(f"Checking operation state for {self.latest_flight_declaration_id} (simulated)...")
         logger.info(f"Waiting for {duration_seconds} seconds for Flight Blender to process state...")
         time.sleep(duration_seconds)
-        logger.info(f"Flight state check for {operation_id} completed (simulated).")
+        logger.info(f"Flight state check for {self.latest_flight_declaration_id} completed (simulated).")
         return f"Waited for Flight Blender to process {expected_state} state."
 
     @scenario_step("Check Operation State Connected")
     def check_operation_state_connected(
         self,
-        operation_id: str,
         expected_state: OperationState,
         duration_seconds: int = 0,
     ) -> Dict[str, Any]:
         """Check the operation state by polling the API until the expected state is reached.
 
         Args:
-            operation_id: The ID of the operation to check.
             expected_state: The expected OperationState.
             duration_seconds: Maximum seconds to poll for the state.
 
@@ -431,36 +430,36 @@ def check_operation_state_connected(
         Raises:
             FlightBlenderError: If the expected state is not reached within the timeout.
         """
-        endpoint = f"/flight_declaration_ops/flight_declaration/{operation_id}"
-        logger.info(f"Checking operation state for {operation_id}, expecting {expected_state.name}")
+        endpoint = f"/flight_declaration_ops/flight_declaration/{self.latest_flight_declaration_id}"
+        logger.info(f"Checking operation state for {self.latest_flight_declaration_id}, expecting {expected_state.name}")
         start_time = time.time()
 
         while time.time() - start_time < duration_seconds:
             response = self.get(endpoint)
             data = response.json()
             current_state_value = data.get("state")
-            logger.debug(f"Current state for {operation_id}: {current_state_value}")
+            logger.debug(f"Current state for {self.latest_flight_declaration_id}: {current_state_value}")
             if current_state_value == expected_state.value:
-                logger.info(f"Operation {operation_id} reached expected state {expected_state.name}")
+                logger.info(f"Operation {self.latest_flight_declaration_id} reached expected state {expected_state.name}")
                 return data
 
             time.sleep(1)
 
-        logger.error(f"Operation {operation_id} did not reach expected state {expected_state.name} within {duration_seconds} seconds")
-        raise FlightBlenderError(f"Operation {operation_id} did not reach expected state {expected_state.name} within {duration_seconds} seconds")
+        logger.error(
+            f"Operation {self.latest_flight_declaration_id} did not reach expected state {expected_state.name} within {duration_seconds} seconds"
+        )
+        raise FlightBlenderError(
+            f"Operation {self.latest_flight_declaration_id} did not reach expected state {expected_state.name} within {duration_seconds} seconds"
+        )
 
     @scenario_step("Delete Flight Declaration")
-    def delete_flight_declaration(self, operation_id: Optional[str] = None) -> Dict[str, Any]:
+    def delete_flight_declaration(self) -> Dict[str, Any]:
         """Delete a flight declaration by ID.
 
-        Args:
-            operation_id: Optional ID of the flight declaration to delete. If not provided,
-                uses the latest uploaded flight declaration ID.
-
         Returns:
             A dictionary with deletion status, including whether it was successful.
         """
-        op_id = operation_id or self.latest_flight_declaration_id
+        op_id = self.latest_flight_declaration_id
         if not op_id:
             logger.warning("No flight declaration ID available for deletion")
             return {
@@ -600,6 +599,7 @@ def initialize_heartbeat_websocket_connection(self, session_id: str) -> Any:
         ws = self.create_websocket_connection(endpoint=endpoint)
         return ws
 
+    @scenario_step("Verify SDSP Track")
     def initialize_verify_sdsp_track(
         self,
         expected_heartbeat_interval_seconds: int,
@@ -672,6 +672,7 @@ def initialize_verify_sdsp_track(
                 duration=duration,
             )
 
+    @scenario_step("Verify SDSP Heartbeat")
     def initialize_verify_sdsp_heartbeat(
         self,
         expected_heartbeat_interval_seconds: int,
@@ -746,3 +747,31 @@ def initialize_verify_sdsp_heartbeat(
 
     def close_heartbeat_websocket_connection(self, ws_connection: Any) -> None:
         ws_connection.close()
+
+    @scenario_step("Setup Flight Declaration")
+    def setup_flight_declaration(self, flight_declaration_path: str, telemetry_path: str) -> None:
+        """Generates data and uploads flight declaration."""
+        from openutm_verification.scenarios.common import (
+            generate_flight_declaration,
+            generate_telemetry,
+        )
+
+        flight_declaration = generate_flight_declaration(flight_declaration_path)
+        telemetry_states = generate_telemetry(telemetry_path)
+
+        self.telemetry_states = telemetry_states
+
+        upload_result = self.upload_flight_declaration(flight_declaration)
+
+        if upload_result.status == Status.FAIL:
+            logger.error(f"Flight declaration upload failed: {upload_result}")
+            raise FlightBlenderError("Failed to upload flight declaration during setup_flight_declaration")
+
+    @contextmanager
+    def flight_declaration(self, data_files: DataFiles):
+        """Context manager to setup and teardown a flight operation based on scenario config."""
+        self.setup_flight_declaration(data_files.flight_declaration, data_files.telemetry)
+        try:
+            yield
+        finally:
+            self.delete_flight_declaration()