[uss_qualifier] Adding the interactions and invalid op checks steps

monitoring/monitorlib/clients/flight_planning/client.py

@@ -93,3 +93,12 @@ def clear_area(self, area: Volume4D) -> TestPreparationActivityResponse:
             * PlanningActivityError
         """
         raise NotImplementedError()
+
+    @abstractmethod
+    def get_base_url(self) -> str:
+        """
+        Get the base_url associated with this FlightPlannerClient
+        Returns:
+
+        """
+        raise NotImplementedError

monitoring/monitorlib/clients/flight_planning/client_scd.py

@@ -260,3 +260,6 @@ def clear_area(self, area: Volume4D) -> TestPreparationActivityResponse:
             errors = [f"[{resp.outcome.timestamp}]: {resp.outcome.message}"]
 
         return TestPreparationActivityResponse(errors=errors, queries=[query])
+
+    def get_base_url(self):
+        return self._session.get_prefix_url()

monitoring/monitorlib/clients/flight_planning/client_v1.py

@@ -222,3 +222,6 @@ def clear_area(self, area: Volume4D) -> TestPreparationActivityResponse:
             errors = [resp.outcome.message]
 
         return TestPreparationActivityResponse(errors=errors, queries=[query])
+
+    def get_base_url(self):
+        return self._session.get_prefix_url()

monitoring/monitorlib/fetch/__init__.py

@@ -3,6 +3,7 @@
 import os
 import traceback
 import uuid
+import jwt
 from typing import Dict, Optional, List, Union
 
 from enum import Enum
@@ -290,6 +291,15 @@ def error_message(self) -> Optional[str]:
             else None
         )
 
+    def get_client_sub(self):
+        headers = self.request.headers
+        if "Authorization" in headers:
+            token = headers.get("Authorization").split(" ")[1]
+            payload = jwt.decode(
+                token, algorithms="RS256", options={"verify_signature": False}
+            )
+            return payload["sub"]
+
 
 class QueryError(RuntimeError):
     """Error encountered when interacting with a server in the UTM ecosystem."""

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

@@ -1,9 +1,8 @@
 from __future__ import annotations
-
 import uuid
-from typing import Tuple, List, Optional
-from urllib.parse import urlparse
+from typing import Tuple, List, Dict, Optional
 
+from urllib.parse import urlparse
 from implicitdict import ImplicitDict
 
 from monitoring.monitorlib import infrastructure, fetch
@@ -125,6 +124,31 @@ def get_full_op_intent(
         op_intent_ref: OperationalIntentReference,
         uss_participant_id: Optional[str] = None,
     ) -> Tuple[OperationalIntent, fetch.Query]:
+        result, query = self.get_full_op_intent_without_validation(
+            op_intent_ref,
+            uss_participant_id,
+        )
+        if query.status_code != 200:
+            result = None
+        else:
+            result = ImplicitDict.parse(
+                query.response.json, GetOperationalIntentDetailsResponse
+            ).operational_intent
+        return result, query
+
+    def get_full_op_intent_without_validation(
+        self,
+        op_intent_ref: OperationalIntentReference,
+        uss_participant_id: Optional[str] = None,
+    ) -> Tuple[Dict, fetch.Query]:
+        """
+        GET OperationalIntent without validating, as invalid data expected for negative tests
+        Args:
+            op_intent_ref:
+            uss_participant_id:
+        Returns:
+            returns the response json when query is successful
+        """
         url = f"{op_intent_ref.uss_base_url}/uss/v1/operational_intents/{op_intent_ref.id}"
         query = fetch.query_and_describe(
             self.client,
@@ -134,12 +158,10 @@ def get_full_op_intent(
             uss_participant_id,
             scope=SCOPE_SC,
         )
-        if query.status_code != 200:
-            result = None
-        else:
-            result = ImplicitDict.parse(
-                query.response.json, GetOperationalIntentDetailsResponse
-            ).operational_intent
+        result = None
+        if query.status_code == 200:
+            result = query.response.json
+
         return result, query
 
     def put_op_intent(

monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/get_op_data_validation.md

@@ -35,12 +35,12 @@ Flight 2 should be successfully planned by the control USS.
 ### [Validate flight 2 sharing test step](../validate_shared_operational_intent.md)
 Validate that flight 2 is planned
 
-### Precondition - check tested_uss has no subscription in flight 2 area test step
+### Check for notification to tested_uss due to subscription in flight 2 area test step
 In order to run this test scenario, we need tested_uss to trigger GET operational intent request to mock_uss.
-So we need to make sure that there is no subscription of tested_uss, that would trigger notification of flight 2 to tested_uss.
-No notification pushed by control_uss to tested_uss, will ensure that tested_uss will make a GET request to control_uss for flight 2 details,
-while planning a nearby flight.
-If a notification is sent to tested_uss, the precondition for running this scenario will not be satisfied.
+But, if there is a subscription by tested_uss, that would trigger notification of flight 2 to tested_uss.
+Some USSes will not make a GET request to control_uss for flight 2 details, while planning a nearby flight,
+if they got a notification for flight2. Hence, if a USS didn't make a GET request, we will only fail it if didn't get
+a notification, or else, a precondition for the test will not be met.
 
 ### [Tested_uss plans flight 1 test step](../../../flight_planning/plan_flight_intent.md)
 The test driver attempts to plan flight 1 via the tested USS. It checks if any conflicts with flight 2
@@ -71,12 +71,12 @@ The control_uss, which is mock_uss is instructed to share invalid data with othe
 ### [Validate flight 2 shared operational intent with invalid data test step](test_steps/validate_sharing_operational_intent_but_with_invalid_interuss_data.md)
 Validate that flight 2 is shared with invalid data as a modified behavior is injected by uss_qualifier for a negative test.
 
-### Precondition - check tested_uss has no subscription in flight 2 area test step
+### Check for notification to tested_uss due to subscription in flight 2 area test step
 In order to run this test scenario, we need tested_uss to trigger GET operational intent request to mock_uss.
-So we need to make sure that there is no subscription of tested_uss, that would trigger notification of flight 2 to tested_uss.
-No notification pushed by control_uss to tested_uss, will ensure that tested_uss will make a GET request to control_uss for flight 2 details,
-while planning a nearby flight.
-If a notification is sent to tested_uss, the precondition for running this scenario will not be satisfied.
+But, if there is a subscription by tested_uss, that would trigger notification of flight 2 to tested_uss.
+Some USSes will not make a GET request to control_uss for flight 2 details, while planning a nearby flight,
+if they got a notification for flight2. Hence, if a USS didn't make a GET request, we will only fail it if didn't get
+a notification, or else, a precondition for the test will not be met.
 
 ### [Tested_uss attempts to plan flight 1, expect failure test step](test_steps/plan_flight_intent_expect_failed.md)
 The test driver attempts to plan the flight 1 via the tested_uss. It checks if any conflicts with flight 2

monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/get_op_data_validation.py

@@ -1,11 +1,12 @@
 from typing import Optional, Dict
-
 from monitoring.monitorlib.clients.flight_planning.flight_info_template import (
     FlightInfoTemplate,
 )
-from monitoring.monitorlib.temporal import TimeDuringTest, Time
+from monitoring.monitorlib.temporal import TimeDuringTest
 import arrow
+from implicitdict import StringBasedDateTime
 
+from monitoring.monitorlib.temporal import Time
 from monitoring.monitorlib.clients.flight_planning.client import FlightPlannerClient
 from monitoring.uss_qualifier.resources.astm.f3548.v21 import DSSInstanceResource
 from monitoring.uss_qualifier.resources.astm.f3548.v21.dss import DSSInstance
@@ -24,8 +25,16 @@
 from monitoring.uss_qualifier.scenarios.astm.utm.data_exchange_validation.test_steps.invalid_op_test_steps import (
     plan_flight_intent_expect_failed,
 )
-from monitoring.uss_qualifier.scenarios.astm.utm.test_steps import OpIntentValidator
-
+from monitoring.uss_qualifier.scenarios.astm.utm.test_steps import (
+    OpIntentValidator,
+    OpIntentValidationFailureType,
+)
+from monitoring.uss_qualifier.scenarios.astm.utm.data_exchange_validation.test_steps.expected_interactions_test_steps import (
+    expect_interuss_post_interactions,
+    expect_get_requests_to_mock_uss,
+    expect_no_interuss_post_interactions,
+    check_any_notification,
+)
 from monitoring.monitorlib.clients.mock_uss.mock_uss_scd_injection_api import (
     MockUssFlightBehavior,
 )
@@ -138,26 +147,32 @@ def _tested_uss_plans_deconflicted_flight_near_existing_flight(
     ):
         times[TimeDuringTest.TimeOfEvaluation] = Time(arrow.utcnow().datetime)
         flight_2 = self.flight_2.resolve(times)
+
         with OpIntentValidator(
             self,
             self.control_uss_client,
             self.dss,
             "Validate flight 2 sharing",
             self._intents_extent,
         ) as validator:
+            planning_time = Time(arrow.utcnow().datetime)
             _, self.flight_2_id = plan_flight(
                 self,
                 "Control_uss plans flight 2",
                 self.control_uss_client,
                 flight_2,
             )
 
-            validator.expect_shared(flight_2)
+            flight_2_oi_ref = validator.expect_shared(flight_2)
 
         self.begin_test_step(
-            "Precondition - check tested_uss has no subscription in flight 2 area"
+            "Check for notification to tested_uss due to subscription in flight 2 area"
+        )
+        tested_uss_notified = check_any_notification(
+            self,
+            self.control_uss,
+            planning_time,
         )
-        # ToDo - Add the test step details
         self.end_test_step()
 
         times[TimeDuringTest.TimeOfEvaluation] = Time(arrow.utcnow().datetime)
@@ -170,24 +185,44 @@ def _tested_uss_plans_deconflicted_flight_near_existing_flight(
             "Validate flight 1 sharing",
             self._intents_extent,
         ) as validator:
-            _, self.flight_1_id = plan_flight(
+            planning_time = Time(arrow.utcnow().datetime)
+            plan_res, self.flight_1_id = plan_flight(
                 self,
                 "Tested_uss plans flight 1",
                 self.tested_uss_client,
                 flight_1,
             )
-
             validator.expect_shared(
                 flight_1,
             )
 
-        self.begin_test_step("Validate flight2 GET interaction")
-        # ToDo - Add the test step details
-        self.end_test_step()
+        get_requested, already_notified = expect_get_requests_to_mock_uss(
+            self,
+            self.control_uss,
+            planning_time,
+            self.control_uss.base_url,
+            flight_2_oi_ref.id,
+            self.tested_uss_client.participant_id,
+            tested_uss_notified,
+            "Validate flight2 GET interaction",
+        )
 
-        self.begin_test_step("Validate flight1 Notification sent to Control_uss")
-        # ToDo - Add the test step details
-        self.end_test_step()
+        if get_requested == False and already_notified == True:
+            msg = (
+                f"GET request was not made by tested_uss, as it was notified of flight2, due to an existing subscription. "
+                f"See documentation of test step - Check for notification to tested_uss due to subscription in flight 2 area"
+            )
+            raise ScenarioCannotContinueError(msg)
+
+        expect_interuss_post_interactions(
+            self,
+            self.control_uss,
+            planning_time,
+            self.control_uss.base_url,
+            self.tested_uss_client.participant_id,
+            plan_res.queries[0].request.timestamp,
+            "Validate flight1 Notification sent to Control_uss",
+        )
 
         delete_flight(
             self, "Delete tested_uss flight", self.tested_uss_client, self.flight_1_id
@@ -202,58 +237,89 @@ def _tested_uss_unable_to_plan_flight_near_invalid_shared_existing_flight(
         times[TimeDuringTest.TimeOfEvaluation] = Time(arrow.utcnow().datetime)
         flight_info = self.flight_2.resolve(times)
 
+        modify_field1 = "state"
+        modify_field2 = "priority"
         # Modifying the request with invalid data
         behavior = MockUssFlightBehavior(
             modify_sharing_methods=["GET", "POST"],
             modify_fields={
-                "reference": {"state": "Flying"},
-                "details": {"priority": -1},
+                "reference": {modify_field1: "Flying"},
+                "details": {modify_field2: 1.2},
             },
         )
 
         additional_fields = {"behavior": behavior}
-
-        _, self.flight_2_id = plan_flight(
+        with OpIntentValidator(
             self,
-            "Control_uss plans flight 2, sharing invalid operational intent data",
             self.control_uss_client,
-            flight_info,
-            additional_fields,
-        )
+            self.dss,
+            "Validate flight 2 shared operational intent with invalid data",
+            self._intents_extent,
+        ) as validator:
+            planning_time = Time(arrow.utcnow().datetime)
+            _, self.flight_2_id = plan_flight(
+                self,
+                "Control_uss plans flight 2, sharing invalid operational intent data",
+                self.control_uss_client,
+                flight_info,
+                additional_fields,
+            )
+            flight_2_oi_ref = validator.expect_shared_with_specified_data(
+                flight_info,
+                validation_failure_type=OpIntentValidationFailureType.DataFormat,
+                invalid_fields=[modify_field1, modify_field2],
+            )
 
         self.begin_test_step(
-            "Validate flight 2 shared operational intent with invalid data"
+            "Check for notification to tested_uss due to subscription in flight 2 area"
         )
-        # ToDo - Add the test step details
-        self.end_test_step()
-
-        self.begin_test_step(
-            "Precondition - check tested_uss has no subscription in flight 2 area"
+        tested_uss_notified = check_any_notification(
+            self,
+            self.control_uss,
+            planning_time,
         )
-        # ToDo - Add the test step details
         self.end_test_step()
 
         times[TimeDuringTest.TimeOfEvaluation] = Time(arrow.utcnow().datetime)
         flight_1 = self.flight_1.resolve(times)
-
-        _, self.flight_1_id = plan_flight_intent_expect_failed(
+        with OpIntentValidator(
             self,
-            "Tested_uss attempts to plan flight 1, expect failure",
             self.tested_uss_client,
-            flight_1,
-        )
+            self.dss,
+            "Validate flight 1 not shared by tested_uss",
+            self._intents_extent,
+        ) as validator:
+            planning_time = Time(arrow.utcnow().datetime)
+            _, self.flight_1_id = plan_flight_intent_expect_failed(
+                self,
+                "Tested_uss attempts to plan flight 1, expect failure",
+                self.tested_uss_client,
+                flight_1,
+            )
+            validator.expect_not_shared()
 
-        self.begin_test_step("Validate flight 1 not shared by tested_uss")
-        # ToDo - Add the test step details
-        self.end_test_step()
+        get_requested, already_notified = expect_get_requests_to_mock_uss(
+            self,
+            self.control_uss,
+            planning_time,
+            self.control_uss.base_url,
+            flight_2_oi_ref.id,
+            self.tested_uss_client.participant_id,
+            tested_uss_notified,
+            "Validate flight 2 GET interaction",
+        )
 
-        self.begin_test_step("Validate flight 2 GET interaction")
-        # ToDo - Add the test step details
-        self.end_test_step()
+        if get_requested == False and already_notified == True:
+            msg = f"GET request was not made by tested_uss, as it was notified of flight2, due to an exisitng subscription."
+            raise ScenarioCannotContinueError(msg)
 
-        self.begin_test_step("Validate flight 1 Notification not sent to Control_uss")
-        # ToDo - Add the test step details
-        self.end_test_step()
+        expect_no_interuss_post_interactions(
+            self,
+            self.control_uss,
+            planning_time,
+            self.tested_uss_client.participant_id,
+            "Validate flight 1 Notification not sent to Control_uss",
+        )
 
         delete_flight(
             self, "Delete Control_uss flight", self.control_uss_client, self.flight_2_id

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

@@ -0,0 +1,7 @@
+MaxTimeToWaitForSubscriptionNotificationSeconds = 7
+"""
+This constant is used for waiting to check notifications were sent to mock_uss for an entity created
+by another USS, due to subscription in that area.
+The details of usage of this constant are in ./validate_notification_operational_intent.md
+and ./validate_no_notification_operational_intent.md
+"""