Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[uss_qualifier/resources/f3548/dss/get_full_op_intent] Use QueryError for error handling #501

Merged
merged 1 commit into from
Feb 9, 2024
Merged

[uss_qualifier/resources/f3548/dss/get_full_op_intent] Use QueryError for error handling #501

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 11 additions & 26 deletions monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -177,28 +177,10 @@ def get_full_op_intent(
op_intent_ref: OperationalIntentReference,
uss_participant_id: Optional[str] = None,
) -> Tuple[OperationalIntent, 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, Query]:
"""
GET OperationalIntent without validating, as invalid data expected for negative tests

Returns:
returns the response json when query is successful
Retrieve a full operational intent from its managing USS.
Raises:
* QueryError: if request failed, if HTTP status code is different than 200, or if the parsing of the response failed.
"""
self._uses_scope(Scope.StrategicCoordination)
op = OPERATIONS[OperationID.GetOperationalIntentDetails]
Expand All @@ -210,11 +192,14 @@ def get_full_op_intent_without_validation(
uss_participant_id,
scope=Scope.StrategicCoordination,
)
result = None
if query.status_code == 200:
result = query.response.json

return result, query
if query.status_code != 200:
raise QueryError(
f"Received code {query.status_code} when attempting to retrieve operational intent details for {op_intent_ref.id}{f'; error message: `{query.error_message}`' if query.error_message is not None else ''}",
query,
)
else:
result = query.parse_json_result(GetOperationalIntentDetailsResponse)
return result.operational_intent, query

def get_op_intent_telemetry(
self,
Expand Down
63 changes: 37 additions & 26 deletions monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
Volume4D,
OperationalIntentReference,
GetOperationalIntentDetailsResponse,
EntityID,
)
from uas_standards.astm.f3548.v21.constants import Scope
from monitoring.monitorlib.clients.flight_planning.flight_info import (
Expand Down Expand Up @@ -237,12 +238,25 @@ def expect_shared_with_invalid_data(
if oi_ref is None:
return None

goidr_json, oi_full_query = self._dss.get_full_op_intent_without_validation(
oi_ref, self._flight_planner.participant_id
)

self._scenario.record_query(oi_full_query)
self._operational_intent_retrievable_check(oi_full_query, oi_ref.id)
with self._scenario.check(
"Operational intent details retrievable",
[self._flight_planner.participant_id],
) as check:
try:
goidr_json, oi_full_query = self._dss.get_full_op_intent(
oi_ref, self._flight_planner.participant_id
)
self._scenario.record_query(oi_full_query)
except QueryError as e:
self._scenario.record_queries(e.queries)
oi_full_query = e.queries[0]
if oi_full_query.status_code != 200:
# fail only if details could not be retrieved, as validation failures are acceptable here
check.record_failed(
summary="Operational intent details could not be retrieved from USS",
details=f"Received status code {oi_full_query.status_code} from {self._flight_planner.participant_id} when querying for details of operational intent {oi_ref.id}; {e}",
query_timestamps=[oi_full_query.request.timestamp],
)

validation_failures = self._evaluate_op_intent_validation(oi_full_query)
expected_validation_failure_found = self._expected_validation_failure_found(
Expand All @@ -264,21 +278,6 @@ def expect_shared_with_invalid_data(

return oi_ref

def _operational_intent_retrievable_check(
self, oi_full_query: fetch.Query, ref_id: str
):
with self._scenario.check(
"Operational intent details retrievable",
[self._flight_planner.participant_id],
) as check:
if oi_full_query.status_code != 200:
check.record_failed(
summary="Operational intent details could not be retrieved from USS",
severity=Severity.High,
details=f"Received status code {oi_full_query.status_code} from {self._flight_planner.participant_id} when querying for details of operational intent {ref_id}",
query_timestamps=[oi_full_query.request.timestamp],
)

def _operational_intent_shared_check(
self,
flight_intent: FlightInfo,
Expand Down Expand Up @@ -365,11 +364,23 @@ def _operational_intent_shared_check(
def _check_op_intent_details(
self, flight_intent: FlightInfo, oi_ref: OperationalIntentReference
):
oi_full, oi_full_query = self._dss.get_full_op_intent(
oi_ref, self._flight_planner.participant_id
)
self._scenario.record_query(oi_full_query)
self._operational_intent_retrievable_check(oi_full_query, oi_ref.id)
with self._scenario.check(
"Operational intent details retrievable",
[self._flight_planner.participant_id],
) as check:
try:
oi_full, oi_full_query = self._dss.get_full_op_intent(
oi_ref, self._flight_planner.participant_id
)
self._scenario.record_query(oi_full_query)
except QueryError as e:
self._scenario.record_queries(e.queries)
oi_full_query = e.queries[0]
check.record_failed(
summary="Operational intent details could not be retrieved from USS",
details=f"Received status code {oi_full_query.status_code} from {self._flight_planner.participant_id} when querying for details of operational intent {oi_ref.id}; {e}",
query_timestamps=[oi_full_query.request.timestamp],
)

validation_failures = self._evaluate_op_intent_validation(oi_full_query)
with self._scenario.check(
Expand Down
Loading