factorize fragments

Author: RustedBones

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

@@ -416,15 +416,12 @@ def get_uss_availability(
     def set_uss_availability(
         self,
         uss_id: str,
-        available: bool | None,
+        availability: UssAvailabilityState,
         version: str = "",
     ) -> tuple[str, Query]:
         """
         Set the availability for the USS identified by 'uss_id'.
 
-        If 'available' is None, the availability will be set to 'Unknown'.
-        True will set it to 'Normal', and False to 'Down'.
-
         Returns:
             A tuple composed of
             1) the new version of the USS availability;
@@ -433,13 +430,6 @@ def set_uss_availability(
             * QueryError: if request failed, if HTTP status code is different than 200, or if the parsing of the response failed.
         """
         self._uses_scope(Scope.AvailabilityArbitration)
-        if available is None:
-            availability = UssAvailabilityState.Unknown
-        elif available:
-            availability = UssAvailabilityState.Normal
-        else:
-            availability = UssAvailabilityState.Down
-
         req = SetUssAvailabilityStatusParameters(
             old_version=version,
             availability=availability,

monitoring/uss_qualifier/scenarios/astm/utm/dss/authentication/authentication_validation.py

@@ -391,7 +391,9 @@ def _ensure_availability_is_unknown(self):
             with self.check("USS Availability can be updated", self._pid) as check:
                 try:
                     availability, q = self._availability_dss.set_uss_availability(
-                        self._test_id, available=None, version=availability.version
+                        self._test_id,
+                        UssAvailabilityState.Unknown,
+                        availability.version,
                     )
                     self.record_query(q)
                 except QueryError as e:

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/sync.md

@@ -1,6 +1,6 @@
 # USS Availability Sync test step fragment
 
-## 🛑USS Availability is consistent across every DSS instance check
+## 🛑 USS Availability is consistent across every DSS instance check
 
 If the reported availability for a USS is not consistent, across a set of DSS instances, with the value that was previously read or set on an arbitrary DSS instance,
 either the DSS through which the value was set or the one through which the values was retrieved is failing to meet at least one of these requirements:
@@ -11,7 +11,7 @@ either the DSS through which the value was set or the one through which the valu
 
 As a consequence, the DSS also fails to meet **[astm.f3548.v21.DSS0210,A2-7-2,6](../../../../../../requirements/astm/f3548/v21.md)** and **[astm.f3548.v21.DSS0020](../../../../../../requirements/astm/f3548/v21.md)**.
 
-## 🛑USS Availability version is consistent across every DSS instance check
+## 🛑 USS Availability version is consistent across every DSS instance check
 
 If the reported availability version for a USS is not consistent, across a set of DSS instances, with the value that was previously read or set on an arbitrary DSS instance,
 either the DSS through which the value was set or the one through which the values was retrieved is failing to meet at least one of these requirements:

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

@@ -1,9 +1,9 @@
-# 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
+## 🛑 USS Availability can be updated check
 
 If, when presented with a valid query to update the availability state of a USS, a DSS
 responds with anything else than a 200 OK response, it is in violation of the OpenAPI specification referenced by **[astm.f3548.v21.DSS0100,1](../../../../../../requirements/astm/f3548/v21.md)**.

monitoring/uss_qualifier/scenarios/astm/utm/dss/synchronization/uss_availability_synchronization.py

@@ -8,6 +8,10 @@
     DSSInstancesResource,
 )
 from monitoring.uss_qualifier.resources.communications import ClientIdentityResource
+from monitoring.uss_qualifier.scenarios.astm.utm.test_steps import (
+    get_uss_availability,
+    set_uss_availability,
+)
 from monitoring.uss_qualifier.scenarios.scenario import TestScenario
 from monitoring.uss_qualifier.suites.suite import ExecutionContext
 
@@ -145,88 +149,51 @@ def _step_unknown_uss_reported_as_unknown(self):
         )
 
     def _step_update_to_unknown(self):
