diff --git a/monitoring/monitorlib/geotemporal.py b/monitoring/monitorlib/geotemporal.py index 51f4d1b861..6b7301ef4f 100644 --- a/monitoring/monitorlib/geotemporal.py +++ b/monitoring/monitorlib/geotemporal.py @@ -405,7 +405,7 @@ def time_start(self) -> Optional[Time]: @property def time_end(self) -> Optional[Time]: return ( - Time(value=min(v.time_end.datetime for v in self.volumes)) + Time(value=max(v.time_end.datetime for v in self.volumes)) if all("time_end" in v and v.time_end for v in self.volumes) else None ) diff --git a/monitoring/monitorlib/scd_automated_testing/scd_injection_api.py b/monitoring/monitorlib/scd_automated_testing/scd_injection_api.py index 793e8e6560..d8023a2288 100644 --- a/monitoring/monitorlib/scd_automated_testing/scd_injection_api.py +++ b/monitoring/monitorlib/scd_automated_testing/scd_injection_api.py @@ -87,7 +87,6 @@ class InjectFlightResponse(ImplicitDict): result: InjectFlightResult notes: Optional[str] - operational_intent_id: Optional[str] class DeleteFlightResult(str, Enum): diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/flight_intent_validation/flight_intent_validation.md b/monitoring/uss_qualifier/scenarios/astm/utm/flight_intent_validation/flight_intent_validation.md index 5a51238d0a..9d572c3d46 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/flight_intent_validation/flight_intent_validation.md +++ b/monitoring/uss_qualifier/scenarios/astm/utm/flight_intent_validation/flight_intent_validation.md @@ -75,6 +75,8 @@ to reject or accept the flight. If the USS indicates that the injection attempt ### [Plan valid flight intent test step](../../../flight_planning/plan_flight_intent.md) The valid flight intent should be successfully planned by the flight planner. +### [Validate flight sharing test step](../validate_shared_operational_intent.md) + ### Attempt to modify planned flight with an off-nominal volume test step The user flight intent that the test driver attempts to modify has an off-nominal volume and is in the `Accepted` state. As such, the modification attempt should be rejected. @@ -95,6 +97,8 @@ Validate that the planned flight intent was not modified with an off-nominal vol ### [Activate valid flight intent test step](../../../flight_planning/activate_flight_intent.md) The valid flight intent should be successfully activated by the flight planner. +### [Validate flight sharing test step](../validate_shared_operational_intent.md) + ### Attempt to modify activated flight with an off-nominal volume test step The user flight intent that the test driver attempts to modify has an off-nominal volume and is in the `Activated` state. As such, the modification attempt should be rejected. @@ -125,6 +129,10 @@ Validate that the flight intent was shared correctly and is discoverable. The flight intent should be successfully transition to Ended state by the flight planner. ### Validate flight intent is non-discoverable test step + +#### DSS responses check +**[astm.f3548.v21.DSS0005](../../../../requirements/astm/f3548/v21.md)** + #### Operational intent not shared check If the operational intent is still discoverable after it was transitioned to Ended, this check will fail per **[astm.f3548.v21.OPIN0040](../../../../requirements/astm/f3548/v21.md)**. diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/flight_intent_validation/flight_intent_validation.py b/monitoring/uss_qualifier/scenarios/astm/utm/flight_intent_validation/flight_intent_validation.py index 24d90d4e2b..b47add0750 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/flight_intent_validation/flight_intent_validation.py +++ b/monitoring/uss_qualifier/scenarios/astm/utm/flight_intent_validation/flight_intent_validation.py @@ -21,8 +21,7 @@ FlightPlannerResource, ) from monitoring.uss_qualifier.scenarios.astm.utm.test_steps import ( - validate_shared_operational_intent, - ValidateNotSharedOperationalIntent, + OpIntentValidator, ) from monitoring.uss_qualifier.scenarios.scenario import TestScenario from monitoring.uss_qualifier.scenarios.flight_planning.test_steps import ( @@ -61,6 +60,15 @@ def __init__( self.dss = dss.dss flight_intents = flight_intents.get_flight_intents() + + extents = [] + for intent in flight_intents.values(): + extents.extend(intent.request.operational_intent.volumes) + extents.extend(intent.request.operational_intent.off_nominal_volumes) + self._intents_extent = Volume4DCollection.from_f3548v21( + extents + ).bounding_volume.to_f3548v21() + try: ( self.valid_flight, @@ -210,14 +218,14 @@ def _setup(self) -> bool: return True def _attempt_invalid(self): - with ValidateNotSharedOperationalIntent( + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight intent too far ahead of time not planned", - self.invalid_too_far_away.request, - ): - resp, _ = submit_flight_intent( + self._intents_extent, + ) as validator: + submit_flight_intent( self, "Attempt to plan flight intent too far ahead of time", "Incorrectly planned", @@ -226,16 +234,17 @@ def _attempt_invalid(self): self.tested_uss, self.invalid_too_far_away.request, ) + validator.expect_not_shared() def _attempt_invalid_offnominal(self): - with ValidateNotSharedOperationalIntent( + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight intent with an off-nominal volume not planned", - self.invalid_accepted_offnominal.request, - ): - resp, _ = submit_flight_intent( + self._intents_extent, + ) as validator: + submit_flight_intent( self, "Attempt to plan flight with an off-nominal volume", "Incorrectly planned", @@ -244,84 +253,115 @@ def _attempt_invalid_offnominal(self): self.tested_uss, self.invalid_accepted_offnominal.request, ) + validator.expect_not_shared() - resp, valid_flight_id = plan_flight_intent( - self, "Plan valid flight intent", self.tested_uss, self.valid_flight.request - ) - valid_flight_op_intent_id = resp.operational_intent_id - - resp, _ = submit_flight_intent( + with OpIntentValidator( self, - "Attempt to modify planned flight with an off-nominal volume", - "Incorrectly modified", - {InjectFlightResult.Rejected}, - {InjectFlightResult.Failed: "Failure"}, self.tested_uss, - self.invalid_accepted_offnominal.request, - ) - validate_shared_operational_intent( + self.dss, + "Validate flight sharing", + self._intents_extent, + ) as validator: + _, valid_flight_id = plan_flight_intent( + self, + "Plan valid flight intent", + self.tested_uss, + self.valid_flight.request, + ) + valid_flight_op_intent_ref = validator.expect_shared( + self.valid_flight.request + ) + + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate planned flight not modified", - self.valid_flight.request, - valid_flight_op_intent_id, - ) + self._intents_extent, + valid_flight_op_intent_ref, + ) as validator: + submit_flight_intent( + self, + "Attempt to modify planned flight with an off-nominal volume", + "Incorrectly modified", + {InjectFlightResult.Rejected}, + {InjectFlightResult.Failed: "Failure"}, + self.tested_uss, + self.invalid_accepted_offnominal.request, + ) + valid_flight_op_intent_ref = validator.expect_shared( + self.valid_flight.request + ) - resp = activate_flight_intent( + with OpIntentValidator( self, - "Activate valid flight intent", self.tested_uss, - self.valid_activated.request, - valid_flight_id, - ) - valid_flight_op_intent_id = resp.operational_intent_id + self.dss, + "Validate flight sharing", + self._intents_extent, + valid_flight_op_intent_ref, + ) as validator: + activate_flight_intent( + self, + "Activate valid flight intent", + self.tested_uss, + self.valid_activated.request, + valid_flight_id, + ) + valid_flight_op_intent_ref = validator.expect_shared( + self.valid_activated.request + ) - resp, _ = submit_flight_intent( - self, - "Attempt to modify activated flight with an off-nominal volume", - "Incorrectly modified", - {InjectFlightResult.Rejected}, - {InjectFlightResult.Failed: "Failure"}, - self.tested_uss, - self.invalid_activated_offnominal.request, - ) - validate_shared_operational_intent( + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate activated flight not modified", - self.valid_activated.request, - valid_flight_op_intent_id, - ) + self._intents_extent, + valid_flight_op_intent_ref, + ) as validator: + submit_flight_intent( + self, + "Attempt to modify activated flight with an off-nominal volume", + "Incorrectly modified", + {InjectFlightResult.Rejected}, + {InjectFlightResult.Failed: "Failure"}, + self.tested_uss, + self.invalid_activated_offnominal.request, + ) + validator.expect_shared(self.valid_flight.request) _ = delete_flight_intent( self, "Delete valid flight intent", self.tested_uss, valid_flight_id ) def _validate_ended_cancellation(self): - with ValidateNotSharedOperationalIntent( + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight intent is non-discoverable", - self.valid_flight.request, - ): - resp, flight_id = plan_flight_intent( - self, "Plan flight intent", self.tested_uss, self.valid_flight.request - ) - validate_shared_operational_intent( + self._intents_extent, + ) as cancelled_validator: + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight intent shared correctly", - self.valid_flight.request, - resp.operational_intent_id, - ) + self._intents_extent, + ) as planned_validator: + _, flight_id = plan_flight_intent( + self, + "Plan flight intent", + self.tested_uss, + self.valid_flight.request, + ) + planned_validator.expect_shared(self.valid_flight.request) _ = delete_flight_intent( self, "Cancel flight intent", self.tested_uss, flight_id ) + cancelled_validator.expect_not_shared() def _validate_precision_intersection(self): _, _ = plan_flight_intent( @@ -331,14 +371,14 @@ def _validate_precision_intersection(self): self.valid_flight.request, ) - with ValidateNotSharedOperationalIntent( + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate conflicting flight not planned", - self.valid_conflict_tiny_overlap.request, - ): - resp, _ = submit_flight_intent( + self._intents_extent, + ) as validator: + submit_flight_intent( self, "Attempt to plan flight conflicting by a tiny overlap", "Incorrectly planned", @@ -347,6 +387,7 @@ def _validate_precision_intersection(self): self.tested_uss, self.valid_conflict_tiny_overlap.request, ) + validator.expect_not_shared() def cleanup(self): self.begin_cleanup() diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_equal_priority_not_permitted/conflict_equal_priority_not_permitted.md b/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_equal_priority_not_permitted/conflict_equal_priority_not_permitted.md index a846f50355..4e5f4896c6 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_equal_priority_not_permitted/conflict_equal_priority_not_permitted.md +++ b/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_equal_priority_not_permitted/conflict_equal_priority_not_permitted.md @@ -69,6 +69,8 @@ Both USSs are requested to remove all flights from the area under test. ### [Plan flight 2 test step](../../../../flight_planning/plan_flight_intent.md) Flight 2 on time range B should be successfully planned by the control USS. +### [Validate flight 2 sharing test step](../../validate_shared_operational_intent.md) + ### [Activate flight 2 test step](../../../../flight_planning/activate_flight_intent.md) Flight 2 on time range B should be successfully activated by the control USS. diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_equal_priority_not_permitted/conflict_equal_priority_not_permitted.py b/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_equal_priority_not_permitted/conflict_equal_priority_not_permitted.py index a5762e58eb..147a0ec01e 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_equal_priority_not_permitted/conflict_equal_priority_not_permitted.py +++ b/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_equal_priority_not_permitted/conflict_equal_priority_not_permitted.py @@ -1,5 +1,8 @@ from typing import Optional +from uas_standards.astm.f3548.v21.api import ( + OperationalIntentReference, +) from monitoring.monitorlib.geotemporal import Volume4DCollection from monitoring.uss_qualifier.common_data_definitions import Severity from uas_standards.astm.f3548.v21.api import OperationalIntentState @@ -22,8 +25,7 @@ FlightPlannerResource, ) from monitoring.uss_qualifier.scenarios.astm.utm.test_steps import ( - validate_shared_operational_intent, - ValidateNotSharedOperationalIntent, + OpIntentValidator, ) from monitoring.uss_qualifier.scenarios.flight_planning.prioritization_test_steps import ( modify_planned_conflict_flight_intent, @@ -82,6 +84,15 @@ def __init__( raise ScenarioCannotContinueError(msg) flight_intents = flight_intents.get_flight_intents() + + extents = [] + for intent in flight_intents.values(): + extents.extend(intent.request.operational_intent.volumes) + extents.extend(intent.request.operational_intent.off_nominal_volumes) + self._intents_extent = Volume4DCollection.from_f3548v21( + extents + ).bounding_volume.to_f3548v21() + try: ( self.flight_1_planned_time_range_A, @@ -203,7 +214,7 @@ def run(self): self.end_test_case() self.begin_test_case("Attempt to plan flight in conflict") - self._attempt_plan_flight_conflict() + flight_2_oi_ref = self._attempt_plan_flight_conflict() self.end_test_case() self.begin_test_case("Attempt to activate flight in conflict") @@ -211,15 +222,19 @@ def run(self): self.end_test_case() self.begin_test_case("Attempt to modify planned flight in conflict") - self._attempt_modify_planned_flight_conflict() + flight_1_oi_ref = self._attempt_modify_planned_flight_conflict() self.end_test_case() self.begin_test_case("Attempt to modify activated flight in conflict") - flight_1_op_intent_id = self._attempt_modify_activated_flight_conflict() + flight_1_oi_ref = self._attempt_modify_activated_flight_conflict( + flight_1_oi_ref + ) self.end_test_case() self.begin_test_case("Modify activated flight with pre-existing conflict") - self._modify_activated_flight_preexisting_conflict(flight_1_op_intent_id) + self._modify_activated_flight_preexisting_conflict( + flight_1_oi_ref, flight_2_oi_ref + ) self.end_test_case() self.end_test_scenario() @@ -261,210 +276,244 @@ def _setup(self) -> bool: return True - def _attempt_plan_flight_conflict(self): - _, self.flight_2_id = plan_flight_intent( - self, - "Plan flight 2", - self.control_uss, - self.flight_2_equal_prio_planned_time_range_B.request, - ) - - resp_flight_2 = activate_flight_intent( + def _attempt_plan_flight_conflict(self) -> OperationalIntentReference: + with OpIntentValidator( self, - "Activate flight 2", self.control_uss, - self.flight_2_equal_prio_activated_time_range_B.request, - self.flight_2_id, - ) + self.dss, + "Validate flight 2 sharing", + self._intents_extent, + ) as validator: + _, self.flight_2_id = plan_flight_intent( + self, + "Plan flight 2", + self.control_uss, + self.flight_2_equal_prio_planned_time_range_B.request, + ) + flight_2_oi_ref = validator.expect_shared( + self.flight_2_equal_prio_planned_time_range_B.request + ) - validate_shared_operational_intent( + with OpIntentValidator( self, self.control_uss, self.dss, "Validate flight 2 sharing", - self.flight_2_equal_prio_activated_time_range_B.request, - resp_flight_2.operational_intent_id, - ) + self._intents_extent, + flight_2_oi_ref, + ) as validator: + activate_flight_intent( + self, + "Activate flight 2", + self.control_uss, + self.flight_2_equal_prio_activated_time_range_B.request, + self.flight_2_id, + ) + flight_2_oi_ref = validator.expect_shared( + self.flight_2_equal_prio_activated_time_range_B.request + ) - with ValidateNotSharedOperationalIntent( + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight 1 not shared", - self.flight_1_planned_time_range_B.request, - ): - _ = plan_conflict_flight_intent( + self._intents_extent, + ) as validator: + plan_conflict_flight_intent( self, "Attempt to plan flight 1", self.tested_uss, self.flight_1_planned_time_range_B.request, ) + validator.expect_not_shared() + + return flight_2_oi_ref def _attempt_activate_flight_conflict(self): - with ValidateNotSharedOperationalIntent( + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight 1 not shared", - self.flight_1_activated_time_range_B.request, - ): - _ = activate_conflict_flight_intent( + self._intents_extent, + ) as validator: + activate_conflict_flight_intent( self, "Attempt to directly activate conflicting flight 1", self.tested_uss, self.flight_1_activated_time_range_B.request, self.flight_1_id, ) + validator.expect_not_shared() - def _attempt_modify_planned_flight_conflict(self): - resp_flight_1, self.flight_1_id = plan_flight_intent( - self, - "Plan flight 1", - self.tested_uss, - self.flight_1_planned_time_range_A.request, - ) - validate_shared_operational_intent( + def _attempt_modify_planned_flight_conflict( + self, + ) -> Optional[OperationalIntentReference]: + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight 1 sharing", - self.flight_1_planned_time_range_A.request, - resp_flight_1.operational_intent_id, - ) - - _ = modify_planned_conflict_flight_intent( - self, - "Attempt to modify planned flight 1 in conflict", - self.tested_uss, - self.flight_1_planned_time_range_B.request, - self.flight_1_id, - ) + self._intents_extent, + ) as validator: + _, self.flight_1_id = plan_flight_intent( + self, + "Plan flight 1", + self.tested_uss, + self.flight_1_planned_time_range_A.request, + ) + flight_1_oi_ref = validator.expect_shared( + self.flight_1_planned_time_range_A.request + ) - validate_shared_operational_intent( + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight 1 not modified", - self.flight_1_planned_time_range_A.request, - resp_flight_1.operational_intent_id, - skip_if_not_found=True, - ) + self._intents_extent, + flight_1_oi_ref, + ) as validator: + modify_planned_conflict_flight_intent( + self, + "Attempt to modify planned flight 1 in conflict", + self.tested_uss, + self.flight_1_planned_time_range_B.request, + self.flight_1_id, + ) + flight_1_oi_ref = validator.expect_shared( + self.flight_1_planned_time_range_A.request, skip_if_not_found=True + ) - def _attempt_modify_activated_flight_conflict(self) -> str: - resp_flight_1 = activate_flight_intent( - self, - "Activate flight 1", - self.tested_uss, - self.flight_1_activated_time_range_A.request, - self.flight_1_id, - ) - validate_shared_operational_intent( + return flight_1_oi_ref + + def _attempt_modify_activated_flight_conflict( + self, flight_1_oi_ref: Optional[OperationalIntentReference] + ) -> Optional[OperationalIntentReference]: + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight 1 sharing", - self.flight_1_activated_time_range_A.request, - resp_flight_1.operational_intent_id, - ) - - _ = modify_activated_conflict_flight_intent( - self, - "Attempt to modify activated flight 1 in conflict", - self.tested_uss, - self.flight_1_activated_time_range_B.request, - self.flight_1_id, - ) + self._intents_extent, + flight_1_oi_ref, + ) as validator: + activate_flight_intent( + self, + "Activate flight 1", + self.tested_uss, + self.flight_1_activated_time_range_A.request, + self.flight_1_id, + ) + flight_1_oi_ref = validator.expect_shared( + self.flight_1_activated_time_range_A.request + ) - validate_shared_operational_intent( + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight 1 not modified", - self.flight_1_activated_time_range_A.request, - resp_flight_1.operational_intent_id, - skip_if_not_found=True, - ) + self._intents_extent, + flight_1_oi_ref, + ) as validator: + modify_activated_conflict_flight_intent( + self, + "Attempt to modify activated flight 1 in conflict", + self.tested_uss, + self.flight_1_activated_time_range_B.request, + self.flight_1_id, + ) + flight_1_oi_ref = validator.expect_shared( + self.flight_1_activated_time_range_A.request, skip_if_not_found=True + ) - return resp_flight_1.operational_intent_id + return flight_1_oi_ref def _modify_activated_flight_preexisting_conflict( - self, orig_flight_1_op_intent_id: str + self, + flight_1_oi_ref: Optional[OperationalIntentReference], + flight_2_oi_ref: Optional[OperationalIntentReference], ): - resp_flight_1 = activate_flight_intent( - self, - "Activate flight 1", - self.tested_uss, - self.flight_1_activated_time_range_A.request, - self.flight_1_id, - ) - validate_shared_operational_intent( + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight 1 sharing", - self.flight_1_activated_time_range_A.request, - resp_flight_1.operational_intent_id, - ) + self._intents_extent, + flight_1_oi_ref, + ) as validator: + activate_flight_intent( + self, + "Activate flight 1", + self.tested_uss, + self.flight_1_activated_time_range_A.request, + self.flight_1_id, + ) + flight_1_oi_ref = validator.expect_shared( + self.flight_1_activated_time_range_A.request + ) # TODO: the following call requires the control USS to support CMSA role, # but as there is currently no explicit way of knowing if it is the case # or not, we assume that a Rejected result means the USS does not # support the CMSA role, in which case we interrupt the scenario. - resp_flight_2, _ = submit_flight_intent( - self, - "Declare flight 2 non-conforming", - "Successful transition to non-conforming state", - {InjectFlightResult.Planned, InjectFlightResult.Rejected}, - {InjectFlightResult.Failed: "Failure"}, - self.control_uss, - self.flight_2_equal_prio_nonconforming_time_range_A.request, - self.flight_2_id, - ) - if resp_flight_2.result == InjectFlightResult.Rejected: - msg = f"{self.control_uss.config.participant_id} rejected transition to a Nonconforming state because it does not support CMSA role, execution of the scenario was stopped without failure" - self.record_note("Control USS does not support CMSA role", msg) - raise ScenarioCannotContinueError(msg) - - validate_shared_operational_intent( + with OpIntentValidator( self, self.control_uss, self.dss, "Validate flight 2 sharing", - self.flight_2_equal_prio_nonconforming_time_range_A.request, - resp_flight_2.operational_intent_id, - ) + self._intents_extent, + flight_2_oi_ref, + ) as validator: + resp_flight_2, _ = submit_flight_intent( + self, + "Declare flight 2 non-conforming", + "Successful transition to non-conforming state", + {InjectFlightResult.Planned, InjectFlightResult.Rejected}, + {InjectFlightResult.Failed: "Failure"}, + self.control_uss, + self.flight_2_equal_prio_nonconforming_time_range_A.request, + self.flight_2_id, + ) + if resp_flight_2.result == InjectFlightResult.Rejected: + msg = f"{self.control_uss.config.participant_id} rejected transition to a Nonconforming state because it does not support CMSA role, execution of the scenario was stopped without failure" + self.record_note("Control USS does not support CMSA role", msg) + raise ScenarioCannotContinueError(msg) + + validator.expect_shared( + self.flight_2_equal_prio_nonconforming_time_range_A.request + ) - resp_flight_1, _ = submit_flight_intent( + with OpIntentValidator( self, - "Attempt to modify activated flight 1 in conflict with activated flight 2", - "Successful modification or rejection", - {InjectFlightResult.ReadyToFly, InjectFlightResult.Rejected}, - {InjectFlightResult.Failed: "Failure"}, self.tested_uss, - self.flight_1_activated_time_range_A_extended.request, - self.flight_1_id, - ) - - if resp_flight_1.result == InjectFlightResult.ReadyToFly: - validate_shared_operational_intent( + self.dss, + "Validate flight 1", + self._intents_extent, + flight_1_oi_ref, + ) as validator: + resp_flight_1, _ = submit_flight_intent( self, + "Attempt to modify activated flight 1 in conflict with activated flight 2", + "Successful modification or rejection", + {InjectFlightResult.ReadyToFly, InjectFlightResult.Rejected}, + {InjectFlightResult.Failed: "Failure"}, self.tested_uss, - self.dss, - "Validate flight 1", self.flight_1_activated_time_range_A_extended.request, - resp_flight_1.operational_intent_id, - ) - elif resp_flight_1.result == InjectFlightResult.Rejected: - validate_shared_operational_intent( - self, - self.tested_uss, - self.dss, - "Validate flight 1", - self.flight_1_activated_time_range_A.request, - orig_flight_1_op_intent_id, - skip_if_not_found=True, + self.flight_1_id, ) + if resp_flight_1.result == InjectFlightResult.ReadyToFly: + validator.expect_shared( + self.flight_1_activated_time_range_A_extended.request + ) + elif resp_flight_1.result == InjectFlightResult.Rejected: + validator.expect_shared( + self.flight_1_activated_time_range_A.request, skip_if_not_found=True + ) + def cleanup(self): self.begin_cleanup() cleanup_flights(self, (self.control_uss, self.tested_uss)) diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_higher_priority/conflict_higher_priority.md b/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_higher_priority/conflict_higher_priority.md index 600e7b7c46..1f7ae2a47a 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_higher_priority/conflict_higher_priority.md +++ b/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_higher_priority/conflict_higher_priority.md @@ -62,14 +62,14 @@ Both USSs are requested to remove all flights from the area under test. ### [Plan flight 2 test step](../../../../flight_planning/plan_flight_intent.md) The higher priority flight should be successfully planned by the control USS. +### [Validate flight 2 sharing test step](../../validate_shared_operational_intent.md) + ### [Attempt to plan flight 1 test step](../../../../flight_planning/plan_priority_conflict_flight_intent.md) The test driver attempts to plan the flight 1 via the tested USS. However, it conflicts with flight 2, which is of higher priority. As such it should be rejected per **[astm.f3548.v21.SCD0015](../../../../../requirements/astm/f3548/v21.md)**. ### [Validate flight 1 not shared test step](../../validate_not_shared_operational_intent.md) -### [Validate flight 2 sharing test step](../../validate_shared_operational_intent.md) - ### [Delete flight 2 test step](../../../../flight_planning/delete_flight_intent.md) @@ -83,6 +83,8 @@ The first flight should be successfully planned by the tested USS. The second flight should be successfully planned by the control USS. It conflicts with flight 1, but it has higher priority. +### [Validate flight 2 sharing test step](../../validate_shared_operational_intent.md) + ### [Attempt to modify planned flight 1 in conflict test step](../../../../flight_planning/modify_planned_priority_conflict_flight_intent.md) The test driver attempts to modify flight 1 via the tested USS, which is planned. However, it conflicts with flight 2, which is of higher priority and was planned in the meantime. @@ -92,8 +94,6 @@ As such it should be rejected per **[astm.f3548.v21.SCD0020](../../../../../requ Because the modification attempt was invalid, either Flight 1 should not have been modified (because the USS kept the original accepted request), or it should have been removed (because the USS rejected the replacement plan provided). -### [Validate flight 2 sharing test step](../../validate_shared_operational_intent.md) - ## Attempt to activate flight in conflict test case ### [Attempt to activate conflicting flight 1 test step](../../../../flight_planning/activate_priority_conflict_flight_intent.md) @@ -119,9 +119,13 @@ directly activated without being planned beforehand. ### [Plan flight 2 test step](../../../../flight_planning/plan_flight_intent.md) The second flight should be successfully planned by the control USS. +### [Validate flight 2 sharing test step](../../validate_shared_operational_intent.md) + ### [Activate flight 2 test step](../../../../flight_planning/activate_flight_intent.md) The test driver activates flight 2, which should be done successfully given that it is the highest-priority flight. +### [Validate flight 2 sharing test step](../../validate_shared_operational_intent.md) + ### [Modify activated flight 1 in conflict with activated flight 2 test step](../../../../flight_planning/modify_activated_flight_intent.md) Before execution of this step, flights 1 and 2 are activated and in conflict. The test driver modifies flight 1 in a way that still conflicts with flight 2. @@ -131,8 +135,6 @@ the modification is accepted per **[astm.f3548.v21.SCD0030](../../../../../requi ### [Validate flight 1 sharing test step](../../validate_shared_operational_intent.md) The first flight should have been modified. -### [Validate flight 2 sharing test step](../../validate_shared_operational_intent.md) - ## Attempt to modify activated flight in conflict test case ### [Modify activated flight 2 to not conflict with activated flight 1 test step](../../../../flight_planning/modify_planned_flight_intent.md) @@ -140,6 +142,8 @@ The test driver modifies (activated) flight 2 with the control USS so that it is flight of test USS. As flight 2 is of higher priority, this should succeed and leave flight 1 clear of conflict. +### [Validate flight 2 sharing test step](../../validate_shared_operational_intent.md) + ### [Attempt to modify activated flight 1 in conflict test step](../../../../flight_planning/modify_activated_priority_conflict_flight_intent.md) The test driver attempts to modify flight 1 so that it becomes in conflict with flight 2. Both flights are activated at that point. However, because the conflict did not exist when the modification was initiated, it should be rejected @@ -149,8 +153,6 @@ per **[astm.f3548.v21.SCD0030](../../../../../requirements/astm/f3548/v21.md)**. Because the modification attempt was invalid, either Flight 1 should not have been modified (because the USS kept the original accepted request), or it should have been removed (because the USS rejected the replacement plan provided). -### [Validate flight 2 sharing test step](../../validate_shared_operational_intent.md) - ## Cleanup ### Successful flight deletion check diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_higher_priority/conflict_higher_priority.py b/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_higher_priority/conflict_higher_priority.py index a904e5193f..5ba3d6dccb 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_higher_priority/conflict_higher_priority.py +++ b/monitoring/uss_qualifier/scenarios/astm/utm/nominal_planning/conflict_higher_priority/conflict_higher_priority.py @@ -1,12 +1,13 @@ -from typing import Optional +from typing import Optional, Tuple + +from uas_standards.astm.f3548.v21.api import ( + OperationalIntentReference, +) from monitoring.monitorlib.geotemporal import Volume4DCollection from monitoring.uss_qualifier.common_data_definitions import Severity from uas_standards.astm.f3548.v21.api import OperationalIntentState -from monitoring.monitorlib.scd_automated_testing.scd_injection_api import ( - Capability, -) 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 ( @@ -22,8 +23,7 @@ FlightPlannerResource, ) from monitoring.uss_qualifier.scenarios.astm.utm.test_steps import ( - validate_shared_operational_intent, - ValidateNotSharedOperationalIntent, + OpIntentValidator, ) from monitoring.uss_qualifier.scenarios.flight_planning.prioritization_test_steps import ( activate_priority_conflict_flight_intent, @@ -72,6 +72,15 @@ def __init__( self.dss = dss.dss flight_intents = flight_intents.get_flight_intents() + + extents = [] + for intent in flight_intents.values(): + extents.extend(intent.request.operational_intent.volumes) + extents.extend(intent.request.operational_intent.off_nominal_volumes) + self._intents_extent = Volume4DCollection.from_f3548v21( + extents + ).bounding_volume.to_f3548v21() + try: ( self.flight_1_planned_time_range_A, @@ -186,19 +195,22 @@ def run(self): self.end_test_case() self.begin_test_case("Attempt to modify planned flight in conflict") - flight_1_op_intent_id = self._attempt_modify_planned_flight_conflict() + flight_1_oi_ref = self._attempt_modify_planned_flight_conflict() self.end_test_case() self.begin_test_case("Attempt to activate flight in conflict") - self._attempt_activate_flight_conflict(flight_1_op_intent_id) + flight_1_oi_ref = self._attempt_activate_flight_conflict(flight_1_oi_ref) self.end_test_case() self.begin_test_case("Modify activated flight with pre-existing conflict") - flight_1_op_intent_id = self._modify_activated_flight_conflict_preexisting() + ( + flight_1_oi_ref, + flight_2_oi_ref, + ) = self._modify_activated_flight_conflict_preexisting(flight_1_oi_ref) self.end_test_case() self.begin_test_case("Attempt to modify activated flight in conflict") - self._attempt_modify_activated_flight_conflict(flight_1_op_intent_id) + self._attempt_modify_activated_flight_conflict(flight_1_oi_ref, flight_2_oi_ref) self.end_test_case() self.end_test_scenario() @@ -241,209 +253,246 @@ def _setup(self) -> bool: return True def _attempt_plan_flight_conflict(self): - resp_flight_2, self.flight_2_id = plan_flight_intent( + with OpIntentValidator( self, - "Plan flight 2", self.control_uss, - self.flight_2_planned_time_range_A.request, - ) + self.dss, + "Validate flight 2 sharing", + self._intents_extent, + ) as validator: + resp_flight_2, self.flight_2_id = plan_flight_intent( + self, + "Plan flight 2", + self.control_uss, + self.flight_2_planned_time_range_A.request, + ) + validator.expect_shared(self.flight_2_planned_time_range_A.request) - with ValidateNotSharedOperationalIntent( + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight 1 not shared", - self.flight_1_planned_time_range_A.request, - ): + self._intents_extent, + ) as validator: _ = plan_priority_conflict_flight_intent( self, "Attempt to plan flight 1", self.tested_uss, self.flight_1_planned_time_range_A.request, ) - - validate_shared_operational_intent( - self, - self.control_uss, - self.dss, - "Validate flight 2 sharing", - self.flight_2_planned_time_range_A.request, - resp_flight_2.operational_intent_id, - ) + validator.expect_not_shared() _ = delete_flight_intent( self, "Delete flight 2", self.control_uss, self.flight_2_id ) self.flight_2_id = None - def _attempt_modify_planned_flight_conflict(self) -> Optional[str]: - resp_flight_1, self.flight_1_id = plan_flight_intent( - self, - "Plan flight 1", - self.tested_uss, - self.flight_1_planned_time_range_A.request, - ) - validate_shared_operational_intent( + def _attempt_modify_planned_flight_conflict( + self, + ) -> Optional[OperationalIntentReference]: + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight 1 sharing", - self.flight_1_planned_time_range_A.request, - resp_flight_1.operational_intent_id, - ) + self._intents_extent, + ) as validator: + resp_flight_1, self.flight_1_id = plan_flight_intent( + self, + "Plan flight 1", + self.tested_uss, + self.flight_1_planned_time_range_A.request, + ) + flight_1_oi_ref = validator.expect_shared( + self.flight_1_planned_time_range_A.request + ) - resp_flight_2, self.flight_2_id = plan_flight_intent( + with OpIntentValidator( self, - "Plan flight 2", self.control_uss, - self.flight_2_planned_time_range_A.request, - ) - - _ = modify_planned_priority_conflict_flight_intent( - self, - "Attempt to modify planned flight 1 in conflict", - self.tested_uss, - self.flight_1_planned_time_range_A_extended.request, - self.flight_1_id, - ) + self.dss, + "Validate flight 2 sharing", + self._intents_extent, + ) as validator: + resp_flight_2, self.flight_2_id = plan_flight_intent( + self, + "Plan flight 2", + self.control_uss, + self.flight_2_planned_time_range_A.request, + ) + validator.expect_shared(self.flight_2_planned_time_range_A.request) - validate_shared_operational_intent( + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight 1 not modified", - self.flight_1_planned_time_range_A.request, - resp_flight_1.operational_intent_id, - skip_if_not_found=True, - ) - validate_shared_operational_intent( - self, - self.control_uss, - self.dss, - "Validate flight 2 sharing", - self.flight_2_planned_time_range_A.request, - resp_flight_2.operational_intent_id, - ) + self._intents_extent, + flight_1_oi_ref, + ) as validator: + _ = modify_planned_priority_conflict_flight_intent( + self, + "Attempt to modify planned flight 1 in conflict", + self.tested_uss, + self.flight_1_planned_time_range_A_extended.request, + self.flight_1_id, + ) + flight_1_oi_ref = validator.expect_shared( + self.flight_1_planned_time_range_A.request, skip_if_not_found=True + ) - return resp_flight_1.operational_intent_id + return flight_1_oi_ref - def _attempt_activate_flight_conflict(self, flight_1_op_intent_id: str): - _ = activate_priority_conflict_flight_intent( + def _attempt_activate_flight_conflict( + self, flight_1_oi_ref: Optional[OperationalIntentReference] + ) -> Optional[OperationalIntentReference]: + with OpIntentValidator( self, - "Attempt to activate conflicting flight 1", self.tested_uss, - self.flight_1_activated_time_range_A.request, - self.flight_1_id, - ) - - validate_shared_operational_intent( - self, - self.control_uss, self.dss, "Validate flight 1 not activated", - self.flight_1_planned_time_range_A.request, - flight_1_op_intent_id, - skip_if_not_found=True, - ) + self._intents_extent, + flight_1_oi_ref, + ) as validator: + _ = activate_priority_conflict_flight_intent( + self, + "Attempt to activate conflicting flight 1", + self.tested_uss, + self.flight_1_activated_time_range_A.request, + self.flight_1_id, + ) + flight_1_oi_ref = validator.expect_shared( + self.flight_1_planned_time_range_A.request, skip_if_not_found=True + ) + + return flight_1_oi_ref - def _modify_activated_flight_conflict_preexisting(self) -> str: + def _modify_activated_flight_conflict_preexisting( + self, flight_1_oi_ref: Optional[OperationalIntentReference] + ) -> Tuple[OperationalIntentReference, OperationalIntentReference]: _ = delete_flight_intent( self, "Delete flight 2", self.control_uss, self.flight_2_id ) self.flight_2_id = None - resp_flight_1 = activate_flight_intent( - self, - "Activate flight 1", - self.tested_uss, - self.flight_1_activated_time_range_A.request, - self.flight_1_id, - ) - validate_shared_operational_intent( + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight 1 sharing", - self.flight_1_activated_time_range_A.request, - resp_flight_1.operational_intent_id, - ) + self._intents_extent, + flight_1_oi_ref, + ) as validator: + activate_flight_intent( + self, + "Activate flight 1", + self.tested_uss, + self.flight_1_activated_time_range_A.request, + self.flight_1_id, + ) + flight_1_oi_ref = validator.expect_shared( + self.flight_1_activated_time_range_A.request + ) - _, self.flight_2_id = plan_flight_intent( + with OpIntentValidator( self, - "Plan flight 2", self.control_uss, - self.flight_2_planned_time_range_A.request, - ) + self.dss, + "Validate flight 2 sharing", + self._intents_extent, + ) as validator: + _, self.flight_2_id = plan_flight_intent( + self, + "Plan flight 2", + self.control_uss, + self.flight_2_planned_time_range_A.request, + ) + flight_2_oi_ref = validator.expect_shared( + self.flight_2_planned_time_range_A.request + ) - resp_flight_2 = activate_flight_intent( + with OpIntentValidator( self, - "Activate flight 2", self.control_uss, - self.flight_2_activated_time_range_A.request, - self.flight_2_id, - ) - - resp_flight_1 = modify_activated_flight_intent( - self, - "Modify activated flight 1 in conflict with activated flight 2", - self.tested_uss, - self.flight_1_activated_time_range_A_extended.request, - self.flight_1_id, - ) + self.dss, + "Validate flight 2 sharing", + self._intents_extent, + flight_2_oi_ref, + ) as validator: + activate_flight_intent( + self, + "Activate flight 2", + self.control_uss, + self.flight_2_activated_time_range_A.request, + self.flight_2_id, + ) + flight_2_oi_ref = validator.expect_shared( + self.flight_2_activated_time_range_A.request + ) - validate_shared_operational_intent( + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight 1 sharing", - self.flight_1_activated_time_range_A_extended.request, - resp_flight_1.operational_intent_id, - ) - validate_shared_operational_intent( - self, - self.control_uss, - self.dss, - "Validate flight 2 sharing", - self.flight_2_activated_time_range_A.request, - resp_flight_2.operational_intent_id, - ) + self._intents_extent, + flight_1_oi_ref, + ) as validator: + modify_activated_flight_intent( + self, + "Modify activated flight 1 in conflict with activated flight 2", + self.tested_uss, + self.flight_1_activated_time_range_A_extended.request, + self.flight_1_id, + ) + flight_1_oi_ref = validator.expect_shared( + self.flight_1_activated_time_range_A_extended.request + ) - return resp_flight_1.operational_intent_id + return flight_1_oi_ref, flight_2_oi_ref - def _attempt_modify_activated_flight_conflict(self, flight_1_op_intent_id: str): - resp_flight_2 = modify_activated_flight_intent( + def _attempt_modify_activated_flight_conflict( + self, + flight_1_oi_ref: OperationalIntentReference, + flight_2_oi_ref: OperationalIntentReference, + ): + with OpIntentValidator( self, - "Modify activated flight 2 to not conflict with activated flight 1", self.control_uss, - self.flight_2_activated_time_range_B.request, - self.flight_2_id, - ) - - _ = modify_activated_priority_conflict_flight_intent( - self, - "Attempt to modify activated flight 1 in conflict", - self.tested_uss, - self.flight_1_activated_time_range_B.request, - self.flight_1_id, - ) + self.dss, + "Validate flight 2 sharing", + self._intents_extent, + flight_2_oi_ref, + ) as validator: + modify_activated_flight_intent( + self, + "Modify activated flight 2 to not conflict with activated flight 1", + self.control_uss, + self.flight_2_activated_time_range_B.request, + self.flight_2_id, + ) + validator.expect_shared(self.flight_2_activated_time_range_B.request) - validate_shared_operational_intent( + with OpIntentValidator( self, self.tested_uss, self.dss, "Validate flight 1 not modified", - self.flight_1_activated_time_range_A_extended.request, - flight_1_op_intent_id, - skip_if_not_found=True, - ) - validate_shared_operational_intent( - self, - self.control_uss, - self.dss, - "Validate flight 2 sharing", - self.flight_2_activated_time_range_B.request, - resp_flight_2.operational_intent_id, - ) + self._intents_extent, + flight_1_oi_ref, + ) as validator: + modify_activated_priority_conflict_flight_intent( + self, + "Attempt to modify activated flight 1 in conflict", + self.tested_uss, + self.flight_1_activated_time_range_B.request, + self.flight_1_id, + ) + validator.expect_shared( + self.flight_1_activated_time_range_A_extended.request, + skip_if_not_found=True, + ) def cleanup(self): self.begin_cleanup() diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py b/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py index c2cee3e5e3..2347b330eb 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py +++ b/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py @@ -1,4 +1,6 @@ -from typing import List +from __future__ import annotations + +from typing import List, Optional from monitoring.monitorlib import schema_validation, fetch from monitoring.monitorlib.geotemporal import Volume4DCollection @@ -19,26 +21,28 @@ from monitoring.uss_qualifier.scenarios.astm.utm.evaluation import ( validate_op_intent_details, ) -from monitoring.uss_qualifier.scenarios.scenario import TestScenarioType - +from monitoring.uss_qualifier.scenarios.scenario import ( + TestScenarioType, + TestRunCannotContinueError, +) -class ValidateNotSharedOperationalIntent(object): - """Validate that an operational intent information was not shared with the DSS by comparing the operational intents - found in the area of the flight intent before and after the planning attempt. - This assumes an area lock on the extent of the flight intent. - This class is meant to be used within a `with` statement. - It implements the test step described in validate_not_shared_operational_intent.md. +class OpIntentValidator(object): + """ + This class enables the validation of the sharing (or not) of an operational + intent with the DSS. It does so by comparing the operational intents found + in the area of the intent before and after a planning attempt. + It is meant to be used within a `with` statement. + It assumes an area lock on the extent of the flight intent. """ - _scenario: TestScenarioType - _flight_planner: FlightPlanner - _dss: DSSInstance - _test_step: str + _before_oi_refs: List[OperationalIntentReference] + _before_query: fetch.Query - _flight_intent_extent: Volume4D - _initial_op_intent_refs: List[OperationalIntentReference] - _initial_query: fetch.Query + _after_oi_refs: List[OperationalIntentReference] + _after_query: fetch.Query + + _new_oi_ref: Optional[OperationalIntentReference] = None def __init__( self, @@ -46,191 +50,246 @@ def __init__( flight_planner: FlightPlanner, dss: DSSInstance, test_step: str, - flight_intent: InjectFlightRequest, + extent: Volume4D, + orig_oi_ref: Optional[OperationalIntentReference] = None, ): - self._scenario = scenario - self._flight_planner = flight_planner - self._dss = dss - self._test_step = test_step - - self._flight_intent_extent = Volume4DCollection.from_f3548v21( - flight_intent.operational_intent.volumes - + flight_intent.operational_intent.off_nominal_volumes - ).bounding_volume.to_f3548v21() - - def __enter__(self): - self._initial_op_intent_refs, self._initial_query = self._dss.find_op_intent( - self._flight_intent_extent + """ + :param scenario: + :param flight_planner: + :param dss: + :param test_step: + :param extent: the extent over which the operational intents are to be compared. + :param orig_oi_ref: if this is validating a previously existing operational intent (e.g. modification), pass the original reference. + """ + self._scenario: TestScenarioType = scenario + self._flight_planner: FlightPlanner = flight_planner + self._dss: DSSInstance = dss + self._test_step: str = test_step + self._extent: Volume4D = extent + self._orig_oi_ref: Optional[OperationalIntentReference] = orig_oi_ref + + def __enter__(self) -> OpIntentValidator: + self._before_oi_refs, self._before_query = self._dss.find_op_intent( + self._extent ) + return self def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is not None: self._scenario.record_note( self._flight_planner.participant_id, - f"Exception occurred during ValidateNotSharedOperationalIntent ({exc_type}: {exc_val}).", + f"Exception occurred during OpIntentValidator ({exc_type}: {exc_val}).", ) raise exc_val + def _find_after_oi(self, oi_id: str) -> Optional[OperationalIntentReference]: + found = [oi_ref for oi_ref in self._after_oi_refs if oi_ref.id == oi_id] + return found[0] if len(found) != 0 else None + + def _begin_step(self): + self._after_oi_refs, self._after_query = self._dss.find_op_intent(self._extent) + oi_ids_delta = {oi_ref.id for oi_ref in self._after_oi_refs} - { + oi_ref.id for oi_ref in self._before_oi_refs + } + + if ( + len(oi_ids_delta) > 1 + ): # TODO: could a USS cut up a submitted flight intent into several op intents? + raise TestRunCannotContinueError( + f"unexpectedly got more than 1 new operational intent after planning request was created (IDs: {oi_ids_delta}): the test scenario might be malformed or some external requests might have interfered" + ) + if len(oi_ids_delta) == 1: + self._new_oi_ref = self._find_after_oi(oi_ids_delta.pop()) + self._scenario.begin_test_step(self._test_step) - self._scenario.record_query(self._initial_query) + self._scenario.record_query(self._before_query) + self._scenario.record_query(self._after_query) - op_intent_refs, query = self._dss.find_op_intent(self._flight_intent_extent) - self._scenario.record_query(query) + with self._scenario.check("DSS responses", [self._dss.participant_id]) as check: + if self._before_query.status_code != 200: + check.record_failed( + summary="Failed to query DSS for operational intents before planning request", + severity=Severity.High, + details=f"Received status code {self._before_query.status_code} from the DSS", + query_timestamps=[self._before_query.request.timestamp], + ) + if self._after_query.status_code != 200: + check.record_failed( + summary="Failed to query DSS for operational intents after planning request", + severity=Severity.High, + details=f"Received status code {self._after_query.status_code} from the DSS", + query_timestamps=[self._after_query.request.timestamp], + ) + + def expect_not_shared(self) -> None: + """Validate that an operational intent information was not shared with the DSS. + + It implements the test step described in validate_not_shared_operational_intent.md. + """ + self._begin_step() - oi_ids_delta = {oi_ref.id for oi_ref in op_intent_refs} - { - oi_ref.id for oi_ref in self._initial_op_intent_refs - } with self._scenario.check( "Operational intent not shared", [self._flight_planner.participant_id] ) as check: - if len(oi_ids_delta) > 0: + if self._new_oi_ref is not None: check.record_failed( summary="Operational intent reference was incorrectly shared with DSS", severity=Severity.High, - details=f"USS {self._flight_planner.participant_id} was not supposed to share an operational intent with the DSS, but new operational intent(s) with ID(s) {oi_ids_delta} were found", - query_timestamps=[query.request.timestamp], + details=f"USS {self._flight_planner.participant_id} was not supposed to share an operational intent with the DSS, but the new operational intent with ID {self._new_oi_ref.id} was found", + query_timestamps=[self._after_query.request.timestamp], ) self._scenario.end_test_step() + def expect_shared( + self, flight_intent: InjectFlightRequest, skip_if_not_found: bool = False + ) -> Optional[OperationalIntentReference]: + """Validate that operational intent information was correctly shared for a flight intent. -def validate_shared_operational_intent( - scenario: TestScenarioType, - flight_planner: FlightPlanner, - dss: DSSInstance, - test_step: str, - flight_intent: InjectFlightRequest, - op_intent_id: str, - skip_if_not_found: bool = False, -) -> bool: - """Validate that operational intent information was correctly shared for a flight intent. + This function implements the test step described in validate_shared_operational_intent.md. - This function implements the test step described in - validate_shared_operational_intent.md. + :param flight_intent: the flight intent that was supposed to have been shared. + :param skip_if_not_found: set to True to skip the execution of the checks if the operational intent was not found while it should have been modified. - :returns: True if the operational intent was validated. May return False without failing a check e.g. if the - operational intent was not found and skip_if_not_found was True. - """ - scenario.begin_test_step(test_step) - extent = Volume4DCollection.from_f3548v21( - flight_intent.operational_intent.volumes - + flight_intent.operational_intent.off_nominal_volumes - ).bounding_volume.to_f3548v21() - op_intent_refs, query = dss.find_op_intent(extent) - scenario.record_query(query) - with scenario.check("DSS response", [dss.participant_id]) as check: - if query.status_code != 200: - check.record_failed( - summary="Failed to query DSS for operational intents", - severity=Severity.High, - details=f"Received status code {query.status_code} from the DSS", - query_timestamps=[query.request.timestamp], - ) + :returns: the shared operational intent reference. None if skipped because not found. + """ + self._begin_step() + + with self._scenario.check( + "Operational intent shared correctly", [self._flight_planner.participant_id] + ) as check: + if self._orig_oi_ref is None: + # we expect a new op intent to have been created + if self._new_oi_ref is None: + check.record_failed( + summary="Operational intent reference not found in DSS", + severity=Severity.High, + details=f"USS {self._flight_planner.participant_id} was supposed to have shared a new operational intent with the DSS, but no matching operational intent references were found in the DSS in the area of the flight intent", + query_timestamps=[self._after_query.request.timestamp], + ) + oi_ref = self._new_oi_ref + + elif self._new_oi_ref is None: + # we expect the original op intent to have been either modified or left untouched, thus must be among the returned op intents + # exception made if skip_if_not_found=True and op intent was deleted: step is skipped + modified_oi_ref = self._find_after_oi(self._orig_oi_ref.id) + if modified_oi_ref is None: + if not skip_if_not_found: + check.record_failed( + summary="Operational intent reference not found in DSS", + severity=Severity.High, + details=f"USS {self._flight_planner.participant_id} was supposed to have shared with the DSS an updated operational intent by modifying it, but no matching operational intent references were found in the DSS in the area of the flight intent", + query_timestamps=[self._after_query.request.timestamp], + ) + else: + self._scenario.record_note( + self._flight_planner.participant_id, + f"Operational intent reference with ID {self._orig_oi_ref.id} not found in DSS, instructed to skip test step.", + ) + self._scenario.end_test_step() + return None + oi_ref = modified_oi_ref + + else: + # we expect the original op intent to have been replaced with a new one, thus old one must NOT be among the returned op intents + if self._find_after_oi(self._orig_oi_ref.id) is not None: + check.record_failed( + summary="Operational intent reference found duplicated in DSS", + severity=Severity.High, + details=f"USS {self._flight_planner.participant_id} was supposed to have shared with the DSS an updated operational intent by replacing it, but it ended up duplicating the operational intent in the DSS", + query_timestamps=[self._after_query.request.timestamp], + ) + oi_ref = self._new_oi_ref - matching_op_intent_refs = [ - op_intent_ref - for op_intent_ref in op_intent_refs - if op_intent_ref.id == op_intent_id - ] - with scenario.check( - "Operational intent shared correctly", [flight_planner.participant_id] - ) as check: - if not matching_op_intent_refs: - if not skip_if_not_found: + oi_full, oi_full_query = self._dss.get_full_op_intent(oi_ref) + self._scenario.record_query(oi_full_query) + with self._scenario.check( + "Operational intent details retrievable", + [self._flight_planner.participant_id], + ) as check: + if oi_full_query.status_code != 200: check.record_failed( - summary="Operational intent reference not found in DSS", + summary="Operational intent details could not be retrieved from USS", severity=Severity.High, - details=f"USS {flight_planner.participant_id} was supposed to have shared an operational intent with ID {op_intent_id}, but no operational intent references with that ID were found in the DSS in the area of the flight intent", - query_timestamps=[query.request.timestamp], - ) - else: - scenario.record_note( - flight_planner.participant_id, - f"Operational intent reference with ID {op_intent_id} not found in DSS, instructed to skip test step.", + details=f"Received status code {oi_full_query.status_code} from {self._flight_planner.participant_id} when querying for details of operational intent {oi_ref.id}", + query_timestamps=[oi_full_query.request.timestamp], ) - scenario.end_test_step() - return False - op_intent_ref = matching_op_intent_refs[0] - - op_intent, query = dss.get_full_op_intent(op_intent_ref) - scenario.record_query(query) - with scenario.check( - "Operational intent details retrievable", [flight_planner.participant_id] - ) as check: - if query.status_code != 200: - check.record_failed( - summary="Operational intent details could not be retrieved from USS", - severity=Severity.High, - details=f"Received status code {query.status_code} from {flight_planner.participant_id} when querying for details of operational intent {op_intent_id}", - query_timestamps=[query.request.timestamp], - ) - with scenario.check( - "Operational intent details data format", [flight_planner.participant_id] - ) as check: - errors = schema_validation.validate( - schema_validation.F3548_21.OpenAPIPath, - schema_validation.F3548_21.GetOperationalIntentDetailsResponse, - query.response.json, - ) - if errors: - check.record_failed( - summary="Operational intent details response failed schema validation", - severity=Severity.Medium, - details="The response received from querying operational intent details failed validation against the required OpenAPI schema:\n" - + "\n".join( - f"At {e.json_path} in the response: {e.message}" for e in errors - ), - query_timestamps=[query.request.timestamp], + with self._scenario.check( + "Operational intent details data format", + [self._flight_planner.participant_id], + ) as check: + errors = schema_validation.validate( + schema_validation.F3548_21.OpenAPIPath, + schema_validation.F3548_21.GetOperationalIntentDetailsResponse, + oi_full_query.response.json, ) + if errors: + check.record_failed( + summary="Operational intent details response failed schema validation", + severity=Severity.Medium, + details="The response received from querying operational intent details failed validation against the required OpenAPI schema:\n" + + "\n".join( + f"At {e.json_path} in the response: {e.message}" for e in errors + ), + query_timestamps=[oi_full_query.request.timestamp], + ) - error_text = validate_op_intent_details( - op_intent.details, flight_intent.operational_intent.priority, extent - ) - with scenario.check( - "Correct operational intent details", [flight_planner.participant_id] - ) as check: - if error_text: - check.record_failed( - summary="Operational intent details do not match user flight intent", - severity=Severity.High, - details=error_text, - query_timestamps=[query.request.timestamp], + with self._scenario.check( + "Correct operational intent details", [self._flight_planner.participant_id] + ) as check: + error_text = validate_op_intent_details( + oi_full.details, + flight_intent.operational_intent.priority, + Volume4DCollection.from_f3548v21( + flight_intent.operational_intent.volumes + + flight_intent.operational_intent.off_nominal_volumes + ).bounding_volume.to_f3548v21(), ) + if error_text: + check.record_failed( + summary="Operational intent details do not match user flight intent", + severity=Severity.High, + details=error_text, + query_timestamps=[oi_full_query.request.timestamp], + ) - with scenario.check( - "Off-nominal volumes", [flight_planner.participant_id] - ) as check: - if ( - op_intent.reference.state == OperationalIntentState.Accepted - or op_intent.reference.state == OperationalIntentState.Activated - ) and op_intent.details.get("off_nominal_volumes", None): - check.record_failed( - summary="Accepted or Activated operational intents are not allowed off-nominal volumes", - severity=Severity.Medium, - details=f"Operational intent {op_intent.reference.id} was {op_intent.reference.state} and had {len(op_intent.details.off_nominal_volumes)} off-nominal volumes", - query_timestamps=[query.request.timestamp], - ) + with self._scenario.check( + "Off-nominal volumes", [self._flight_planner.participant_id] + ) as check: + if ( + oi_full.reference.state == OperationalIntentState.Accepted + or oi_full.reference.state == OperationalIntentState.Activated + ) and oi_full.details.get("off_nominal_volumes", None): + check.record_failed( + summary="Accepted or Activated operational intents are not allowed off-nominal volumes", + severity=Severity.Medium, + details=f"Operational intent {oi_full.reference.id} was {oi_full.reference.state} and had {len(oi_full.details.off_nominal_volumes)} off-nominal volumes", + query_timestamps=[oi_full_query.request.timestamp], + ) + + with self._scenario.check( + "Vertices", [self._flight_planner.participant_id] + ) as check: - all_volumes = op_intent.details.get("volumes", []) + op_intent.details.get( - "off_nominal_volumes", [] - ) - - def volume_vertices(v4): - if "outline_circle" in v4.volume: - return 1 - if "outline_polygon" in v4.volume: - return len(v4.volume.outline_polygon.vertices) - - n_vertices = sum(volume_vertices(v) for v in all_volumes) - with scenario.check("Vertices", [flight_planner.participant_id]) as check: - if n_vertices > 10000: - check.record_failed( - summary="Too many vertices", - severity=Severity.Medium, - details=f"Operational intent {op_intent.reference.id} had {n_vertices} vertices total", - query_timestamps=[query.request.timestamp], + def volume_vertices(v4): + if "outline_circle" in v4.volume: + return 1 + if "outline_polygon" in v4.volume: + return len(v4.volume.outline_polygon.vertices) + + all_volumes = oi_full.details.get("volumes", []) + oi_full.details.get( + "off_nominal_volumes", [] ) + n_vertices = sum(volume_vertices(v) for v in all_volumes) - scenario.end_test_step() - return True + if n_vertices > 10000: + check.record_failed( + summary="Too many vertices", + severity=Severity.Medium, + details=f"Operational intent {oi_full.reference.id} had {n_vertices} vertices total", + query_timestamps=[oi_full_query.request.timestamp], + ) + + self._scenario.end_test_step() + return oi_ref diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/validate_not_shared_operational_intent.md b/monitoring/uss_qualifier/scenarios/astm/utm/validate_not_shared_operational_intent.md index 3053515fe4..e80962bfc2 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/validate_not_shared_operational_intent.md +++ b/monitoring/uss_qualifier/scenarios/astm/utm/validate_not_shared_operational_intent.md @@ -4,7 +4,11 @@ This step verifies that a previous attempt to create a flight did not result in It does so by querying the DSS for operational intents in the area of the flight before and after an attempted creation. This assumes an area lock on the extent of the flight intent. -See `ValidateNotSharedOperationalIntent` in [test_steps.py](test_steps.py). +See `OpIntentValidator.expect_not_shared()` in [test_steps.py](test_steps.py). + +## DSS responses check + +**astm.f3548.v21.DSS0005** ## Operational intent not shared check If there are new operational intent references in the area of the flight intent, this check will fail per diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/validate_shared_operational_intent.md b/monitoring/uss_qualifier/scenarios/astm/utm/validate_shared_operational_intent.md index e10fe2ffd4..35430975b7 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/validate_shared_operational_intent.md +++ b/monitoring/uss_qualifier/scenarios/astm/utm/validate_shared_operational_intent.md @@ -1,8 +1,8 @@ # Validate flight sharing test step -This step verifies that a created flight is shared properly per ASTM F3548-21 by querying the DSS for flights in the area of the flight intent, and then retrieving the details from the USS if the operational intent reference is found. See `validate_shared_operational_intent` in [test_steps.py](test_steps.py). +This step verifies that a created flight is shared properly per ASTM F3548-21 by querying the DSS for flights in the area of the flight intent, and then retrieving the details from the USS if the operational intent reference is found. See `OpIntentValidator.expect_shared()` in [test_steps.py](test_steps.py). -## DSS response check +## DSS responses check **astm.f3548.v21.DSS0005**