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_op_intent_reference] Use QueryError for error handling #497

Merged
merged 2 commits into from
Feb 9, 2024
Merged

[uss_qualifier/resources/f3548/dss/get_op_intent_reference] Use QueryError for error handling #497

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
afaafd7
[uss_qualifier/resources/f3548/dss/get_op_intent_reference] Use Query…
mickmis Feb 6, 2024
a238cb2
remove added docstring
mickmis Feb 6, 2024
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
13 changes: 8 additions & 5 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 @@ -144,6 +144,8 @@ def get_op_intent_reference(
) -> Tuple[OperationalIntentReference, Query]:
"""
Retrieve an OP Intent from the DSS, using only its ID
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.GetOperationalIntentReference]
Expand All @@ -156,12 +158,13 @@ def get_op_intent_reference(
scope=Scope.StrategicCoordination,
)
if query.status_code != 200:
result = None
raise QueryError(
f"Received code {query.status_code} when attempting to retrieve operational intent reference {op_intent_id}",
query,
)
else:
result = ImplicitDict.parse(
query.response.json, GetOperationalIntentReferenceResponse
).operational_intent_reference
return result, query
result = query.parse_json_result(GetOperationalIntentReferenceResponse)
return result.operational_intent_reference, query

def get_full_op_intent(
self,
Expand Down
57 changes: 35 additions & 22 deletions monitoring/uss_qualifier/scenarios/astm/utm/op_intent_ref_access_control.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,19 +136,24 @@ def run(self, context: ExecutionContext):
self.end_test_scenario()

def _clean_known_op_intents_ids(self):
(oi_ref, q) = self._dss.get_op_intent_reference(self._oid_1)
self.record_query(q)

with self.check(
"Operational intent references can be queried by ID",
self._pid,
) as check:
# If the Op Intent does not exist, it's fine to run into a 404.
if q.response.status_code not in [200, 404]:
check.record_failed(
f"Could not access operational intent using main credentials",
details=f"DSS responded with {q.response.status_code} to attempt to access OI {self._oid_1}",
query_timestamps=[q.request.timestamp],
)
try:
(oi_ref, q) = self._dss.get_op_intent_reference(self._oid_1)
self.record_query(q)
except QueryError as e:
self.record_queries(e.queries)
q = e.queries[0]
# If the Op Intent does not exist, it's fine to run into a 404.
if q.response.status_code != 404:
check.record_failed(
f"Could not access operational intent using main credentials",
details=f"DSS responded with {q.response.status_code} to attempt to access OI {self._oid_1}; {e}",
query_timestamps=[q.request.timestamp],
)
if q.response.status_code != 404:
with self.check(
"Operational intent references can be searched for",
Expand All @@ -167,18 +172,24 @@ def _clean_known_op_intents_ids(self):
query_timestamps=[dq.request.timestamp],
)

(oi_ref, q) = self._dss_separate_creds.get_op_intent_reference(self._oid_2)
self.record_query(q)
with self.check(
"Operational intent references can be queried by ID",
self._pid,
) as check:
if q.response.status_code not in [200, 404]:
check.record_failed(
f"Could not access operational intent using second credentials",
details=f"DSS responded with {q.response.status_code} to attempt to access OI {self._oid_2}",
query_timestamps=[q.request.timestamp],
try:
(oi_ref, q) = self._dss_separate_creds.get_op_intent_reference(
self._oid_2
)
self.record_query(q)
except QueryError as e:
self.record_queries(e.queries)
q = e.queries[0]
if q.response.status_code != 404:
check.record_failed(
f"Could not access operational intent using second credentials",
details=f"DSS responded with {q.response.status_code} to attempt to access OI {self._oid_2}; {e}",
query_timestamps=[q.request.timestamp],
)
if q.response.status_code != 404:
with self.check(
"Operational intent references can be deleted by their owner", self._pid
Expand Down Expand Up @@ -441,19 +452,21 @@ def _check_mutation_on_non_owned_intent_fails(self):
query_timestamps=[dq.request.timestamp],
)

# Query again to confirm that the op intent has not been modified in any way:
(op_1_current, qcheck) = self._dss.get_op_intent_reference(self._oid_1)
self.record_query(qcheck)

with self.check(
"Operational intent references can be queried directly by their ID",
self._pid,
) as check:
if qcheck.response.status_code != 200:
try:
# Query again to confirm that the op intent has not been modified in any way:
(op_1_current, qcheck) = self._dss.get_op_intent_reference(self._oid_1)
self.record_query(qcheck)
except QueryError as e:
self.record_queries(e.queries)
qcheck = e.queries[0]
check.record_failed(
f"Could not access operational intent using main credentials",
details=f"DSS responded with {qcheck.response.status_code} to attempt to access OI {self._oid_1} "
f"while this OI should have been available.",
f"while this OI should have been available; {e}",
query_timestamps=[qcheck.request.timestamp],
)

Expand Down
Loading