-        with self.check(
-            "USS Availability can be updated", self._dss.participant_id
-        ) as check:
-            try:
-                self._current_version, q = self._dss.set_uss_availability(
-                    self._uss_id, None, self._current_version
-                )
-                self.record_query(q)
-            except QueryError as qe:
-                self.record_queries(qe.queries)
-                check.record_failed(
-                    summary="Failed to set USS availability to Unknown",
-                    details=qe.msg,
-                    query_timestamps=qe.query_timestamps,
-                )
+        self._current_version = set_uss_availability(
+            self,
+            self._dss,
+            self._uss_id,
+            UssAvailabilityState.Unknown,
+            self._current_version,
+        )
 
     def _step_update_to_down(self):
-        with self.check(
-            "USS Availability can be updated", self._dss.participant_id
-        ) as check:
-            try:
-                self._current_version, q = self._dss.set_uss_availability(
-                    self._uss_id, False, self._current_version
-                )
-                self.record_query(q)
-            except QueryError as qe:
-                self.record_queries(qe.queries)
-                check.record_failed(
-                    summary="Failed to set USS availability to Down",
-                    details=qe.msg,
-                    query_timestamps=qe.query_timestamps,
-                )
+        self._current_version = set_uss_availability(
+            self,
+            self._dss,
+            self._uss_id,
+            UssAvailabilityState.Down,
+            self._current_version,
+        )
 
     def _step_update_to_normal(self):
-        with self.check(
-            "USS Availability can be updated", self._dss.participant_id
-        ) as check:
-            try:
-                self._current_version, q = self._dss.set_uss_availability(
-                    self._uss_id, True, self._current_version
-                )
-                self.record_query(q)
-            except QueryError as qe:
-                self.record_queries(qe.queries)
-                check.record_failed(
-                    summary="Failed to set USS availability to Normal",
-                    details=qe.msg,
-                    query_timestamps=qe.query_timestamps,
-                )
+        self._current_version = set_uss_availability(
+            self,
+            self._dss,
+            self._uss_id,
+            UssAvailabilityState.Normal,
+            self._current_version,
+        )
 
     def _ensure_test_uss_availability_unknown(self, check_consistency: bool = True):
         """
         Ensure that the availability of the USS being used for the test is set to 'Unknown',
         the default state for USS availability when nothing else is known.
         We want to both start and end this scenario with this state.
         """
-
-        with self.check(
-            "USS Availability can be requested", self._dss.participant_id
-        ) as check:
-            try:
-                availability, q = self._dss.get_uss_availability(
-                    self._uss_id, Scope.AvailabilityArbitration
-                )
-                self.record_query(q)
-            except QueryError as qe:
-                self.record_queries(qe.queries)
-                check.record_failed(
-                    summary="Failed to get USS availability",
-                    details=qe.msg,
-                    query_timestamps=qe.query_timestamps,
-                )
-                return
-
-        self._current_version = availability.version
+        availability, version = get_uss_availability(
+            self, self._dss, self._uss_id, Scope.AvailabilityArbitration
+        )
+        self._current_version = version
 
         # If the state is not currently unknown, we set it to unknown
-        if availability.status.availability != UssAvailabilityState.Unknown:
+        if availability != UssAvailabilityState.Unknown:
             with self.check("USS Availability can be set to Unknown") as check:
                 try:
                     self._current_version, q = self._dss.set_uss_availability(
-                        self._uss_id, None, self._current_version
+                        self._uss_id,
+                        UssAvailabilityState.Unknown,
+                        self._current_version,
                     )
                     self.record_query(q)
                 except QueryError as qe:
@@ -252,40 +219,27 @@ def _query_and_expect_on_secondary(
         expected_availability: UssAvailabilityState,
         expected_version: str,
     ):
