Merge remote-tracking branch 'interuss/main' into netrid-testing-visualization

Author: BenjaminPelletier

monitoring/monitorlib/inspection.py

@@ -2,6 +2,8 @@
 import inspect
 import pkgutil
 
+_modules_imported = set()
+
 
 def import_submodules(module) -> None:
     """Ensure that all descendant modules of a module are loaded.
@@ -10,10 +12,13 @@ def import_submodules(module) -> None:
 
     :param module: Parent module from which to start explicitly importing modules.
     """
+    if module in _modules_imported:
+        return
     for loader, module_name, is_pkg in pkgutil.walk_packages(
         module.__path__, module.__name__ + "."
     ):
         importlib.import_module(module_name)
+    _modules_imported.add(module)
 
 
 def get_module_object_by_name(parent_module, object_name: str):

monitoring/uss_qualifier/configurations/configuration.py

@@ -168,24 +168,13 @@ class FullyQualifiedCheck(ImplicitDict):
     """Scenario in which the check occurs."""
 
     test_case_name: str
-    """Test case in which the check occurs."""
+    """Test case in which the check occurs, omitting the ' test case' suffix.  Must be an exact match to documentation; sensitive to case and spacing."""
 
     test_step_name: str
-    """Test step in which the check occurs."""
+    """Test step in which the check occurs, omitting the ' test step' suffix.  Must be an exact match to documentation; sensitive to case and spacing."""
 
     check_name: str
-    """Name of the check."""
-
-    def contained_in(self, collection: Iterable[FullyQualifiedCheck]) -> bool:
-        for other in collection:
-            if (
-                self.scenario_type == other.scenario_type
-                and self.test_case_name == other.test_case_name
-                and self.test_step_name == other.test_step_name
-                and self.check_name == other.check_name
-            ):
-                return True
-        return False
+    """Name of the check, omitting the ' check' suffix.  Must be an exact match to documentation; sensitive to case and spacing."""
 
 
 class TestedRequirementsConfiguration(ImplicitDict):

monitoring/uss_qualifier/configurations/dev/f3548_self_contained.yaml

@@ -417,7 +417,7 @@ v1:
 
         # It is acceptable for the checks listed below to fail (in terms of determining the status of tested requirements in this artifact)
         acceptable_findings:
-          - scenario_type: scenarios.astm.utm.dss.DSSInteroperability
+          - scenario_type: scenarios.astm.utm.dss.dss_interoperability.DSSInteroperability
             test_case_name: "Prerequisites"
             test_step_name: "Test environment requirements"
             check_name: "DSS instance is publicly addressable"

monitoring/uss_qualifier/reports/tested_requirements/breakdown.py

@@ -37,7 +37,11 @@
 from monitoring.uss_qualifier.requirements.definitions import RequirementID
 from monitoring.uss_qualifier.scenarios.definitions import TestScenarioTypeName
 from monitoring.uss_qualifier.scenarios.documentation.parsing import get_documentation
