[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/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
@@ -95,6 +94,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,
@@ -104,12 +128,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.py

@@ -1,11 +1,13 @@
 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
+from urllib.parse import urlsplit
 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
@@ -25,7 +27,12 @@
     plan_flight_intent_expect_failed,
 )
 from monitoring.uss_qualifier.scenarios.astm.utm.test_steps import OpIntentValidator
-
+from monitoring.uss_qualifier.scenarios.astm.utm.data_exchange_validation.test_steps.expected_interactions_test_steps import (
+    expect_interuss_post_interactions,
+    expect_interuss_get_interactions,
+    expect_no_interuss_post_interactions,
+    precondition_no_post_interaction,
+)
 from monitoring.monitorlib.clients.mock_uss.mock_uss_scd_injection_api import (
     MockUssFlightBehavior,
 )
@@ -138,6 +145,8 @@ def _tested_uss_plans_deconflicted_flight_near_existing_flight(
     ):
         times[TimeDuringTest.TimeOfEvaluation] = Time(arrow.utcnow().datetime)
         flight_2 = self.flight_2.resolve(times)
+
+        planning_time = Time(arrow.utcnow().datetime)
         with OpIntentValidator(
             self,
             self.control_uss_client,
@@ -152,17 +161,18 @@ def _tested_uss_plans_deconflicted_flight_near_existing_flight(
                 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"
+        self._precondition_no_post_interaction(
+            self.control_uss,
+            planning_time,
+            "Precondition - check tested_uss has no subscription in flight 2 area",
         )
-        # ToDo - Add the test step details
-        self.end_test_step()
 
         times[TimeDuringTest.TimeOfEvaluation] = Time(arrow.utcnow().datetime)
         flight_1 = self.flight_1.resolve(times)
 
+        planning_time = Time(arrow.utcnow().datetime)
         with OpIntentValidator(
             self,
             self.tested_uss_client,
@@ -176,18 +186,26 @@ def _tested_uss_plans_deconflicted_flight_near_existing_flight(
                 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()
-
-        self.begin_test_step("Validate flight1 Notification sent to Control_uss")
-        # ToDo - Add the test step details
-        self.end_test_step()
+        control_uss_domain = self.control_uss.base_url
+        expect_interuss_get_interactions(
+            self,
+            self.control_uss,
+            planning_time,
+            control_uss_domain,
+            flight_2_oi_ref.id,
+            "Validate flight2 GET interaction",
+        )
+        expect_interuss_post_interactions(
+            self,
+            self.control_uss,
+            planning_time,
+            control_uss_domain,
+            "Validate flight1 Notification sent to Control_uss",
+        )
 
         delete_flight(
             self, "Delete tested_uss flight", self.tested_uss_client, self.flight_1_id
@@ -213,52 +231,83 @@ def _tested_uss_unable_to_plan_flight_near_invalid_shared_existing_flight(
 
         additional_fields = {"behavior": behavior}
 
-        _, self.flight_2_id = plan_flight(
+        planning_time = Time(arrow.utcnow().datetime)
+        with OpIntentValidator(
             self,
-            "Control_uss plans flight 2, sharing invalid operational intent data",
             self.control_uss_client,
-            flight_info,
-            additional_fields,
-        )
-
-        self.begin_test_step(
-            "Validate flight 2 shared operational intent with invalid data"
-        )
-        # ToDo - Add the test step details
-        self.end_test_step()
+            self.dss,
+            "Validate flight 2 shared operational intent with invalid data",
+            self._intents_extent,
+        ) as validator:
+            _, 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_invalid_data(flight_info)
 
-        self.begin_test_step(
-            "Precondition - check tested_uss has no subscription in flight 2 area"
+        self._precondition_no_post_interaction(
+            self.control_uss,
+            planning_time,
+            "Precondition - check tested_uss has no subscription in flight 2 area",
         )
-        # 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(
+        planning_time = Time(arrow.utcnow().datetime)
+        with OpIntentValidator(
             self,
-            "Tested_uss attempts to plan flight 1, expect failure",
             self.tested_uss_client,
-            flight_1,
-        )
-
-        self.begin_test_step("Validate flight 1 not shared by tested_uss")
-        # ToDo - Add the test step details
-        self.end_test_step()
-
-        self.begin_test_step("Validate flight 2 GET interaction")
-        # ToDo - Add the test step details
-        self.end_test_step()
+            self.dss,
+            "Validate flight 1 not shared by tested_uss",
+            self._intents_extent,
+        ) as validator:
+            _, 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 Notification not sent to Control_uss")
-        # ToDo - Add the test step details
-        self.end_test_step()
+        control_uss_domain = self.control_uss.base_url
+        expect_interuss_get_interactions(
+            self,
+            self.control_uss,
+            planning_time,
+            control_uss_domain,
+            flight_2_oi_ref.id,
+            "Validate flight 2 GET interaction",
+        )
+        expect_no_interuss_post_interactions(
+            self,
+            self.control_uss,
+            planning_time,
+            "Validate flight 1 Notification not sent to Control_uss",
+        )
 
         delete_flight(
             self, "Delete Control_uss flight", self.control_uss_client, self.flight_2_id
         )
 
+    def _precondition_no_post_interaction(
+        self,
+        mock_uss: MockUSSClient,
+        st: StringBasedDateTime,
+        test_step: str,
+    ):
+        self.begin_test_step(test_step)
+
+        mock_uss_sent_notification = precondition_no_post_interaction(
+            self, mock_uss, st
+        )
+        if mock_uss_sent_notification:
+            msg = f"As a precondition for the scenario tests, there should have been no post made to {posted_to_url}"
+            raise ScenarioCannotContinueError(msg)
+        self.end_test_step()
+
     def cleanup(self):
         self.begin_cleanup()
         cleanup_flights_fp_client(