[uss_qualifier/resources/f3548/dss/delete_op_intent] Use QueryError f…

…or error handling (#483)

* [uss_qualifier/resources/f3548/dss/delete_op_intent] Use QueryError for error handling

* fix hanging 'make format'

monitoring/monitorlib/fetch/__init__.py

@@ -4,7 +4,7 @@
 import traceback
 import uuid
 from enum import Enum
-from typing import Dict, Optional, List, Union
+from typing import Dict, Optional, List, Union, TypeVar, Type
 from urllib.parse import urlparse
 
 import flask
@@ -387,6 +387,9 @@ def dss_delete_isa(rid_version: RIDVersion):
             raise ValueError(f"Unsupported RID version: {rid_version}")
 
 
+ResponseType = TypeVar("ResponseType", bound=ImplicitDict)
+
+
 class Query(ImplicitDict):
     request: RequestDescription
     response: ResponseDescription
@@ -422,6 +425,24 @@ def get_client_sub(self):
             )
             return payload["sub"]
 
+    def parse_json_result(self, parse_type: Type[ResponseType]) -> ResponseType:
+        """Parses the JSON result into the specified type.
+
+        Args:
+            parse_type: ImplicitDict type to parse into.
+        Returns:
+             the parsed response (of type `parse_type`).
+        Raises:
+            * QueryError: if the parsing failed.
+        """
+        try:
+            return parse_type(ImplicitDict.parse(self.response.json, parse_type))
+        except (ValueError, TypeError, KeyError) as e:
+            raise QueryError(
+                f"Parsing JSON response into type {parse_type.__name__} failed with exception {type(e).__name__}: {e}",
+                self,
+            )
+
 
 class QueryError(RuntimeError):
     """Error encountered when interacting with a server in the UTM ecosystem."""

monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py

@@ -34,7 +34,7 @@
 from uas_standards.astm.f3548.v21.constants import Scope
 
 from monitoring.monitorlib import infrastructure, fetch
-from monitoring.monitorlib.fetch import QueryType, Query, query_and_describe
+from monitoring.monitorlib.fetch import QueryType, Query, query_and_describe, QueryError
 from monitoring.monitorlib.fetch import scd as fetch
 from monitoring.monitorlib.fetch.scd import FetchedSubscription, FetchedSubscriptions
 from monitoring.monitorlib.inspection import calling_function_name, fullname