-from monitoring.uss_qualifier.scenarios.scenario import get_scenario_type_by_name
+from monitoring.uss_qualifier.scenarios.scenario import (
+    are_scenario_types_equal,
+    fully_qualified_check_in_collection,
+    get_scenario_type_by_name,
+)
 from monitoring.uss_qualifier.suites.definitions import (
     ActionType,
     TestSuiteActionDeclaration,
@@ -188,7 +192,7 @@ def _populate_breakdown_with_scenario_report(
                 matches = [
                     s
                     for s in tested_requirement.scenarios
-                    if s.type == scenario_type_name
+                    if are_scenario_types_equal(s.type, scenario_type_name)
                 ]
                 if matches:
                     tested_scenario = matches[0]
@@ -237,8 +241,8 @@ def _populate_breakdown_with_scenario_report(
                         name=check.name,
                         url="",
                         has_todo=False,
-                        is_finding_acceptable=current_check.contained_in(
-                            acceptable_findings
+                        is_finding_acceptable=fully_qualified_check_in_collection(
+                            current_check, acceptable_findings
                         ),
                     )  # TODO: Consider populating has_todo with documentation instead
                     if isinstance(check, FailedCheck):
@@ -340,7 +344,7 @@ def _populate_breakdown_with_scenario(
                     matches = [
                         s
                         for s in tested_requirement.scenarios
-                        if s.type == scenario_type_name
+                        if are_scenario_types_equal(s.type, scenario_type_name)
                     ]
                     if matches:
                         tested_scenario = matches[0]
@@ -383,8 +387,8 @@ def _populate_breakdown_with_scenario(
                             name=check.name,
                             url=check.url,
                             has_todo=check.has_todo,
-                            is_finding_acceptable=current_check.contained_in(
-                                acceptable_findings
+                            is_finding_acceptable=fully_qualified_check_in_collection(
+                                current_check, acceptable_findings
                             ),
                         )
                         tested_step.checks.append(tested_check)

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

@@ -20,3 +20,6 @@
 )
 from .subscription_simple import SubscriptionSimple as SubscriptionSimple
 from .subscription_validation import SubscriptionValidation as SubscriptionValidation
+from .uss_availability_mutation import (
+    UssAvailabilityMutation as UssAvailabilityMutation,
+)

monitoring/uss_qualifier/scenarios/astm/utm/dss/fragments/availability/read.md

@@ -1,6 +1,6 @@
 # USS Availability Read test step fragment
 
-This fragment contains the steps for the USS Availability synchronization scenario
+This fragment contains the steps for the USS Availability scenario
 where we confirm that a USS availability can be correctly read from a DSS instance
 
 ## 🛑 USS Availability can be requested check

monitoring/uss_qualifier/scenarios/astm/utm/dss/fragments/availability/update.md

@@ -1,6 +1,6 @@
-# USS Availability Read test step fragment
+# USS Availability Update test step fragment
 
-This fragment contains the steps for the USS Availability synchronization scenario
+This fragment contains the steps for the USS Availability scenario
 where we confirm that a USS availability can be correctly read from a DSS instance
 
 ## 🛑 USS Availability can be updated check

monitoring/uss_qualifier/scenarios/astm/utm/dss/uss_availability_mutation.md

@@ -0,0 +1,44 @@
+# ASTM Availability DSS: USS Availability Mutation test scenario
+
+## Overview
+
+Verifies the behavior of a DSS for simple interactions pertaining to USS availability status.
+
+## Resources
+
+### dss
+
+[`DSSInstanceResource`](../../../../resources/astm/f3548/v21/dss.py) the DSS instance through which entities are created, modified and deleted.
+
+### client_identity
+
+[`ClientIdentityResource`](../../../../resources/communications/client_identity.py) the client identity with the `utm.availability_arbitration` scope that will be used to report the availability status.
+
+## Update USS availability state test case
+
+### Declare USS as available at DSS test step
+
+#### [Availability can be read](./fragments/availability/read.md)
+
+#### [Availability can be updated](./fragments/availability/update.md)
+
+## Update requires correct version test case
+
+Test DSS behavior when update requests are not providing the required version.
+
+### Attempt update with missing version test step
+
+This step verifies that an existing USS availability status cannot be mutated with a missing version.
+
+#### 🛑 Request to update USS availability status with empty version fails check
+
+If the DSS under test allows the qualifier to update the USS availability status with a request that provided an empty version, it is in violation of **[astm.f3548.v21.DSS0100,1](../../../../requirements/astm/f3548/v21.md)**
+
+### Attempt update with incorrect version test step
+
+This step verifies that an existing OIR cannot be mutated with an incorrect version.
+
+#### 🛑 Request to update USS availability status with incorrect version fails check
+
+If the DSS under test allows the qualifier to update the USS availability status with a request that provided an incorrect version,
+it is in violation of **[astm.f3548.v21.DSS0100,1](../../../../requirements/astm/f3548/v21.md)**

monitoring/uss_qualifier/scenarios/astm/utm/dss/uss_availability_mutation.py

@@ -0,0 +1,129 @@
+from uas_standards.astm.f3548.v21.api import UssAvailabilityState
+from uas_standards.astm.f3548.v21.constants import Scope
+
+from monitoring.monitorlib.fetch import QueryError
+from monitoring.uss_qualifier.resources.astm.f3548.v21.dss import (
+    DSSInstance,
+    DSSInstanceResource,
+)
+from monitoring.uss_qualifier.resources.communications import ClientIdentityResource
+from monitoring.uss_qualifier.scenarios.astm.utm.dss.test_step_fragments import (
+    get_uss_availability,
+    set_uss_availability,
+)
+from monitoring.uss_qualifier.scenarios.scenario import TestScenario
+from monitoring.uss_qualifier.suites.suite import ExecutionContext
+
+
+class UssAvailabilityMutation(TestScenario):
+    """
+    A scenario that verifies that USS availability status cannot be updated with the incorrect version.
+    """
+
+    _dss: DSSInstance
+    _pid: list[str]
+
+    _uss_id: str
+
+    def __init__(
+        self,
+        dss: DSSInstanceResource,
+        client_identity: ClientIdentityResource,
+    ):
+        """
+        Args:
+            dss: dss to test
+            client_identity: tells us the identity we should expect as an entity's manager
+        """
+        super().__init__()
+        scopes: dict[str, str] = {
+            Scope.AvailabilityArbitration: "read and set availability for a USS"
+        }
+
+        self._dss = dss.get_instance(scopes)
+        self._pid = [self._dss.participant_id]
+
+        self._uss_id = client_identity.subject()
+
+    def run(self, context: ExecutionContext):
+        self.begin_test_scenario(context)
+
+        self.begin_test_case("Update USS availability state")
+        self._step_declare_uss_available()
+        self.end_test_case()
+
+        self.begin_test_case("Update requires correct version")
+        self._step_attempt_update_missing_version()
+        self._step_attempt_update_incorrect_version()
+        self.end_test_case()
+
+    def _step_declare_uss_available(self):
+        self.begin_test_step("Declare USS as available at DSS")
+        _, version = get_uss_availability(
+            self,
+            self._dss,
+            self._uss_id,
+            Scope.AvailabilityArbitration,
+        )
+        set_uss_availability(
+            self, self._dss, self._uss_id, UssAvailabilityState.Normal, version
+        )
+        self.end_test_step()
+
+    def _step_attempt_update_missing_version(self):
+        self.begin_test_step("Attempt update with missing version")
+        with self.check(
+            "Request to update USS availability status with empty version fails",
+            self._pid,
+        ) as check:
+            try:
+                _, q = self._dss.set_uss_availability(
+                    self._uss_id,
+                    UssAvailabilityState.Down,
+                    "",
+                )
+                self.record_query(q)
+                # We don't expect the reach this point:
+                check.record_failed(
+                    summary="Set USS availability with missing version was not expected to succeed",
+                    details=f"Was expecting an HTTP 409 response because of an missing version, but got {q.status_code} instead",
+                    query_timestamps=[q.request.timestamp],
+                )
+            except QueryError as qe:
+                self.record_queries(qe.queries)
+                if qe.cause_status_code != 409:
+                    check.record_failed(
+                        summary="Set USS availability with missing version failed for unexpected reason",
+                        details=f"Was expecting an HTTP 409 response because of an missing version, but got {qe.cause_status_code} instead",
+                        query_timestamps=qe.query_timestamps,
+                    )
+        self.end_test_step()
+
+    def _step_attempt_update_incorrect_version(self):
+        self.begin_test_step("Attempt update with incorrect version")
+        with self.check(
+            "Request to update USS availability status with incorrect version fails",
+            self._pid,
+        ) as check:
+            try:
+                _, q = self._dss.set_uss_availability(
+                    self._uss_id,
+                    UssAvailabilityState.Down,
+                    "ThisIsAnIncorrectVersion",
+                )
+                self.record_query(q)
+                # We don't expect the reach this point:
+                check.record_failed(
+                    summary="Set USS availability with incorrect version was not expected to succeed",
+                    details=f"Was expecting an HTTP 409 response because of an incorrect version, but got {q.status_code} instead",
+                    query_timestamps=[q.request.timestamp],
+                )
+            except QueryError as qe:
+                self.record_queries(qe.queries)
+                if qe.cause_status_code != 409:
+                    check.record_failed(
+                        summary="Set USS availability with incorrect version failed for unexpected reason",
+                        details=f"Was expecting an HTTP 409 response because of an incorrect version, but got {qe.cause_status_code} instead",
+                        query_timestamps=qe.query_timestamps,
+                    )
+        self.end_test_step()

monitoring/uss_qualifier/scenarios/definitions.py

@@ -3,7 +3,12 @@
 from monitoring.uss_qualifier.resources.definitions import ResourceID
 
 TestScenarioTypeName = str
-"""This plain string represents a type of test scenario, expressed as a Python class name qualified relative to the `uss_qualifier` module"""
+"""This plain string represents a type of test scenario, expressed as a Python class name qualified relative to the
+`uss_qualifier` module.
+
+Note that equality between different TestScenarioTypeNames (whether they refer to the same type of test scenario) should
+be determined via are_scenario_types_equal as multiple TestScenarioTypeNames may resolve to the same test scenario type.
+"""
 
 
 class TestScenarioDeclaration(ImplicitDict):