DSS0210,A2-7-2,7 OIR

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

@@ -1,10 +1,19 @@
 import datetime
-from typing import List, Dict, Any, Optional, Self
+from typing import List, Optional
 
-from implicitdict import ImplicitDict, StringBasedDateTime
-from uas_standards.astm.f3548.v21.api import Volume4D
+from implicitdict import ImplicitDict
+from uas_standards.astm.f3548.v21.api import (
+    EntityOVN,
+    OperationalIntentState,
+    UssBaseURL,
+    EntityID,
+    PutOperationalIntentReferenceParameters,
+    ImplicitSubscriptionParameters,
+)
 
-from monitoring.monitorlib.geo import LatLngPoint, make_latlng_rect, Volume3D, Polygon
+from monitoring.monitorlib.geo import make_latlng_rect, Volume3D
+from monitoring.monitorlib.geotemporal import Volume4D
+from monitoring.monitorlib.temporal import Time
 from monitoring.uss_qualifier.resources.astm.f3548.v21.subscription_params import (
     SubscriptionParams,
 )
@@ -52,6 +61,46 @@ def get_new_subscription_params(
             notify_for_constraints=notify_for_constraints,
         )
 
+    def get_new_operational_intent_ref_params(
+        self,
+        key: List[EntityOVN],
+        state: OperationalIntentState,
+        uss_base_url: UssBaseURL,
+        time_start: datetime.datetime,
+        time_end: datetime.datetime,
+        subscription_id: Optional[EntityID],
+        implicit_sub_base_url: Optional[UssBaseURL] = None,
+        implicit_sub_for_constraints: Optional[bool] = None,
+    ) -> PutOperationalIntentReferenceParameters:
+        """
+        Build a PutOperationalIntentReferenceParameters object that can be used against the DSS OIR API.
+
+        The extents contained in these parameters contain a single 4DVolume, which may not be entirely realistic,
+        but is sufficient in situations where the content of the OIR is irrelevant as long as it is valid, such
+        as for testing authentication or parameter validation.
+
+        Note that this method allows building inconsistent parameters:
+        """
+        return PutOperationalIntentReferenceParameters(
+            extents=[
+                Volume4D(
+                    volume=self.volume,
+                    time_start=Time(time_start),
+                    time_end=Time(time_end),
+                ).to_f3548v21()
+            ],
+            key=key,
+            state=state,
+            uss_base_url=uss_base_url,
+            subscription_id=subscription_id,
+            new_subscription=ImplicitSubscriptionParameters(
+                uss_base_url=implicit_sub_base_url,
+                notify_for_constraints=implicit_sub_for_constraints,
+            )
+            if implicit_sub_base_url
+            else None,
+        )
+
 
 class PlanningAreaResource(Resource[PlanningAreaSpecification]):
     specification: PlanningAreaSpecification

monitoring/uss_qualifier/scenarios/astm/utm/dss/authentication/generic.py

@@ -0,0 +1,99 @@
+from uas_standards.astm.f3548.v21.constants import Scope
+
+from monitoring.monitorlib import fetch
+from monitoring.monitorlib.auth import InvalidTokenSignatureAuth
+from monitoring.monitorlib.infrastructure import UTMClientSession
+from monitoring.uss_qualifier.resources.astm.f3548.v21.dss import DSSInstance
+from monitoring.uss_qualifier.scenarios.scenario import TestScenario
+
+
+class GenericAuthValidator:
+    """
+    Utility class for common DSS authentication validation requirements.
+    """
+
+    def __init__(
+        self,
+        scenario: TestScenario,
+        dss: DSSInstance,
+    ):
+        self._scenario = scenario
+        self._authenticated_session = dss.client
+        self._invalid_token_session = UTMClientSession(
+            dss.base_url, auth_adapter=InvalidTokenSignatureAuth()
+        )
+        self._no_auth_session = UTMClientSession(dss.base_url, auth_adapter=None)
+
+    def query_no_auth(self, **query_kwargs) -> fetch.Query:
+        """Issue a query to the DSS without any credentials being passed"""
+        q = fetch.query_and_describe(client=self._no_auth_session, **query_kwargs)
+        self._scenario.record_query(q)
+        return q
+
+    def query_invalid_token(self, **query_kwargs) -> fetch.Query:
+        """
+        Issue a query to the DSS with an invalid token signature, but a valid token.
+        An appropriate scope is provided.
+        """
+        q = fetch.query_and_describe(
+            client=self._invalid_token_session,
+            scope=Scope.StrategicCoordination,
+            **query_kwargs,
+        )
+        self._scenario.record_query(q)
+        return q
+
+    def query_missing_scope(self, **query_kwargs) -> fetch.Query:
+        """
+        Issue a query to the DSS with a valid token, but omits specifying a scope.
+        """
+        q = fetch.query_and_describe(client=self._authenticated_session, **query_kwargs)
+        self._scenario.record_query(q)
+        return q
+
+    def query_wrong_scope(self, **query_kwargs) -> fetch.Query:
+        """
+        Issue a query to the DSS with a valid token, but with a scope that is not allowed
+        to perform the operation.
+        """
+        q = fetch.query_and_describe(
+            client=self._authenticated_session,
+            scope=Scope.AvailabilityArbitration,  # TODO make these configurable via the constructor?
+            **query_kwargs,
+        )
+        self._scenario.record_query(q)
+        return q
+
+    def query_valid_auth(self, **query_kwargs) -> fetch.Query:
+        """
+        Issue a query to the DSS with valid credentials.
+        """
+        q = fetch.query_and_describe(
+            client=self._authenticated_session,
+            scope=Scope.StrategicCoordination,
+            # TODO make these configurable via the constructor? 'valid' may be different for certain endpoints
+            **query_kwargs,
+        )
+        self._scenario.record_query(q)
+        return q
+
+    def verify_4xx_response(self, q: fetch.Query):
+        """Verifies that the passed query response's body is a valid ErrorResponse:
+        it is either empty or contains a single 'message' field, as per the OpenAPI spec.
+
+        Note that 409 responses to Operational Intent Reference mutations will contain more fields,
+        these are not handled here.
+        """
+        with self._scenario.check(
+            "Unauthorized requests return the proper error message body"
+        ) as check:
+            if len(q.response.json) == 0:
+                return
+            elif len(q.response.json) == 1 and "message" in q.response.json:
+                return
+            else:
+                check.record_failed(
+                    summary="Unexpected error response body",
+                    details=f"Response body for {q.request.method} query to {q.request.url} should be empty or contain a single 'message' field. Was: {q.response.json}",
+                    query_timestamps=[q.request.timestamp],
+                )