[uss_qualifier/scenarios/utm] Fix op intents cleanup by determining intents extents in run rather than __init__

Author: mickmis

.basedpyright/baseline.json

@@ -19255,14 +19255,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 35,
-                    "endColumn": 53,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportArgumentType",
                 "range": {

monitoring/uss_qualifier/resources/flight_planning/flight_intent_validation.py

@@ -48,7 +48,9 @@ def validate_flight_intent_templates(
     expected_intents: list[ExpectedFlightIntent],
 ) -> Volume4D:
     """
-    Returns: the bounding extents of the flight intent templates
+    Returns: the bounding extents of the flight intent templates.
+    When this return value is used to determine the extents in which to perform cleanup,
+    be sure to invoke it within the `run()` method of the scenario and not in the `__init__()` constructor.
     """
     extents = Volume4DCollection([])
 

monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_equal_priority_not_permitted/conflict_equal_priority_not_permitted.py

@@ -21,6 +21,9 @@
 from monitoring.uss_qualifier.resources.astm.f3548.v21 import DSSInstanceResource
 from monitoring.uss_qualifier.resources.astm.f3548.v21.dss import DSSInstance
 from monitoring.uss_qualifier.resources.flight_planning import FlightIntentsResource
+from monitoring.uss_qualifier.resources.flight_planning.flight_intent import (
+    FlightIntentID,
+)
 from monitoring.uss_qualifier.resources.flight_planning.flight_intent_validation import (
     ExpectedFlightIntent,
     validate_flight_intent_templates,
@@ -69,6 +72,7 @@ class ConflictEqualPriorityNotPermitted(TestScenario):
     tested_uss: FlightPlannerClient
     control_uss: FlightPlannerClient
     dss: DSSInstance
+    flight_intents_templates: dict[FlightIntentID, FlightInfoTemplate]
 
     def __init__(
         self,
@@ -90,8 +94,13 @@ def __init__(
             )
 
         self.dss = dss.get_instance(scopes)
+        self.flight_intents_templates = (
+            flight_intents.get_flight_intents() if flight_intents else {}
+        )
 
-        expected_flight_intents = [
+    @property
+    def _expected_flight_intents(self) -> list[ExpectedFlightIntent]:
+        return [
             ExpectedFlightIntent(
                 "flight1_planned",
                 "Flight 1",
@@ -159,25 +168,26 @@ def __init__(
             ),  # Note: this intent expected to produce Nonconforming state, but this is hard to verify without telemetry.  UAS state is not actually off-nominal.
         ]
 
-        templates = flight_intents.get_flight_intents()
+    def resolve_flight(self, flight_template: FlightInfoTemplate) -> FlightInfo:
+        return flight_template.resolve(self.time_context.evaluate_now())
+
+    def run(self, context: ExecutionContext):
         try:
-            self._intents_extent = validate_flight_intent_templates(
-                templates, expected_flight_intents
+            intents_extents = validate_flight_intent_templates(
+                self.flight_intents_templates, self._expected_flight_intents
             )
         except ValueError as e:
             raise ValueError(
                 f"`{self.me()}` TestScenario requirements for flight_intents not met: {e}"
             )
 
-        for efi in expected_flight_intents:
+        for efi in self._expected_flight_intents:
             setattr(
-                self, efi.intent_id.replace("equal_prio_", ""), templates[efi.intent_id]
+                self,
+                efi.intent_id.replace("equal_prio_", ""),
+                self.flight_intents_templates[efi.intent_id],
             )
 
-    def resolve_flight(self, flight_template: FlightInfoTemplate) -> FlightInfo:
-        return flight_template.resolve(self.time_context.evaluate_now())
-
-    def run(self, context: ExecutionContext):
         self.begin_test_scenario(context)
 
         self.record_note(
@@ -194,7 +204,7 @@ def run(self, context: ExecutionContext):
         validate_clear_area(
             self,
             self.dss,
-            [self._intents_extent],
+            [intents_extents],
             ignore_self=True,
         )
         self.end_test_step()

monitoring/uss_qualifier/scenarios/astm/utm/off_nominal_planning/down_uss.py

@@ -18,10 +18,14 @@
     PlanningActivityResult,
 )
 from monitoring.monitorlib.fetch import QueryError
+from monitoring.monitorlib.geotemporal import Volume4D
 from monitoring.monitorlib.testing import make_fake_url
 from monitoring.uss_qualifier.resources.astm.f3548.v21 import DSSInstanceResource
 from monitoring.uss_qualifier.resources.astm.f3548.v21.dss import DSSInstance
 from monitoring.uss_qualifier.resources.flight_planning import FlightIntentsResource
+from monitoring.uss_qualifier.resources.flight_planning.flight_intent import (
+    FlightIntentID,
+)
 from monitoring.uss_qualifier.resources.flight_planning.flight_intent_validation import (
     ExpectedFlightIntent,
     validate_flight_intent_templates,
@@ -49,10 +53,12 @@ class DownUSS(TestScenario):
     flight1_planned: FlightInfoTemplate
 
     uss_qualifier_sub: str
+    intents_extents: Volume4D
 
     tested_uss: FlightPlannerClient
     dss_resource: DSSInstanceResource
     dss: DSSInstance
+    flight_intents_templates: dict[FlightIntentID, FlightInfoTemplate]
 
     def __init__(
         self,
@@ -64,19 +70,7 @@ def __init__(
         self.dss_resource = dss
         self.tested_uss = tested_uss.client
         self.dss = dss.get_instance(self._dss_req_scopes)
-
-        templates = flight_intents.get_flight_intents()
-        try:
-            self._intents_extent = validate_flight_intent_templates(
-                templates, self._expected_flight_intents
-            )
-        except ValueError as e:
-            raise ValueError(
-                f"`{self.me()}` TestScenario requirements for flight_intents not met: {e}"
-            )
-
-        for efi in self._expected_flight_intents:
-            setattr(self, efi.intent_id, templates[efi.intent_id])
+        self.flight_intents_templates = flight_intents.get_flight_intents()
 
     @property
     def _dss_req_scopes(self) -> dict[str, str]:
@@ -120,11 +114,23 @@ def run(self, context: ExecutionContext):
         self.end_test_scenario()
 
     def _setup(self):
+        try:
+            self.intents_extents = validate_flight_intent_templates(
+                self.flight_intents_templates, self._expected_flight_intents
+            )
+        except ValueError as e:
+            raise ValueError(
+                f"`{self.me()}` TestScenario requirements for flight_intents not met: {e}"
+            )
+
+        for efi in self._expected_flight_intents:
+            setattr(self, efi.intent_id, self.flight_intents_templates[efi.intent_id])
+
         self.begin_test_step("Resolve USS ID of virtual USS")
         with self.check("Successful dummy query", [self.dss.participant_id]) as check:
             try:
                 _, dummy_query = self.dss.find_op_intent(
-                    self._intents_extent.to_f3548v21()
+                    self.intents_extents.to_f3548v21()
                 )
                 self.record_query(dummy_query)
             except QueryError as e:
@@ -154,7 +160,7 @@ def _setup(self):
         validate_clear_area(
             self,
             self.dss,
-            [self._intents_extent],
+            [self.intents_extents],
             ignore_self=True,
         )
         self.end_test_step()
@@ -281,14 +287,14 @@ def _clear_op_intents(self):
         ) as check:
             try:
                 oi_refs, find_query = self.dss.find_op_intent(
-                    self._intents_extent.to_f3548v21()
+                    self.intents_extents.to_f3548v21()
                 )
                 self.record_query(find_query)
             except QueryError as e:
                 self.record_queries(e.queries)
                 find_query = e.queries[0]
                 check.record_failed(
-                    summary=f"Failed to query operational intent references from DSS in {self._intents_extent} for cleanup",
+                    summary=f"Failed to query operational intent references from DSS in {self.intents_extents} for cleanup",
                     details=f"DSS responded code {find_query.status_code}; {e}",
                     query_timestamps=[find_query.request.timestamp],
                 )