@@ -285,11 +285,7 @@ def delete_op_intent(
         self,
         id: str,
         ovn: str,
-    ) -> Tuple[
-        Optional[OperationalIntentReference],
-        Optional[List[SubscriberToNotify]],
-        Query,
-    ]:
+    ) -> Tuple[OperationalIntentReference, List[SubscriberToNotify], Query]:
         self._uses_scope(Scope.StrategicCoordination)
         op = OPERATIONS[OperationID.DeleteOperationalIntentReference]
         query = query_and_describe(
@@ -301,17 +297,13 @@ def delete_op_intent(
             scope=Scope.StrategicCoordination,
         )
         if query.status_code != 200:
-            return None, None, query
+            raise QueryError(
+                f"Received code {query.status_code} when attempting to delete operational intent {id}",
+                query,
+            )
         else:
-            try:
-                result = ChangeOperationalIntentReferenceResponse(
-                    ImplicitDict.parse(
-                        query.response.json, ChangeOperationalIntentReferenceResponse
-                    )
-                )
-                return result.operational_intent_reference, result.subscribers, query
-            except ValueError as e:
-                return None, None, query
+            result = query.parse_json_result(ChangeOperationalIntentReferenceResponse)
+            return result.operational_intent_reference, result.subscribers, query
 
     def set_uss_availability(
         self,

monitoring/uss_qualifier/scenarios/astm/utm/dss/test_step_fragments.py

@@ -3,6 +3,7 @@
 import loguru
 
 from monitoring.monitorlib.mutate.scd import MutatedSubscription
+from monitoring.monitorlib import fetch
 from monitoring.uss_qualifier.resources.astm.f3548.v21.dss import DSSInstance
 from monitoring.uss_qualifier.scenarios.scenario import TestScenarioType
 from uas_standards.astm.f3548.v21.api import EntityID, Volume4D
@@ -17,16 +18,17 @@ def remove_op_intent(
 
     This function implements the test step fragment described in remove_op_intent.md.
     """
-    removed_ref, subscribers_to_notify, query = dss.delete_op_intent(oi_id, ovn)
-    scenario.record_query(query)
-
     with scenario.check(
         "Operational intent reference removed", dss.participant_id
     ) as check:
-        if removed_ref is None:
+        try:
+            removed_ref, subscribers_to_notify, query = dss.delete_op_intent(oi_id, ovn)
+            scenario.record_query(query)
+        except fetch.QueryError as e:
+            scenario.record_queries(e.queries)
             check.record_failed(
                 summary=f"Could not remove op intent reference {oi_id}",
-                details=f"When attempting to remove op intent reference {oi_id} from the DSS, received {query.status_code}",
+                details=f"When attempting to remove op intent reference {oi_id} from the DSS, received {query.status_code}; {e}",
                 query_timestamps=[query.request.timestamp],
             )
 

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

@@ -2,6 +2,7 @@
 
 import arrow
 
+from monitoring.monitorlib.fetch import QueryError
 from monitoring.monitorlib.geotemporal import Volume4DCollection
 from monitoring.uss_qualifier.common_data_definitions import Severity
 from uas_standards.astm.f3548.v21.api import (
@@ -286,15 +287,16 @@ def _clear_op_intents(self):
 
             for oi_ref in oi_refs:
                 if oi_ref.manager == self.uss_qualifier_sub:
-                    del_oi, _, del_query = self.dss.delete_op_intent(
-                        oi_ref.id, oi_ref.ovn
-                    )
-                    self.record_query(del_query)
-
-                    if del_oi is None:
+                    try:
+                        del_oi, _, del_query = self.dss.delete_op_intent(
+                            oi_ref.id, oi_ref.ovn
+                        )
+                        self.record_query(del_query)
+                    except QueryError as e:
+                        self.record_queries(e.queries)
                         check.record_failed(
                             summary=f"Failed to delete op intent {oi_ref.id} from DSS",
-                            details=f"DSS responded code {del_query.status_code}; error message: {del_query.error_message}",
+                            details=f"DSS responded code {del_query.status_code}; error message: {del_query.error_message}; {e}",
                             query_timestamps=[del_query.request.timestamp],
                         )
 

monitoring/uss_qualifier/scenarios/astm/utm/op_intent_ref_access_control.py

@@ -4,6 +4,7 @@
 from uas_standards.astm.f3548.v21.api import OperationalIntentState
 from uas_standards.astm.f3548.v21.constants import Scope
 
+from monitoring.monitorlib.fetch import QueryError
 from monitoring.monitorlib.geotemporal import Volume4DCollection
 from monitoring.prober.infrastructure import register_resource_type
 from monitoring.uss_qualifier.resources.astm.f3548.v21 import DSSInstanceResource
@@ -149,16 +150,20 @@ def _clean_known_op_intents_ids(self):
                     query_timestamps=[q.request.timestamp],
                 )
         if q.response.status_code != 404:
-            (_, notifs, dq) = self._dss.delete_op_intent(self._oid_1, oi_ref.ovn)
-            self.record_query(dq)
-            if dq.response.status_code != 200:
-                with self.check(
-                    "Operational intent references can be searched",
-                    self._pid,
-                ) as check:
+            with self.check(
+                "Operational intent references can be searched for",
+                self._pid,
+            ) as check:
+                try:
+                    (_, notifs, dq) = self._dss.delete_op_intent(
+                        self._oid_1, oi_ref.ovn
+                    )
+                    self.record_query(dq)
+                except QueryError as e:
+                    self.record_queries(e.queries)
                     check.record_failed(
                         f"Could not delete operational intent using main credentials",
-                        details=f"DSS responded with {dq.response.status_code} to attempt to delete OI {self._oid_1}",
+                        details=f"DSS responded with {dq.response.status_code} to attempt to delete OI {self._oid_1}; {e}",
                         query_timestamps=[dq.request.timestamp],
                     )
 
@@ -175,17 +180,19 @@ def _clean_known_op_intents_ids(self):
                     query_timestamps=[q.request.timestamp],
                 )
         if q.response.status_code != 404:
-            (_, notifs, dq) = self._dss_separate_creds.delete_op_intent(
-                self._oid_2, oi_ref.ovn
-            )
-            self.record_query(dq)
             with self.check(
                 "Operational intent references can be deleted by their owner", self._pid
             ) as check:
-                if dq.response.status_code != 200:
+                try:
+                    (_, notifs, dq) = self._dss_separate_creds.delete_op_intent(
+                        self._oid_2, oi_ref.ovn
+                    )
+                    self.record_query(dq)
+                except QueryError as e:
+                    self.record_queries(e.queries)
                     check.record_failed(
                         f"Could not delete operational intent using second credentials",
-                        details=f"DSS responded with {dq.response.status_code} to attempt to delete OI {self._oid_2}",
+                        details=f"DSS responded with {dq.response.status_code} to attempt to delete OI {self._oid_2}; {e}",
                         query_timestamps=[dq.request.timestamp],
                     )
 
@@ -208,16 +215,20 @@ def _attempt_to_delete_remaining_op_intents(self):
         for op_intent in op_intents_1:
             # We look for an op_intent where the uss_qualifier is the manager;
             if op_intent.manager == self._dss.client.auth_adapter.get_sub():
-                (_, _, dq) = self._dss.delete_op_intent(op_intent.id, op_intent.ovn)
-                self.record_query(dq)
                 with self.check(
                     "Operational intent references can be deleted by their owner",
                     self._pid,
                 ) as check:
-                    if dq.response.status_code != 200:
+                    try:
+                        (_, _, dq) = self._dss.delete_op_intent(
+                            op_intent.id, op_intent.ovn
+                        )
+                        self.record_query(dq)
+                    except QueryError as e:
+                        self.record_queries(e.queries)
                         check.record_failed(
                             f"Could not delete operational intent using main credentials",
-                            details=f"DSS responded with {dq.response.status_code} to attempt to delete OI {op_intent.id}",
+                            details=f"DSS responded with {dq.response.status_code} to attempt to delete OI {op_intent.id}; {e}",
                             query_timestamps=[dq.request.timestamp],
                         )
 
@@ -242,18 +253,20 @@ def _attempt_to_delete_remaining_op_intents(self):
                 op_intent.manager
                 == self._dss_separate_creds.client.auth_adapter.get_sub()
             ):
-                (_, _, dq) = self._dss_separate_creds.delete_op_intent(
-                    op_intent.id, op_intent.ovn
-                )
-                self.record_query(dq)
                 with self.check(
                     "Operational intent references can be deleted by their owner",
                     self._pid,
                 ) as check:
-                    if dq.response.status_code != 200:
+                    try:
+                        (_, _, dq) = self._dss_separate_creds.delete_op_intent(
+                            op_intent.id, op_intent.ovn
+                        )
+                        self.record_query(dq)
+                    except QueryError as e:
+                        self.record_queries(e.queries)
                         check.record_failed(
                             f"Could not delete operational intent using second credentials",
-                            details=f"DSS responded with {dq.response.status_code} to attempt to delete OI {op_intent.id}",
+                            details=f"DSS responded with {dq.response.status_code} to attempt to delete OI {op_intent.id}; {e}",
                             query_timestamps=[dq.request.timestamp],
                         )
 
@@ -404,20 +417,29 @@ def _check_mutation_on_non_owned_intent_fails(self):
                 )
 
         # Try to delete
-        (_, _, dq) = self._dss_separate_creds.delete_op_intent(
-            self._oid_1, self._current_ref_1.ovn
-        )
-        self.record_query(dq)
         with self.check(
             "Non-owning credentials cannot delete operational intent",
             self._pid,
         ) as check:
-            if dq.response.status_code != 403:
+            try:
+                (_, _, dq) = self._dss_separate_creds.delete_op_intent(
+                    self._oid_1, self._current_ref_1.ovn
+                )
+                self.record_query(dq)
                 check.record_failed(
                     f"Could delete operational intent using second credentials",
                     details=f"DSS responded with {dq.response.status_code} to attempt to delete OI {self._oid_1}",
                     query_timestamps=[dq.request.timestamp],
                 )
+            except QueryError as e:
+                self.record_queries(e.queries)
+                dq = e.queries[0]
+                if dq.response.status_code != 403:
+                    check.record_failed(
+                        "DSS did not fail with expected status code 403",
+                        details=f"DSS responded with {dq.response.status_code} to attempt to delete OI {self._oid_1}; {e}",
+                        query_timestamps=[dq.request.timestamp],
+                    )
 
         # Query again to confirm that the op intent has not been modified in any way:
         (op_1_current, qcheck) = self._dss.get_op_intent_reference(self._oid_1)

monitoring/uss_qualifier/scenarios/scenario.py

@@ -377,6 +377,10 @@ def _begin_test_step(self, step: TestStepDocumentation) -> None:
         self._case_report.steps.append(self._step_report)
         self._phase = ScenarioPhase.RunningTestStep
 
+    def record_queries(self, queries: List[fetch.Query]) -> None:
+        for q in queries:
+            self.record_query(q)
+
     def record_query(self, query: fetch.Query) -> None:
         self._expect_phase({ScenarioPhase.RunningTestStep, ScenarioPhase.CleaningUp})
         if "queries" not in self._step_report: