[uss_qualifier/scenarios/utm/data_exchange_validation] Factor away as test step fragment 'get_mock_uss_interactions' (#833)

Author: mickmis

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

@@ -51,7 +51,7 @@ mock_uss via a GET request.  This assertion is only valid, however, if tested_us
 intent details in a different way -- specifically, a notification due to a pre-existing subscription.  In this test
 step, we determine if tested_uss had a pre-existing subscription by:
 
-#### [checking if mock_uss sent a notification to tested_uss](test_steps/query_mock_uss_interactions.md)
+#### [checking if mock_uss sent a notification to tested_uss](../../../interuss/mock_uss/get_mock_uss_interactions.md)
 
 ### [Validate flight2 GET interaction, if no notification test step](test_steps/validate_get_operational_intent.md)
 This step is skipped if a notification to tested_uss was found in the previous step since tested_uss obtained the operational intent details of flight 2 without needing to perform a GET interaction.
@@ -99,7 +99,7 @@ mock_uss via a GET request.  This assertion is only valid, however, if tested_us
 intent details in a different way -- specifically, a notification due to a pre-existing subscription.  In this test
 step, we determine if tested_uss had a pre-existing subscription by:
 
-#### [Check if mock_uss sent a notification to tested_uss](test_steps/query_mock_uss_interactions.md)
+#### [Check if mock_uss sent a notification to tested_uss](../../../interuss/mock_uss/get_mock_uss_interactions.md)
 
 ### [Validate flight2 GET interaction, if no notification test step](test_steps/validate_get_operational_intent.md)
 This step is skipped if a notification to tested_uss was found in the previous step.

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

@@ -40,14 +40,16 @@
 from monitoring.uss_qualifier.scenarios.astm.utm.data_exchange_validation.test_steps.expected_interactions_test_steps import (
     expect_no_interuss_post_interactions,
     expect_mock_uss_receives_op_intent_notification,
-    mock_uss_interactions,
-    notif_op_intent_id_filter,
-    operation_filter,
-    direction_filter,
 )
 from monitoring.monitorlib.clients.mock_uss.mock_uss_scd_injection_api import (
     MockUssFlightBehavior,
 )
+from monitoring.uss_qualifier.scenarios.interuss.mock_uss.test_steps import (
+    get_mock_uss_interactions,
+    operation_filter,
+    direction_filter,
+    notif_op_intent_id_filter,
+)
 from monitoring.uss_qualifier.scenarios.scenario import (
     TestScenario,
     ScenarioCannotContinueError,
@@ -199,7 +201,7 @@ def _plan_successfully_test_case(self, times: Dict[TimeDuringTest, Time]):
         self.begin_test_step(
             "Check for notification to tested_uss due to subscription in flight 2 area"
         )
-        tested_uss_notifications, _ = mock_uss_interactions(
+        tested_uss_notifications, _ = get_mock_uss_interactions(
             self,
             self.mock_uss,
             flight_2_planning_time,
@@ -211,7 +213,7 @@ def _plan_successfully_test_case(self, times: Dict[TimeDuringTest, Time]):
 
         self.begin_test_step("Validate flight2 GET interaction, if no notification")
         if not tested_uss_notifications:
-            tested_uss_get_requests, query = mock_uss_interactions(
+            tested_uss_get_requests, query = get_mock_uss_interactions(
                 self,
                 self.mock_uss,
                 flight_1_planning_time,
@@ -322,7 +324,7 @@ def _plan_unsuccessfully_test_case(self, times: Dict[TimeDuringTest, Time]):
         self.begin_test_step(
             "Check for notification to tested_uss due to subscription in flight 2 area"
         )
-        tested_uss_notifications, _ = mock_uss_interactions(
+        tested_uss_notifications, _ = get_mock_uss_interactions(
             self,
             self.mock_uss,
             flight_2_planning_time,
@@ -334,7 +336,7 @@ def _plan_unsuccessfully_test_case(self, times: Dict[TimeDuringTest, Time]):
 
         self.begin_test_step("Validate flight2 GET interaction, if no notification")
         if not tested_uss_notifications:
-            tested_uss_get_requests, query = mock_uss_interactions(
+            tested_uss_get_requests, query = get_mock_uss_interactions(
                 self,
                 self.mock_uss,
                 flight_1_planning_time,

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

@@ -1,28 +1,27 @@
 from __future__ import annotations
 
-from datetime import datetime, timedelta
-import re
-from typing import Callable, Dict, List, Tuple, Optional, Set
+from datetime import datetime
+from typing import Set
 
-import arrow
 from implicitdict import StringBasedDateTime, ImplicitDict
-from uas_standards.astm.f3548.v21 import api
 from uas_standards.astm.f3548.v21.api import (
     OperationID,
     EntityID,
     PutOperationalIntentDetailsParameters,
-    OperationalIntentReference,
 )
 
-from monitoring.monitorlib.clients.mock_uss.interactions import Interaction
 from monitoring.monitorlib.clients.mock_uss.interactions import QueryDirection
 from monitoring.monitorlib.delay import sleep
-from monitoring.monitorlib.fetch import QueryError, Query
 from monitoring.uss_qualifier.resources.interuss.mock_uss.client import MockUSSClient
 from monitoring.uss_qualifier.scenarios.astm.utm.data_exchange_validation.test_steps.wait import (
     wait_in_intervals,
     MaxTimeToWaitForSubscriptionNotificationSeconds as max_wait_time,
 )
+from monitoring.uss_qualifier.scenarios.interuss.mock_uss.test_steps import (
+    get_mock_uss_interactions,
+    operation_filter,
+    direction_filter,
+)
 from monitoring.uss_qualifier.scenarios.scenario import TestScenarioType
 
 
@@ -42,7 +41,7 @@ def expect_mock_uss_receives_op_intent_notification(
     """
 
     # Check for 'notification found' will be done periodically by waiting for a duration till max_wait_time
-    found, query = wait_in_intervals(mock_uss_interactions)(
+    found, query = wait_in_intervals(get_mock_uss_interactions)(
         scenario,
         mock_uss,
         st,
@@ -77,7 +76,7 @@ def expect_no_interuss_post_interactions(
         max_wait_time,
         "we have to wait the longest it may take a USS to send a notification before we can establish that they didn't send a notification",
     )
-    interactions, query = mock_uss_interactions(
+    interactions, query = get_mock_uss_interactions(
         scenario,
         mock_uss,
         st,
@@ -112,96 +111,3 @@ def expect_no_interuss_post_interactions(
                     details=f"Notification for operational intent ID {req.operational_intent_id} triggered by subscriptions {', '.join([sub.subscription_id for sub in req.subscriptions])} with timestamp {interaction.query.request.timestamp}.",
                     query_timestamps=[query.request.timestamp],
                 )
-
-
-def mock_uss_interactions(
-    scenario: TestScenarioType,
-    mock_uss: MockUSSClient,
-    since: StringBasedDateTime,
-    *is_applicable: Callable[[Interaction], bool],
-) -> Tuple[List[Interaction], Query]:
-    """Retrieve mock_uss interactions given specific criteria."""
-
-    with scenario.check(
-        "Mock USS interactions logs retrievable", [mock_uss.participant_id]
-    ) as check:
-        try:
-            interactions, query = mock_uss.get_interactions(since)
-            scenario.record_query(query)
-        except QueryError as e:
-            for q in e.queries:
-                scenario.record_query(q)
-            check.record_failed(
-                summary=f"Error from mock_uss when attempting to get interactions since {since}",
-                details=f"{str(e)}\n\nStack trace:\n{e.stacktrace}",
-                query_timestamps=[q.request.timestamp for q in e.queries],
-            )
-
-    return filter_interactions(interactions, is_applicable), query
-
-
-def filter_interactions(
-    interactions: List[Interaction], filters: Iterable[Callable[[Interaction], bool]]
-) -> List[Interaction]:
-    return list(filter(lambda x: all(f(x) for f in filters), interactions))
-
-
-def notif_op_intent_id_filter(
-    op_intent_id: EntityID,
-) -> Callable[[Interaction], bool]:
-    """Returns an `is_applicable` function that detects whether an op intent notification refers to the specified operational intent."""
-
-    def is_applicable(interaction: Interaction) -> bool:
-        if "json" in interaction.query.request and interaction.query.request.json:
-            return (
-                interaction.query.request.json.get("operational_intent_id", None)
-                == op_intent_id
-            )
-        return False
-
-    return is_applicable
-
-
-def base_url_filter(
-    base_url: str,
-) -> Callable[[Interaction], bool]:
-    """Returns an `is_applicable` function that detects if the request in an interaction is sent to the given base url."""
-
-    def is_applicable(interaction: Interaction) -> bool:
-        return interaction.query.request.url.startswith(base_url)
-
-    return is_applicable
-
-
-def direction_filter(
-    direction: QueryDirection,
-) -> Callable[[Interaction], bool]:
-    """Returns an `is_applicable` filter that filters according to query direction."""
-
-    def is_applicable(interaction: Interaction) -> bool:
-        return interaction.direction == direction
-
-    return is_applicable
-
-
-def operation_filter(
-    op_id: OperationID,
-    **query_params: str,
-) -> Callable[[Interaction], bool]:
-    """
-    Returns an `is_applicable` filter that filters according to operation ID.
-    If the operation has query parameters, they must be provided through `**query_params`.
-
-    Raises:
-        KeyError: if query_params contains a non-existing parameter
-        IndexError: if query_params is missing a parameter
-    """
-    op = api.OPERATIONS[op_id]
-    op_path = op.path.format(**query_params)  # raises KeyError, IndexError
-
-    def is_applicable(interaction: Interaction) -> bool:
-        return interaction.query.request.method == op.verb and re.search(
-            op_path, interaction.query.request.url
-        )
-
-    return is_applicable

monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/query_mock_uss_interactions.md

@@ -2,8 +2,7 @@
 
 This step verifies that a USS makes a GET request to get the intent_details of an existing operation when needed as per ASTM F3548-21 by checking the interuss interactions of mock uss
 
-## 🛑 Mock USS interactions logs retrievable check
-**[interuss.mock_uss.hosted_instance.ExposeInterface](../../../../../requirements/interuss/mock_uss/hosted_instance.md)**.
+## [Get Mock USS interactions logs](../../../../interuss/mock_uss/get_mock_uss_interactions.md)
 
 ## 🛑 Expect GET request when no notification check
 **[astm.f3548.v21.SCD0035](../../../../../requirements/astm/f3548/v21.md)**

monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/validate_get_operational_intent.md

@@ -2,8 +2,7 @@
 
 This step verifies when a flight is not created, it is also not notified by checking the interuss interactions of mock_uss instance.
 
-## 🛑 Mock USS interactions logs retrievable check
-**[interuss.mock_uss.hosted_instance.ExposeInterface](../../../../../requirements/interuss/mock_uss/hosted_instance.md)**.
+## [Get Mock USS interactions logs](../../../../interuss/mock_uss/get_mock_uss_interactions.md)
 
 ## ℹ️ Mock USS interaction can be parsed check
 **[interuss.mock_uss.hosted_instance.ExposeInterface](../../../../../requirements/interuss/mock_uss/hosted_instance.md)**.

monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/validate_no_notification_operational_intent.md

@@ -2,8 +2,7 @@
 
 This step verifies that, when creating or modifying an operational intent, a USS sent the required notification for a relevant subscription owned by a mock_uss instance by checking the interactions of that mock_uss instance.
 
-## 🛑 Mock USS interactions logs retrievable check
-**[interuss.mock_uss.hosted_instance.ExposeInterface](../../../../../requirements/interuss/mock_uss/hosted_instance.md)**.
+## [Get Mock USS interactions logs](../../../../interuss/mock_uss/get_mock_uss_interactions.md)
 
 ## ⚠️ Expect Notification sent check
 As per **[astm.f3548.v21.SCD0085](../../../../../requirements/astm/f3548/v21.md)**, the notification should be sent by a

monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/validate_notification_operational_intent.md

@@ -3,11 +3,9 @@
 This step verifies that a Tested USS successfully received a notification about a relevant operational intent from a Mock USS instance.
 This is done by checking the interactions of that Mock USS instance.
 
-## 🛑 Mock USS interactions logs retrievable check
-**[interuss.mock_uss.hosted_instance.ExposeInterface](../../../../../requirements/interuss/mock_uss/hosted_instance.md)**.
+## [Get Mock USS interactions logs](../../../../interuss/mock_uss/get_mock_uss_interactions.md)
 Mock USS provides a GET endpoint to retrieve all the interactions that took place between Mock USS
 and other USSes after a particular time.
-If there is any error retrieving these interactions, this check will fail.
 These interactions also include the notifications sent and received by Mock USS.
 
 ## ⚠️ Mock USS sends valid notification check

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

@@ -1,25 +1,23 @@
 from implicitdict import StringBasedDateTime, ImplicitDict
-from typing import Callable, List, Tuple
-from monitoring.uss_qualifier.scenarios.scenario import TestScenarioType
-from monitoring.uss_qualifier.resources.interuss.mock_uss.client import MockUSSClient
-from datetime import datetime
-from monitoring.uss_qualifier.scenarios.astm.utm.data_exchange_validation.test_steps.expected_interactions_test_steps import (
-    mock_uss_interactions,
+from typing import List, Tuple
+
+from monitoring.uss_qualifier.scenarios.interuss.mock_uss.test_steps import (
+    get_mock_uss_interactions,
     operation_filter,
     direction_filter,
     notif_op_intent_id_filter,
     base_url_filter,
 )
+from monitoring.uss_qualifier.scenarios.scenario import TestScenarioType
+from monitoring.uss_qualifier.resources.interuss.mock_uss.client import MockUSSClient
+from datetime import datetime
 from monitoring.uss_qualifier.scenarios.astm.utm.data_exchange_validation.test_steps.wait import (
     wait_in_intervals,
 )
 from monitoring.monitorlib.clients.mock_uss.interactions import (
     Interaction,
     QueryDirection,
 )
-from monitoring.monitorlib.fetch import (
-    QueryType,
-)
 from uas_standards.astm.f3548.v21.api import (
     OperationID,
     PutOperationalIntentDetailsParameters,
@@ -55,7 +53,7 @@ def expect_tested_uss_receives_notification_from_mock_uss(
         "Mock USS sends valid notification", mock_uss.participant_id
     ) as check:
         try:
-            interactions, query = wait_in_intervals(mock_uss_interactions)(
+            interactions, query = wait_in_intervals(get_mock_uss_interactions)(
                 scenario,
                 mock_uss,
                 StringBasedDateTime(interactions_since_time),

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

@@ -0,0 +1,7 @@
+# Get mock_uss interactions test step fragment
+This step obtains interactions of interest from mock_uss.
+
+## 🛑 Mock USS interactions logs retrievable check
+The Mock USSes provide a GET endpoint to retrieve all the interactions that took place between them and other USSes
+after a particular time.
+If there is any error retrieving these interactions, this check will fail as per **[interuss.mock_uss.hosted_instance.ExposeInterface](../../../requirements/interuss/mock_uss/hosted_instance.md)**.