[uss_qualifier] Receive notifications test case implementation (#529)

* receive notifications impl

* Test case implementation for receive notification

* Documentation format

* Revert "Documentation format"

This reverts commit 7fe09a8.

* Formatted documentation

* More documentation format fix

* Removing priority check for intents

* Fix documentation format

* Fix doc format

monitoring/uss_qualifier/scenarios/astm/utm/subscription_notifications/__init__.py

@@ -0,0 +1 @@
+from .receive_notifications_for_awareness import ReceiveNotificationsForAwareness

monitoring/uss_qualifier/scenarios/astm/utm/subscription_notifications/receive_notifications_for_awareness/__init__.py

@@ -0,0 +1 @@
+from .receive_notifications_for_awareness import ReceiveNotificationsForAwareness

monitoring/uss_qualifier/scenarios/astm/utm/subscription_notifications/receive_notifications_for_awareness/receive_notifications_for_awareness.py

@@ -0,0 +1,217 @@
+from typing import Optional, Dict
+
+from monitoring.monitorlib.clients.flight_planning.flight_info import (
+    AirspaceUsageState,
+    UasState,
+)
+from monitoring.monitorlib.geotemporal import Volume4DCollection
+from monitoring.monitorlib.temporal import TimeDuringTest
+from monitoring.uss_qualifier.resources.flight_planning.flight_intent import (
+    FlightIntent,
+)
+from monitoring.uss_qualifier.resources.flight_planning.flight_intent_validation import (
+    ExpectedFlightIntent,
+    validate_flight_intent_templates,
+)
+from monitoring.uss_qualifier.suites.suite import ExecutionContext
+from monitoring.uss_qualifier.scenarios.scenario import (
+    TestScenario,
+)
+from monitoring.monitorlib.clients.flight_planning.flight_info_template import (
+    FlightInfoTemplate,
+)
+from monitoring.monitorlib.clients.flight_planning.client import FlightPlannerClient
+from monitoring.uss_qualifier.resources.flight_planning.flight_planners import (
+    FlightPlannerResource,
+)
+from monitoring.uss_qualifier.resources.interuss.mock_uss.client import (
+    MockUSSClient,
+    MockUSSResource,
+)
+from monitoring.uss_qualifier.resources.astm.f3548.v21.dss import DSSInstance
+from monitoring.uss_qualifier.resources.astm.f3548.v21 import DSSInstanceResource
+from monitoring.uss_qualifier.resources.flight_planning import (
+    FlightIntentsResource,
+)
+from uas_standards.astm.f3548.v21.constants import Scope
+import arrow
+from monitoring.monitorlib.temporal import Time
+from monitoring.uss_qualifier.scenarios.flight_planning.test_steps import (
+    cleanup_flights_fp_client,
+    plan_flight,
+    activate_flight,
+)
+from monitoring.uss_qualifier.scenarios.astm.utm.test_steps import (
+    OpIntentValidator,
+)
+from monitoring.uss_qualifier.scenarios.astm.utm.subscription_notifications.test_steps.validate_notification_received import (
+    expect_tested_uss_receives_notification_from_mock_uss,
+)
+
+
+class ReceiveNotificationsForAwareness(TestScenario):
+    flight_1_planned: FlightInfoTemplate
+    flight_2_planned: FlightInfoTemplate
+    flight_1_activated: FlightInfoTemplate
+
+    tested_uss_client: FlightPlannerClient
+    mock_uss: MockUSSClient
+    mock_uss_client: FlightPlannerClient
+    dss: DSSInstance
+
+    def __init__(
+        self,
+        tested_uss: FlightPlannerResource,
+        mock_uss: MockUSSResource,
+        dss: DSSInstanceResource,
+        flight_intents: Optional[FlightIntentsResource] = None,
+    ):
+        super().__init__()
+        self.tested_uss_client = tested_uss.client
+        self.mock_uss = mock_uss.mock_uss
+        self.mock_uss_client = mock_uss.mock_uss.flight_planner
+        self.dss = dss.get_instance(
+            {
+                Scope.StrategicCoordination: "search for operational intent references to verify outcomes of planning activities"
+            }
+        )
+
+        expected_flight_intents = [
+            ExpectedFlightIntent(
+                "flight_1_planned",
+                "Flight 1",
+                must_not_conflict_with=["Flight 2"],
+                usage_state=AirspaceUsageState.Planned,
+                uas_state=UasState.Nominal,
+            ),
+            ExpectedFlightIntent(
+                "flight_1_activated",
+                "Flight 1",
+                must_not_conflict_with=["Flight 2"],
+                usage_state=AirspaceUsageState.InUse,
+                uas_state=UasState.Nominal,
+            ),
+            ExpectedFlightIntent(
+                "flight_2_planned",
+                "Flight 2",
+                must_not_conflict_with=["Flight 1"],
+                usage_state=AirspaceUsageState.Planned,
+                uas_state=UasState.Nominal,
+            ),
+        ]
+
+        templates = flight_intents.get_flight_intents()
+        try:
+            validate_flight_intent_templates(templates, expected_flight_intents)
+        except ValueError as e:
+            raise ValueError(
+                f"`{self.me()}` TestScenario requirements for flight_intents not met: {e}"
+            )
+        extents = []
+
+        for efi in expected_flight_intents:
+            intent = FlightIntent.from_flight_info_template(templates[efi.intent_id])
+            extents.extend(intent.request.operational_intent.volumes)
+            extents.extend(intent.request.operational_intent.off_nominal_volumes)
+            setattr(self, efi.intent_id, templates[efi.intent_id])
+
+        self._intents_extent = Volume4DCollection.from_interuss_scd_api(
+            extents
+        ).bounding_volume.to_f3548v21()
+
+    def run(self, context: ExecutionContext):
+        times = {
+            TimeDuringTest.StartOfTestRun: Time(context.start_time),
+            TimeDuringTest.StartOfScenario: Time(arrow.utcnow().datetime),
+        }
+        self.begin_test_scenario(context)
+        self.record_note(
+            "Tested USS",
+            f"{self.tested_uss_client.participant_id}",
+        )
+        self.record_note(
+            "Mock USS",
+            f"{self.mock_uss_client.participant_id}",
+        )
+        self.begin_test_case(
+            "Activated operational intent receives notification of relevant intent"
+        )
+        self._receive_notification_successfully_when_activated_test_case(times)
+        self.end_test_case()
+
+        self.end_test_scenario()
+
+    def _receive_notification_successfully_when_activated_test_case(
+        self, times: Dict[TimeDuringTest, Time]
+    ):
+        times[TimeDuringTest.TimeOfEvaluation] = Time(arrow.utcnow().datetime)
+        flight_1_planned = self.flight_1_planned.resolve(times)
+        flight_1_activated = self.flight_1_activated.resolve(times)
+        flight_2_planned = self.flight_2_planned.resolve(times)
+
+        self.begin_test_step("Tested_uss plans and activates Flight 1")
+        with OpIntentValidator(
+            self,
+            self.tested_uss_client,
+            self.dss,
+            self._intents_extent,
+        ) as validator:
+            _, self.flight_1_id = plan_flight(
+                self,
+                self.tested_uss_client,
+                flight_1_planned,
+            )
+            flight_1_oi_ref = validator.expect_shared(flight_1_planned)
+
+        with OpIntentValidator(
+            self,
+            self.tested_uss_client,
+            self.dss,
+            self._intents_extent,
+            flight_1_oi_ref,
+        ) as validator:
+            _, self.flight_1_id = activate_flight(
+                self,
+                self.tested_uss_client,
+                flight_1_activated,
+                self.flight_1_id,
+            )
+            flight_1_oi_ref = validator.expect_shared(flight_1_activated)
+
+        self.end_test_step()
+        subscription_id = flight_1_oi_ref.subscription_id
+
+        self.begin_test_step("Mock_uss plans Flight 2")
+        with OpIntentValidator(
+            self,
+            self.mock_uss_client,
+            self.dss,
+            self._intents_extent,
+        ) as validator:
+            flight_2_planning_time = Time(arrow.utcnow().datetime)
+            _, self.flight_2_id = plan_flight(
+                self,
+                self.mock_uss_client,
+                flight_2_planned,
+            )
+            flight_2_oi_ref = validator.expect_shared(flight_2_planned)
+
+        self.end_test_step()
+
+        self.begin_test_step("Validate Flight 2 notification received by tested_uss")
+        expect_tested_uss_receives_notification_from_mock_uss(
+            self,
+            self.mock_uss,
+            flight_2_planning_time,
+            flight_2_oi_ref.id,
+            subscription_id,
+            flight_1_oi_ref.uss_base_url,
+            self.tested_uss_client.participant_id,
+            flight_2_planning_time,
+        )
+        self.end_test_step()
+
+    def cleanup(self):
+        self.begin_cleanup()
+        cleanup_flights_fp_client(self, (self.mock_uss_client, self.tested_uss_client))
+        self.end_cleanup()

monitoring/uss_qualifier/scenarios/astm/utm/subscription_notifications/test_steps/validate_notification_received.md

@@ -23,16 +23,16 @@ As per **[astm.f3548.v21.SCD0080](../../../../../requirements/astm/f3548/v21.md)
 operational intents relevant to their own ones when they are in the Activated, NonConforming or Contingent states.
 In DSS, there is a subscription associated with an operational intent managed by a USS. A tested USS should successfully
 receive a notification of relevant intent from Mock USS based on this subscription.
-The check will be done if valid notification is sent by Mock USS, which is determined in in
- **[Mock USS sends valid notification check](#⚠️-mock-uss-sends-valid-notification-check)** above.
+The check will be done if valid notification is sent by Mock USS, which is determined in
+[Mock USS sends valid notification check](#⚠️-mock-uss-sends-valid-notification-check) above.
 This check will fail if tested USS does not respond with http status 204 to a valid notification attempt by Mock USS.
 
 ## ⚠️ Tested USS rejects invalid notification check
 
 As per **[astm.f3548.v21.USS0105](../../../../../requirements/astm/f3548/v21.md)**, Tested USS should validate that the notification
 received includes the subscription_id associated with its managed operation.
 The check will be done if invalid notification is sent by Mock USS, which is determined in
- **[Mock USS sends valid notification check](#⚠️-mock-uss-sends-valid-notification-check)** above.
+[Mock USS sends valid notification check](#⚠️-mock-uss-sends-valid-notification-check) above.
 This check will fail if tested USS does not respond with http status 400 for an invalid notification attempt by Mock USS.
 
 

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

@@ -70,7 +70,7 @@ def expect_tested_uss_receives_notification_from_mock_uss(
         if not interactions:
             check.record_failed(
                 summary=f"No notification sent by mock_uss to tested_uss",
-                details=f"Notification to tested_uss with pre-existing relevant operational intent not sent, even though DSS instructed mock_uss to notify due to subscription.",
+                details=f"Notification not sent for intent id {op_intent_ref_id} to tested_uss url {tested_uss_base_url}, even though DSS instructed mock_uss to notify tested_uss based on subscription associated with its relevant operational intent.",
                 query_timestamps=[plan_request_time, query.request.timestamp],
             )
 
@@ -139,11 +139,11 @@ def is_applicable(interaction: Interaction) -> bool:
                         f"Parsing mock_uss notification to type PutOperationalIntentDetailsParameters failed - {e}"
                     )
             return (notification.operational_intent_id == op_intent_id) and (
-                base_url in interaction.query.request.url_hostname
+                base_url in interaction.query.request.url
             )
         return False
 
-    return is_applicable()
+    return is_applicable
 
 
 def _check_notification_sent_with_subscription_id_and_response(

monitoring/uss_qualifier/scenarios/flight_planning/test_steps.py

@@ -395,6 +395,34 @@ def plan_flight(
     )
 
 
+def activate_flight(
+    scenario: TestScenarioType,
+    flight_planner: FlightPlannerClient,
+    flight_info: FlightInfo,
+    flight_id: Optional[str] = None,
+    additional_fields: Optional[dict] = None,
+) -> Tuple[PlanningActivityResponse, Optional[str]]:
+    """Activate a flight intent that should result in success.
+
+    This function implements the test step fragment described in
+    activate_flight_intent.md.
+
+    Returns:
+      * The injection response.
+      * The ID of the injected flight if it is returned, None otherwise.
+    """
+    return submit_flight(
+        scenario=scenario,
+        success_check="Successful activation",
+        expected_results={(PlanningActivityResult.Completed, FlightPlanStatus.OkToFly)},
+        failed_checks={PlanningActivityResult.Failed: "Failure"},
+        flight_planner=flight_planner,
+        flight_info=flight_info,
+        flight_id=flight_id,
+        additional_fields=additional_fields,
+    )
+
+
 def submit_flight(
     scenario: TestScenarioType,
     success_check: str,