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.

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] Fix no-notification check logic #872

Merged
merged 1 commit into from
Dec 19, 2024
Merged
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -85,33 +85,35 @@ def expect_no_interuss_post_interactions(
direction_filter(QueryDirection.Incoming),
)

for interaction in interactions:
with scenario.check(
"Mock USS interaction can be parsed", [mock_uss.participant_id]
) as check:
try:
req = PutOperationalIntentDetailsParameters(
ImplicitDict.parse(
interaction.query.request.json,
PutOperationalIntentDetailsParameters,
with scenario.check(
"Expect Notification not sent", [participant_id]
) as no_notification_check:
for interaction in interactions:
with scenario.check(
"Mock USS interaction can be parsed", [mock_uss.participant_id]
) as check:
try:
req = PutOperationalIntentDetailsParameters(
ImplicitDict.parse(
interaction.query.request.json,
PutOperationalIntentDetailsParameters,
)
)
except (ValueError, TypeError, KeyError) as e:
check.record_failed(
summary=f"Failed to parse request of a 'NotifyOperationalIntentDetailsChanged' interaction with mock_uss as a PutOperationalIntentDetailsParameters",
details=f"{str(e)}\nRequest: {interaction.query.request.json}\n\nStack trace:\n{e.stacktrace}",
query_timestamps=[query.request.timestamp],
)
continue # low priority failure: continue checking interactions if one cannot be parsed

op_intent_id = EntityID(req.operational_intent_id)
if op_intent_id not in shared_op_intent_ids:
no_notification_check.record_failed(
summary=f"Observed unexpected notification for operational intent ID {req.operational_intent_id}.",
details=f"Notification for operational intent ID {req.operational_intent_id} triggered by subscriptions {', '.join([sub.subscription_id for sub in req.subscriptions])} with timestamp {interaction.query.request.timestamp}.",
query_timestamps=[query.request.timestamp],
)
)
except (ValueError, TypeError, KeyError) as e:
check.record_failed(
summary=f"Failed to parse request of a 'NotifyOperationalIntentDetailsChanged' interaction with mock_uss as a PutOperationalIntentDetailsParameters",
details=f"{str(e)}\nRequest: {interaction.query.request.json}\n\nStack trace:\n{e.stacktrace}",
query_timestamps=[query.request.timestamp],
)
continue # low priority failure: continue checking interactions if one cannot be parsed

with scenario.check("Expect Notification not sent", [participant_id]) as check:
op_intent_id = EntityID(req.operational_intent_id)
if op_intent_id not in shared_op_intent_ids:
check.record_failed(
summary=f"Observed unexpected notification for operational intent ID {req.operational_intent_id}.",
details=f"Notification for operational intent ID {req.operational_intent_id} triggered by subscriptions {', '.join([sub.subscription_id for sub in req.subscriptions])} with timestamp {interaction.query.request.timestamp}.",
query_timestamps=[query.request.timestamp],
)


def expect_uss_obtained_op_intent_details(
Expand Down
Loading