diff --git a/monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py b/monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py index f60b1a8f4f..76c29fbdb6 100644 --- a/monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py +++ b/monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py @@ -29,6 +29,8 @@ UssAvailabilityState, UssAvailabilityStatusResponse, GetOperationalIntentReferenceResponse, + OPERATIONS, + OperationID, ) # A base URL for a USS that is not expected to be ever called @@ -77,12 +79,12 @@ def __init__( def find_op_intent( self, extent: Volume4D ) -> Tuple[List[OperationalIntentReference], fetch.Query]: - url = "/dss/v1/operational_intent_references/query" + op = OPERATIONS[OperationID.QueryOperationalIntentReferences] req = QueryOperationalIntentReferenceParameters(area_of_interest=extent) query = fetch.query_and_describe( self.client, - "POST", - url, + op.verb, + op.path, QueryType.F3548v21DSSQueryOperationalIntentReferences, self.participant_id, scope=SCOPE_SC, @@ -103,11 +105,11 @@ def get_op_intent_reference( """ Retrieve an OP Intent from the DSS, using only its ID """ - url = f"/dss/v1/operational_intent_references/{op_intent_id}" + op = OPERATIONS[OperationID.GetOperationalIntentReference] query = fetch.query_and_describe( self.client, - "GET", - url, + op.verb, + op.path.format(entityid=op_intent_id), QueryType.F3548v21DSSGetOperationalIntentReference, self.participant_id, scope=SCOPE_SC, @@ -125,11 +127,11 @@ def get_full_op_intent( op_intent_ref: OperationalIntentReference, uss_participant_id: Optional[str] = None, ) -> Tuple[OperationalIntent, fetch.Query]: - url = f"{op_intent_ref.uss_base_url}/uss/v1/operational_intents/{op_intent_ref.id}" + op = OPERATIONS[OperationID.GetOperationalIntentDetails] query = fetch.query_and_describe( self.client, - "GET", - url, + op.verb, + f"{op_intent_ref.uss_base_url}{op.path.format(entityid=op_intent_ref.id)}", QueryType.F3548v21USSGetOperationalIntentDetails, uss_participant_id, scope=SCOPE_SC, @@ -157,10 +159,12 @@ def put_op_intent( ]: oi_uuid = str(uuid.uuid4()) if id is None else id if ovn is None: - url = f"/dss/v1/operational_intent_references/{oi_uuid}" + op = OPERATIONS[OperationID.CreateOperationalIntentReference] + url = op.path.format(entityid=oi_uuid) query_type = QueryType.F3548v21DSSCreateOperationalIntentReference else: - url = f"/dss/v1/operational_intent_references/{oi_uuid}/{ovn}" + op = OPERATIONS[OperationID.UpdateOperationalIntentReference] + url = op.path.format(entityid=oi_uuid, ovn=ovn) query_type = QueryType.F3548v21DSSUpdateOperationalIntentReference req = PutOperationalIntentReferenceParameters( @@ -172,7 +176,7 @@ def put_op_intent( ) query = fetch.query_and_describe( self.client, - "PUT", + op.verb, url, query_type, self.participant_id, @@ -198,10 +202,11 @@ def delete_op_intent( Optional[List[SubscriberToNotify]], fetch.Query, ]: + op = OPERATIONS[OperationID.DeleteOperationalIntentReference] query = fetch.query_and_describe( self.client, - "DELETE", - f"/dss/v1/operational_intent_references/{id}/{ovn}", + op.verb, + op.path.format(entityid=id, ovn=ovn), QueryType.F3548v21DSSDeleteOperationalIntentReference, self.participant_id, scope=SCOPE_SC, @@ -237,10 +242,11 @@ def set_uss_availability( old_version=version, availability=availability, ) + op = OPERATIONS[OperationID.SetUssAvailability] query = fetch.query_and_describe( self.client, - "PUT", - f"/dss/v1/uss_availability/{uss_id}", + op.verb, + op.path.format(uss_id=uss_id), QueryType.F3548v21DSSSetUssAvailability, self.participant_id, scope=SCOPE_AA,