-        with self.check(
-            "USS Availability can be requested", dss.participant_id
-        ) as check:
-            try:
-                availability, q = dss.get_uss_availability(
-                    uss_id, Scope.StrategicCoordination
-                )
-                self.record_query(q)
-            except QueryError as qe:
-                self.record_queries(qe.queries)
-                check.record_failed(
-                    summary="Failed to get USS availability",
-                    details=qe.msg,
-                    query_timestamps=qe.query_timestamps,
-                )
-                return
+        availability, version = get_uss_availability(
+            self, dss, uss_id, Scope.StrategicCoordination
+        )
 
         with self.check(
             "USS Availability is consistent across every DSS instance", participants
         ) as check:
-            if availability.status.availability != expected_availability:
+            if availability != expected_availability:
                 check.record_failed(
                     summary="USS availability not as expected on secondary DSS",
-                    details=f"Expected {expected_availability}, got {availability.status.availability}",
+                    details=f"Expected {expected_availability}, got {availability}",
                 )
 
         with self.check(
             "USS Availability version is consistent across every DSS instance",
             participants,
         ) as check:
-            if availability.version != expected_version:
+            if version != expected_version:
                 check.record_failed(
                     summary="USS availability version not as expected on secondary DSS",
-                    details=f"Expected {expected_version}, got {availability.version}",
+                    details=f"Expected {expected_version}, got {version}",
                 )
 
     def _query_and_expect_on_secondaries(

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

@@ -12,7 +12,7 @@ Verifies the behavior of a DSS for simple interactions pertaining to USS availab
 
 ### client_identity
 
-[`ClientIdentityResource`](../../../../resources/communications/client_identity.py) the client identity that will be used to report the availability status.
+[`ClientIdentityResource`](../../../../resources/communications/client_identity.py) the client identity with the `utm.availability_arbitration` scope that will be used to report the availability status.
 
 ## Setup test case
 

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

@@ -1,3 +1,4 @@
+from uas_standards.astm.f3548.v21.api import UssAvailabilityState
 from uas_standards.astm.f3548.v21.constants import Scope
 
 from monitoring.monitorlib.fetch import QueryError
@@ -73,7 +74,7 @@ def _step_attempt_update_missing_version(self):
             try:
                 _, q = self._dss.set_uss_availability(
                     self._uss_id,
-                    False,
+                    UssAvailabilityState.Down,
                     "",
                 )
                 self.record_query(q)
@@ -105,7 +106,7 @@ def _step_attempt_update_incorrect_version(self):
             try:
                 _, q = self._dss.set_uss_availability(
                     self._uss_id,
-                    False,
+                    UssAvailabilityState.Down,
                     "ThisIsAnIncorrectVersion",
                 )
                 self.record_query(q)

monitoring/uss_qualifier/scenarios/astm/utm/off_nominal_planning/down_uss.py

@@ -2,6 +2,7 @@
 from uas_standards.astm.f3548.v21.api import (
     OperationalIntentReference,
     OperationalIntentState,
+    UssAvailabilityState,
 )
 from uas_standards.astm.f3548.v21.constants import Scope
 
@@ -319,7 +320,7 @@ def cleanup(self):
             try:
                 availability_version, avail_query = self.dss.set_uss_availability(
                     self.uss_qualifier_sub,
-                    True,
+                    UssAvailabilityState.Normal,
                 )
                 self.record_query(avail_query)
             except QueryError as e:

monitoring/uss_qualifier/scenarios/astm/utm/set_uss_available.md

@@ -4,5 +4,6 @@ This step sets the USS availability to 'Available' at the DSS.
 
 See `set_uss_available` in [test_steps.py](test_steps.py).
 
-## 🛑 USS availability successfully set to 'Available' check
-**[astm.f3548.v21.DSS0100,1](../../../requirements/astm/f3548/v21.md)**
+## [Availability can be read](./dss/fragments/availability/read.md)
+
+## [Availability can be updated](./dss/fragments/availability/update.md)