[uss_qualifier/utm] Add make_report test step fragment

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

@@ -30,6 +30,8 @@
     OperationID,
     GetOperationalIntentTelemetryResponse,
     VehicleTelemetry,
+    ExchangeRecord,
+    ErrorReport,
 )
 from uas_standards.astm.f3548.v21.constants import Scope
 
@@ -104,6 +106,16 @@ def _uses_scope(self, *scopes: Tuple[str]) -> None:
                     f"{fullname(type(self))} client called {calling_function_name(1)} which requires the use of the scope `{scope}`, but this DSSInstance is only authorized to perform actions with the scopes {' or '.join(self._scopes_authorized)}"
                 )
 
+    def _uses_any_scope(self, *scopes: str) -> str:
+        """Validates that at least a required scope is authorized for a request.
+        Additionally, returns a valid scope that may be used for the request."""
+        for scope in scopes:
+            if scope in self._scopes_authorized:
+                return scope
+        raise ValueError(
+            f"{fullname(type(self))} client called {calling_function_name(1)} which requires the use of any of the scopes `{', '.join(scopes)}`, but this DSSInstance is only authorized to perform actions with the scopes {' or '.join(self._scopes_authorized)}"
+        )
+
     def can_use_scope(self, scope: str) -> bool:
         return scope in self._scopes_authorized
 
@@ -369,6 +381,58 @@ def set_uss_availability(
             result = query.parse_json_result(UssAvailabilityStatusResponse)
             return result.version, query
 
+    def make_report(
+        self,
+        exchange: ExchangeRecord,
+    ) -> Tuple[Optional[str], Query]:
+        """
+        Make a DSS report.
+        Returns:
+            A tuple composed of
+            1) the report ID;
+            2) the query.
+        Raises:
+            * QueryError: if request failed, if HTTP status code is different than 201, or if the parsing of the response failed.
+        """
+        use_scope = self._uses_any_scope(
+            Scope.ConstraintManagement,
+            Scope.ConstraintProcessing,
+            Scope.StrategicCoordination,
+            Scope.ConformanceMonitoringForSituationalAwareness,
+            Scope.AvailabilityArbitration,
+        )
+
+        req = ErrorReport(exchange=exchange)
+        op = OPERATIONS[OperationID.MakeDssReport]
+        query = query_and_describe(
+            self.client,
+            op.verb,
+            op.path,
+            QueryType.F3548v21DSSMakeDssReport,
+            self.participant_id,
+            scope=use_scope,
+            json=req,
+        )
+
+        # TODO: this is a temporary hack: the endpoint is currently not implemented in the DSS, as such we expect the
+        #  DSS to respond with a 400 and a specific error message. This must be updated once this endpoint is actually
+        #  implemented in the DSS.
+        # if query.status_code != 201:
+        #     raise QueryError(
+        #         f"Received code {query.status_code} when attempting to make DSS report{f'; error message: `{query.error_message}`' if query.error_message is not None else ''}",
+        #         query,
+        #     )
+        # else:
+        #     result = query.parse_json_result(ErrorReport)
+        #     return result.report_id, query
+        if query.status_code != 400 or "Not yet implemented" not in query.error_message:
+            raise QueryError(
+                f"Received code {query.status_code} when attempting to make DSS report{f'; error message: `{query.error_message}`' if query.error_message is not None else ''}",
+                query,
+            )
+        else:
+            return "dummy_report_id", query
+
     def query_subscriptions(
         self,
         volume: Volume4D,

monitoring/uss_qualifier/scenarios/astm/utm/make_dss_report.md

@@ -0,0 +1,10 @@
+# Make report to DSS test step fragment
+This step makes a report to the DSS.
+
+See `make_dss_report` in [test_steps.py](test_steps.py).
+
+## 🛑 DSS report successfully submitted check
+If the submission of the report to the DSS does not succeed, this check will fail per **[astm.f3548.v21.DSS0100,2](../../../requirements/astm/f3548/v21.md)**.
+
+## ⚠️ DSS returned a valid report ID check
+If the ID returned by the DSS is not present or is empty, this check will fail per **[astm.f3548.v21.DSS0100,2](../../../requirements/astm/f3548/v21.md)**.

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

@@ -11,6 +11,7 @@
     OperationalIntentReference,
     GetOperationalIntentDetailsResponse,
     EntityID,
+    ExchangeRecord,
 )
 from uas_standards.astm.f3548.v21.constants import Scope
 from monitoring.monitorlib.clients.flight_planning.flight_info import (
@@ -698,3 +699,42 @@ def set_uss_down(
                 query_timestamps=[avail_query.request.timestamp],
             )
     return availability_version
+
+
+def make_report(
+    scenario: TestScenarioType,
+    dss: DSSInstance,
+    exchange: ExchangeRecord,
+) -> str:
+    """Make a DSS report.
+
+    This function implements the test step fragment described in make_dss_report.md.
+
+    Returns:
+        The report ID.
+    """
+    with scenario.check(
+        "DSS report successfully submitted", [dss.participant_id]
+    ) as check:
+        try:
+            report_id, report_query = dss.make_report(exchange)
+            scenario.record_query(report_query)
+        except QueryError as e:
+            scenario.record_queries(e.queries)
+            report_query = e.cause
+            check.record_failed(
+                summary="DSS report could not be submitted",
+                details=f"DSS responded code {report_query.status_code}; {e}",
+                query_timestamps=[report_query.request.timestamp],
+            )
+
+    with scenario.check(
+        "DSS returned a valid report ID", [dss.participant_id]
+    ) as check:
+        if not report_id:
+            check.record_failed(
+                summary="Submitted DSS report returned no or empty ID",
+                details=f"DSS responded code {report_query.status_code} but with no ID for the report",
+                query_timestamps=[report_query.request.timestamp],
+            )
+    return report_id