[uss_qualifier] rid net0470 - nominalbehavior: check SP issues notifi…

…cations to subscribers

monitoring/prober/infrastructure.py

@@ -100,7 +100,7 @@ def wrapper_default_scope(*args, **kwargs):
 resource_type_code_descriptions: Dict[ResourceType, str] = {}
 
 
-# Next code: 399
+# Next code: 400
 def register_resource_type(code: int, description: str) -> ResourceType:
     """Register that the specified code refers to the described resource.
 

monitoring/uss_qualifier/configurations/dev/library/environment_containers.yaml

@@ -226,6 +226,22 @@ mock_uss_instance_uss1:
     participant_id: mock_uss
     mock_uss_base_url: http://scdsc.uss1.localutm
 
+mock_uss_instance_dp_v19:
+  resource_type: resources.interuss.mock_uss.client.MockUSSResource
+  dependencies:
+      auth_adapter: utm_auth
+  specification:
+      participant_id: mock_uss
+      mock_uss_base_url: http://v19.riddp.uss3.localutm
+
+mock_uss_instance_dp_v22a:
+  resource_type: resources.interuss.mock_uss.client.MockUSSResource
+  dependencies:
+      auth_adapter: utm_auth
+  specification:
+      participant_id: mock_uss
+      mock_uss_base_url: http://v22a.riddp.uss1.localutm
+
 mock_uss_instance_uss6:
   resource_type: resources.interuss.mock_uss.client.MockUSSResource
   dependencies:

monitoring/uss_qualifier/configurations/dev/netrid_v19.yaml

@@ -15,6 +15,8 @@ v1:
         netrid_observers_v19: {$ref: 'library/environment.yaml#/netrid_observers_v19'}
         netrid_dss_instances_v19: {$ref: 'library/environment.yaml#/netrid_dss_instances_v19'}
 
+        mock_uss_instance_dp_v19: {$ref: 'library/environment.yaml#/mock_uss_instance_dp_v19'}
+
         test_exclusions: { $ref: 'library/resources.yaml#/test_exclusions' }
     non_baseline_inputs:
       - v1.test_run.resources.resource_declarations.utm_auth
@@ -28,6 +30,7 @@ v1:
           flights_data: kentland_flights_data
           service_providers: netrid_service_providers_v19
           observers: netrid_observers_v19
+          mock_uss: mock_uss_instance_dp_v19
           evaluation_configuration: netrid_observation_evaluation_configuration
           dss_instances: netrid_dss_instances_v19
           utm_client_identity: utm_client_identity

monitoring/uss_qualifier/configurations/dev/netrid_v22a.yaml

@@ -15,6 +15,8 @@ v1:
         netrid_observers_v22a: {$ref: 'library/environment.yaml#/netrid_observers_v22a'}
         netrid_dss_instances_v22a: {$ref: 'library/environment.yaml#/netrid_dss_instances_v22a'}
 
+        mock_uss_instance_dp_v22a: {$ref: 'library/environment.yaml#/mock_uss_instance_dp_v22a'}
+
         test_exclusions: { $ref: 'library/resources.yaml#/test_exclusions' }
     non_baseline_inputs:
       - v1.test_run.resources.resource_declarations.utm_auth
@@ -28,6 +30,7 @@ v1:
           flights_data: kentland_flights_data
           service_providers: netrid_service_providers_v22a
           observers: netrid_observers_v22a
+          mock_uss: mock_uss_instance_dp_v22a
           evaluation_configuration: netrid_observation_evaluation_configuration
           dss_instances: netrid_dss_instances_v22a
           utm_client_identity: utm_client_identity

monitoring/uss_qualifier/configurations/dev/uspace.yaml

@@ -20,6 +20,7 @@ v1:
         second_utm_auth: {$ref: 'library/environment.yaml#/second_utm_auth'}
         mock_uss_instances_scdsc: {$ref: 'library/environment.yaml#/mock_uss_instances_scdsc'}
         mock_uss_instance_uss6: {$ref: 'library/environment.yaml#/mock_uss_instance_uss6'}
+        mock_uss_instance_dp: {$ref: 'library/environment.yaml#/mock_uss_instance_dp_v22a'}
         test_env_version_providers: {$ref: 'library/environment.yaml#/scd_version_providers'}
         prod_env_version_providers: {$ref: 'library/environment.yaml#/scd_version_providers'}
         all_flight_planners: {$ref: 'library/environment.yaml#/all_flight_planners'}
@@ -59,6 +60,7 @@ v1:
           non_conflicting_flights: che_non_conflicting_flights
           flight_planners: all_flight_planners?
           mock_uss: mock_uss_instance_uss6
+          mock_uss_dp: mock_uss_instance_dp
           scd_dss: scd_dss
           scd_dss_instances: scd_dss_instances
 
@@ -92,6 +94,7 @@ v1:
                 non_conflicting_flights: non_conflicting_flights
                 flight_planners: flight_planners?
                 mock_uss: mock_uss
+                mock_uss_dp: mock_uss_dp
                 scd_dss: scd_dss
                 scd_dss_instances: scd_dss_instances
 

monitoring/uss_qualifier/scenarios/astm/netrid/common/nominal_behavior.py

@@ -1,11 +1,24 @@
-from typing import List, Optional
+from datetime import timedelta
+from typing import List, Optional, Type
 
+from future.backports.datetime import datetime
+from implicitdict import ImplicitDict
+from loguru import logger
 from requests.exceptions import RequestException
 from s2sphere import LatLngRect
+from uas_standards.astm.f3411.v19 import api as api_v19
+from uas_standards.astm.f3411.v22a import api as api_v22a
 
 from monitoring.monitorlib.errors import stacktrace_string
 from monitoring.monitorlib.rid import RIDVersion
+from monitoring.monitorlib.temporal import Time
+from monitoring.prober.infrastructure import register_resource_type
 from monitoring.uss_qualifier.resources.astm.f3411.dss import DSSInstancesResource
+from monitoring.uss_qualifier.resources.interuss import IDGeneratorResource
+from monitoring.uss_qualifier.resources.interuss.mock_uss.client import (
+    MockUSSResource,
+    MockUSSClient,
+)
 from monitoring.uss_qualifier.resources.netrid import (
     FlightDataResource,
     NetRIDServiceProviders,
@@ -16,6 +29,7 @@
     display_data_evaluator,
     injection,
 )
+from monitoring.uss_qualifier.scenarios.astm.netrid.dss_wrapper import DSSWrapper
 from monitoring.uss_qualifier.scenarios.astm.netrid.injected_flight_collection import (
     InjectedFlightCollection,
 )
@@ -31,30 +45,50 @@
 
 
 class NominalBehavior(GenericTestScenario):
+    SUB_TYPE = register_resource_type(399, "Subscription")
+
     _flights_data: FlightDataResource
     _service_providers: NetRIDServiceProviders
     _observers: NetRIDObserversResource
+    _mock_uss: MockUSSClient
     _evaluation_configuration: EvaluationConfigurationResource
 
     _injected_flights: List[InjectedFlight]
     _injected_tests: List[InjectedTest]
 
+    _dss_wrapper: DSSWrapper
+    _subscription_id: str
+
     def __init__(
         self,
         flights_data: FlightDataResource,
         service_providers: NetRIDServiceProviders,
         observers: NetRIDObserversResource,
+        mock_uss: Optional[MockUSSResource],
         evaluation_configuration: EvaluationConfigurationResource,
+        id_generator: IDGeneratorResource,
         dss_pool: Optional[DSSInstancesResource] = None,
     ):
         super().__init__()
         self._flights_data = flights_data
         self._service_providers = service_providers
         self._observers = observers
+        self._mock_uss = mock_uss.mock_uss
         self._evaluation_configuration = evaluation_configuration
         self._dss_pool = dss_pool
+        if dss_pool:
+            self._dss_wrapper = DSSWrapper(self, dss_pool.dss_instances[0])
+        else:
+            self._dss_wrapper = None
+
         self._injected_tests = []
 
+        self._subscription_id = id_generator.id_factory.make_id(self.SUB_TYPE)
+
+        # For running the subscription checks we need both a wrapped dss instance
+        # and a mock USS
+        self._check_subscription = self._mock_uss and self._dss_wrapper
+
     @property
     def _rid_version(self) -> RIDVersion:
         raise NotImplementedError(
@@ -63,17 +97,74 @@ def _rid_version(self) -> RIDVersion:
 
     def run(self, context: ExecutionContext):
         self.begin_test_scenario(context)
+
+        self.begin_test_case("Setup")
+
+        if not self._check_subscription:
+            self.record_note(
+                "notification_testing",
+                "Mock USS not available or DSS instance not available, will skip checks related to notifications",
+            )
+
+        self.begin_test_step("Clean workspace")
+        # Test flights are being taken care of by preparation step before this scenario
+        self._dss_wrapper.cleanup_sub(self._subscription_id)
+
+        self.end_test_step()
+        self.end_test_case()
+
         self.begin_test_case("Nominal flight")
 
+        if self._check_subscription:
+            self.begin_test_step("Mock USS Subscription")
+            self._subscribe_mock_uss()
+            self.end_test_step()
+
         self.begin_test_step("Injection")
         self._inject_flights()
         self.end_test_step()
 
         self._poll_during_flights()
 
+        if self._check_subscription:
+            self.begin_test_step("Validate Mock USS received notification")
+            self._validate_mock_uss_notifications(context.start_time)
+            self.end_test_step()
+
         self.end_test_case()
         self.end_test_scenario()
 
+    def _subscribe_mock_uss(self):
+        dss_wrapper = DSSWrapper(self, self._dss_pool.dss_instances[0])
+        # Get all bounding rects for flights
+        flight_rects = [f.get_rect() for f in self._flights_data.get_test_flights()]
+        flight_union: Optional[LatLngRect] = None
+        for fr in flight_rects:
+            if flight_union is None:
+                flight_union = fr
+            else:
+                flight_union = flight_union.union(fr)
+        with self.check(
+            "Subscription creation succeeds", dss_wrapper.participant_id
+        ) as check:
+            cs = dss_wrapper.put_sub(
+                check,
+                [flight_union.get_vertex(k) for k in range(4)],
+                0,
+                3000,
+                datetime.now(),
+                datetime.now() + timedelta(hours=1),
+                self._mock_uss.base_url + "/mock/riddp",
+                self._subscription_id,
+                None,
+            )
+            if not cs.success:
+                check.record_failed(
+                    summary="Error while creating a Subscription for the Mock USS on the DSS",
+                    details=f"Error message: {cs.errors}",
+                )
+                return
+
     def _inject_flights(self):
         (self._injected_flights, self._injected_tests) = injection.inject_flights(
             self, self._flights_data, self._service_providers
@@ -110,8 +201,98 @@ def poll_fct(rect: LatLngRect) -> bool:
             poll_fct,
         )
 
+    def _validate_mock_uss_notifications(self, scenario_start_time: datetime):
+        interactions, q = self._mock_uss.get_interactions(Time(scenario_start_time))
+        if q.status_code != 200:
+            logger.error(
+                f"Failed to get interactions from mock uss: HTTP {q.status_code} - {q.response.json}"
+            )
+            self.record_note(
+                "mock_uss_interactions",
+                f"failed to obtain interactions with http status {q.status_code}",
+            )
+            return
+
+        logger.debug(
+            f"Received {len(interactions)} interactions from mock uss:\n{interactions}"
+        )
+
+        # For each of the service providers we injected flights in,
+        # we're looking for an inbound notification for the mock_uss's subscription:
+        for test_flight in self._injected_flights:
+            notification_reception_times = []
+            with self.check(
+                "Service Provider issued a notification", test_flight.uss_participant_id
+            ) as check:
+                notif_param_type = self._notif_param_type()
+                sub_notif_interactions: List[(datetime, notif_param_type)] = [
+                    (
+                        i.query.request.received_at.datetime,
+                        ImplicitDict.parse(i.query.request.json, notif_param_type),
+                    )
+                    for i in interactions
+                    if i.query.request.method == "POST"
+                    and i.direction == "Incoming"
+                    and "/uss/identification_service_areas/" in i.query.request.url
+                ]
+                for (received_at, notification) in sub_notif_interactions:
+                    for sub in notification.subscriptions:
+                        if (
+                            sub.subscription_id == self._subscription_id
+                            and notification.service_area.owner
+                            == test_flight.uss_participant_id
+                        ):
+                            notification_reception_times.append(received_at)
+
+                if len(notification_reception_times) == 0:
+                    check.record_failed(
+                        summary="No notification received",
+                        details=f"No notification received from {test_flight.uss_participant_id} for subscription {self._subscription_id} about flight {test_flight.test_id} that happened within the subscription's boundaries.",
+                    )
+                    continue
+
+            # The performance requirements define 95th and 99th percentiles for the SP to respect,
+            # which we can't strictly check with one (or very few) samples.
+            # Furthermore, we use the time of injection as the 'starting point', which is necessarily before the SP
+            # actually becomes aware of the subscription (when the ISA is created at the DSS)
+            # the p95 to respect is 1 second, the p99 is 3 seconds.
+            # As an approximation, we check that the single sample (or the average of the few) is below the p99.
+            notif_latencies = [
+                l - test_flight.query_timestamp for l in notification_reception_times
+            ]
+            avg_latency = (
+                sum(notif_latencies, timedelta(0)) / len(notif_latencies)
+                if notif_latencies
+                else None
+            )
+            with self.check(
+                "Service Provider notification was received within delay",
+                test_flight.uss_participant_id,
+            ) as check:
+                if avg_latency.seconds > self._rid_version.dp_data_resp_percentile99_s:
+                    check.record_failed(
+                        summary="Notification received too late",
+                        details=f"Notification(s) received {avg_latency} after the flight ended, which is more than the allowed 99th percentile of {self._rid_version.dp_data_resp_percentile99_s} seconds.",
+                    )
+
+    def _notif_param_type(
+        self,
+    ) -> Type[
+        api_v19.PutIdentificationServiceAreaNotificationParameters
+        | api_v22a.PutIdentificationServiceAreaNotificationParameters
+    ]:
+        if self._rid_version == RIDVersion.f3411_19:
+            return api_v19.PutIdentificationServiceAreaNotificationParameters
+        elif self._rid_version == RIDVersion.f3411_22a:
+            return api_v22a.PutIdentificationServiceAreaNotificationParameters
+        else:
+            raise ValueError(f"Unsupported RID version: {self._rid_version}")
+
     def cleanup(self):
         self.begin_cleanup()
+
+        self._dss_wrapper.cleanup_sub(self._subscription_id)
+
         while self._injected_tests:
             injected_test = self._injected_tests.pop()
             matching_sps = [