From 8c42a555e4135383afdf254b2142b8151cb6e9b1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micka=C3=ABl=20Misbach?= Date: Fri, 9 Feb 2024 14:30:55 +0100 Subject: [PATCH] [uss_qualifier/resources/f3548/dss/get_full_op_intent] Use QueryError for error handling --- .../resources/astm/f3548/v21/dss.py | 37 ++++------- .../scenarios/astm/utm/test_steps.py | 63 +++++++++++-------- 2 files changed, 48 insertions(+), 52 deletions(-) diff --git a/monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py b/monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py index 6ac6f5456e..d1f4e77485 100644 --- a/monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py +++ b/monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py @@ -177,28 +177,10 @@ def get_full_op_intent( op_intent_ref: OperationalIntentReference, uss_participant_id: Optional[str] = None, ) -> Tuple[OperationalIntent, Query]: - result, query = self.get_full_op_intent_without_validation( - op_intent_ref, - uss_participant_id, - ) - if query.status_code != 200: - result = None - else: - result = ImplicitDict.parse( - query.response.json, GetOperationalIntentDetailsResponse - ).operational_intent - return result, query - - def get_full_op_intent_without_validation( - self, - op_intent_ref: OperationalIntentReference, - uss_participant_id: Optional[str] = None, - ) -> Tuple[Dict, Query]: """ - GET OperationalIntent without validating, as invalid data expected for negative tests - - Returns: - returns the response json when query is successful + Retrieve a full operational intent from its managing USS. + Raises: + * QueryError: if request failed, if HTTP status code is different than 200, or if the parsing of the response failed. """ self._uses_scope(Scope.StrategicCoordination) op = OPERATIONS[OperationID.GetOperationalIntentDetails] @@ -210,11 +192,14 @@ def get_full_op_intent_without_validation( uss_participant_id, scope=Scope.StrategicCoordination, ) - result = None - if query.status_code == 200: - result = query.response.json - - return result, query + if query.status_code != 200: + raise QueryError( + f"Received code {query.status_code} when attempting to retrieve operational intent details for {op_intent_ref.id}{f'; error message: `{query.error_message}`' if query.error_message is not None else ''}", + query, + ) + else: + result = query.parse_json_result(GetOperationalIntentDetailsResponse) + return result.operational_intent, query def get_op_intent_telemetry( self, diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py b/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py index 3c23af23d2..6e8c2e05c1 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py +++ b/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py @@ -10,6 +10,7 @@ Volume4D, OperationalIntentReference, GetOperationalIntentDetailsResponse, + EntityID, ) from uas_standards.astm.f3548.v21.constants import Scope from monitoring.monitorlib.clients.flight_planning.flight_info import ( @@ -237,12 +238,25 @@ def expect_shared_with_invalid_data( if oi_ref is None: return None - goidr_json, oi_full_query = self._dss.get_full_op_intent_without_validation( - oi_ref, self._flight_planner.participant_id - ) - - self._scenario.record_query(oi_full_query) - self._operational_intent_retrievable_check(oi_full_query, oi_ref.id) + with self._scenario.check( + "Operational intent details retrievable", + [self._flight_planner.participant_id], + ) as check: + try: + goidr_json, oi_full_query = self._dss.get_full_op_intent( + oi_ref, self._flight_planner.participant_id + ) + self._scenario.record_query(oi_full_query) + except QueryError as e: + self._scenario.record_queries(e.queries) + oi_full_query = e.queries[0] + if oi_full_query.status_code != 200: + # fail only if details could not be retrieved, as validation failures are acceptable here + check.record_failed( + summary="Operational intent details could not be retrieved from USS", + 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}; {e}", + query_timestamps=[oi_full_query.request.timestamp], + ) validation_failures = self._evaluate_op_intent_validation(oi_full_query) expected_validation_failure_found = self._expected_validation_failure_found( @@ -264,21 +278,6 @@ def expect_shared_with_invalid_data( return oi_ref - def _operational_intent_retrievable_check( - self, oi_full_query: fetch.Query, ref_id: str - ): - 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 details could not be retrieved from USS", - severity=Severity.High, - details=f"Received status code {oi_full_query.status_code} from {self._flight_planner.participant_id} when querying for details of operational intent {ref_id}", - query_timestamps=[oi_full_query.request.timestamp], - ) - def _operational_intent_shared_check( self, flight_intent: FlightInfo, @@ -365,11 +364,23 @@ def _operational_intent_shared_check( def _check_op_intent_details( self, flight_intent: FlightInfo, oi_ref: OperationalIntentReference ): - oi_full, oi_full_query = self._dss.get_full_op_intent( - oi_ref, self._flight_planner.participant_id - ) - self._scenario.record_query(oi_full_query) - self._operational_intent_retrievable_check(oi_full_query, oi_ref.id) + with self._scenario.check( + "Operational intent details retrievable", + [self._flight_planner.participant_id], + ) as check: + try: + oi_full, oi_full_query = self._dss.get_full_op_intent( + oi_ref, self._flight_planner.participant_id + ) + self._scenario.record_query(oi_full_query) + except QueryError as e: + self._scenario.record_queries(e.queries) + oi_full_query = e.queries[0] + check.record_failed( + summary="Operational intent details could not be retrieved from USS", + 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}; {e}", + query_timestamps=[oi_full_query.request.timestamp], + ) validation_failures = self._evaluate_op_intent_validation(oi_full_query) with self._scenario.check(