From 572bfeb21fed5b1ed571d7a1d1349f9947715618 Mon Sep 17 00:00:00 2001 From: Punam Verma Date: Tue, 31 Oct 2023 11:30:29 -0700 Subject: [PATCH] Per PR comments removing py files, and improving description --- .../get_op_data_validation.md | 6 +- .../expected_interactions_test_steps.py | 158 ------------ .../test_steps/invalid_op_test_steps.py | 226 ------------------ .../plan_flight_intent_expect_failed.md | 2 +- ...l_intent_but_with_invalid_interuss_data.md | 2 +- 5 files changed, 4 insertions(+), 390 deletions(-) delete mode 100644 monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/expected_interactions_test_steps.py delete mode 100644 monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/invalid_op_test_steps.py diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/get_op_data_validation.md b/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/get_op_data_validation.md index 5721888ddb..d3371833f0 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/get_op_data_validation.md +++ b/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/get_op_data_validation.md @@ -76,7 +76,7 @@ Teardown Flight 2 on time range A should be successfully planned by the control USS. ### [Validate flight 2 shared intent with invalid interuss data test step](test_steps/validate_sharing_operational_intent_but_with_invalid_interuss_data.md) -Validate that flight 2 is planned +Validate that flight 2 is shared with invalid data as a modified behavior is injected by uss_qualifier for a negative test. ### [Validate no notification pushed for flight 2](../validate_not_shared_operational_intent.md) There should be no subscription by tested_uss to trigger notification of flight 2. @@ -98,11 +98,9 @@ In a previous step, we checked there was no notification of flight 2 to tested_u ### [Validate flight 1 Notification not sent to Control_uss test step](test_steps/validate_no_notification_operational_intent.md) ### [Delete Control_uss flight test step](../../../flight_planning/delete_flight_intent.md) -Teardown -### [Delete tested_uss flight test step](../../../flight_planning/delete_flight_intent.md) -Teardown if created. ## Cleanup ### Successful flight deletion check +This cleanup is for both - after testcase ends and after test scenario ends **[interuss.automated_testing.flight_planning.DeleteFlightSuccess](../../../../requirements/interuss/automated_testing/flight_planning.md)** diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/expected_interactions_test_steps.py b/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/expected_interactions_test_steps.py deleted file mode 100644 index 2879769938..0000000000 --- a/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/expected_interactions_test_steps.py +++ /dev/null @@ -1,158 +0,0 @@ -from __future__ import annotations - -from typing import List, Optional -import time -import jwt - -from monitoring.monitorlib.fetch import QueryError -from monitoring.uss_qualifier.common_data_definitions import Severity -from monitoring.uss_qualifier.scenarios.scenario import TestScenarioType -from monitoring.uss_qualifier.resources.interuss.mock_uss.client import MockUSSClient -from implicitdict import StringBasedDateTime -from loguru import logger -from monitoring.monitorlib.clients.mock_uss.interactions import Interaction - - -def expect_interuss_post_interactions( - scenario: TestScenarioType, - mock_uss: MockUSSClient, - st: StringBasedDateTime, - posted_to_url: str, - test_step: str, -): - found = False - if mock_uss is not None: - interactions = _get_interuss_interactions(scenario, mock_uss, st, test_step) - logger.debug(f"Checking for Post to {posted_to_url}") - with scenario.check("Expect Notification sent") as check: - found = False - for interaction in interactions: - method = interaction.query.request.method - url = interaction.query.request.url - if method == "POST" and posted_to_url in url: - found = True - if found == False: - check.record_failed( - summary=f"Notification to {posted_to_url} not received", - severity=Severity.Medium, - details=f"Notification to {posted_to_url} not received", - requirements="SCDxxxx", - ) - scenario.end_test_step() - - return found - - -def expect_no_interuss_post_interactions( - scenario: TestScenarioType, - mock_uss: MockUSSClient, - st: StringBasedDateTime, - posted_to_url: str, - test_step: str, -): - found = False - if mock_uss is not None: - interactions = _get_interuss_interactions(scenario, mock_uss, st, test_step) - logger.debug(f"Checking for POST request to {posted_to_url}") - with scenario.check("Expect Notification not sent") as check: - found = False - for interaction in interactions: - method = interaction.query.request.method - url = interaction.query.request.url - if method == "POST" and posted_to_url in url: - found = True - if found == True: - check.record_failed( - summary=f"Notification to {posted_to_url} wrongly sent", - severity=Severity.Medium, - details=f"Notification to {posted_to_url} wrongly sent", - requirements="SCDxxxx", - ) - scenario.end_test_step() - - return found - - -def expect_interuss_get_interactions( - scenario: TestScenarioType, - mock_uss: MockUSSClient, - st: StringBasedDateTime, - get_from_url: str, - id: str, - test_step: str, -): - found = False - if mock_uss is not None: - interactions = _get_interuss_interactions(scenario, mock_uss, st, test_step) - logger.debug(f"Checking for GET request to {get_from_url} for id {id}") - with scenario.check("Expect GET request") as check: - found = False - for interaction in interactions: - method = interaction.query.request.method - url = interaction.query.request.url - if method == "GET" and get_from_url in url and id in url: - found = True - if found == False: - check.record_failed( - summary=f"No GET request received at {get_from_url} for {id} ", - severity=Severity.Medium, - details=f"No GET request received at {get_from_url} for {id}", - requirements="SCDxxxx", - ) - - scenario.end_test_step() - return found - - -def _get_interuss_interactions( - scenario: TestScenarioType, - mock_uss: MockUSSClient, - st: StringBasedDateTime, - test_step: str, -) -> List[Interaction]: - scenario.begin_test_step(test_step) - time.sleep(5) - with scenario.check("MockUSS interactions request") as check: - try: - all_interactions, query = mock_uss.get_interactions(st) - scenario.record_query(query) - except QueryError as e: - for q in e.queries: - scenario.record_query(q) - check.record_failed( - summary=f"Error from mock_uss when attempting to get interactions from_time {st}", - severity=Severity.High, - details=f"{str(e)}\n\nStack trace:\n{e.stacktrace}", - query_timestamps=[q.request.timestamp for q in e.queries], - ) - - exclude_sub = mock_uss.session.auth_adapter.get_sub() - - def is_uss_interaction(interaction: Interaction, excl_sub: str) -> bool: - headers = interaction.query.request.headers - if "Authorization" in headers: - token = headers.get("Authorization").split(" ")[1] - payload = jwt.decode( - token, algorithms="RS256", options={"verify_signature": False} - ) - sub = payload["sub"] - logger.debug(f"sub of interuss_interaction token: {sub}") - if sub == excl_sub: - logger.debug(f"Excluding interaction with sub: {sub} ") - return False - else: - return True - else: - logger.error(f"Interaction received without Authorization : {interaction}") - return False - - interuss_interactions = [] - for interaction in all_interactions: - if is_uss_interaction(interaction, exclude_sub): - interuss_interactions.append(interaction) - logger.debug( - f"Interuss interaction reported : {interaction.query.request.method} {interaction.query.request.url} " - f"with response {interaction.query.response.status_code}" - ) - - return interuss_interactions diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/invalid_op_test_steps.py b/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/invalid_op_test_steps.py deleted file mode 100644 index 7969f07e3f..0000000000 --- a/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/invalid_op_test_steps.py +++ /dev/null @@ -1,226 +0,0 @@ -from typing import Optional -from monitoring.monitorlib import schema_validation, fetch -from implicitdict import ImplicitDict -from uas_standards.astm.f3548.v21.api import ( - OperationalIntentState, - OperationalIntentReference, - GetOperationalIntentDetailsResponse, -) -from loguru import logger -from uas_standards.interuss.automated_testing.scd.v1.api import ( - InjectFlightRequest, - InjectFlightResponseResult, - InjectFlightResponse, -) -from monitoring.uss_qualifier.common_data_definitions import Severity -from monitoring.uss_qualifier.resources.astm.f3548.v21.dss import DSSInstance -from monitoring.uss_qualifier.resources.flight_planning.flight_planner import ( - FlightPlanner, -) -from monitoring.uss_qualifier.scenarios.astm.utm.test_steps import OpIntentValidator -from monitoring.uss_qualifier.scenarios.flight_planning.test_steps import ( - submit_flight_intent, - expect_flight_intent_state, -) -from monitoring.uss_qualifier.scenarios.scenario import TestScenarioType -from monitoring.uss_qualifier.resources.interuss.mock_uss.client import MockUSSClient - - -def plan_flight_intent_expect_failed( - scenario: TestScenarioType, - test_step: str, - flight_planner: FlightPlanner, - flight_intent: InjectFlightRequest, -) -> InjectFlightResponse: - """Attempt to plan a flight intent that would result in a Failed result. - - This function implements the test step described in scd_data_exchange_validation.md. - It validates requirement astm.f3548.v21.SCD00abc. - - Returns: The injection response. - """ - expect_flight_intent_state( - flight_intent, OperationalIntentState.Accepted, scenario, test_step - ) - - return submit_flight_intent( - scenario, - test_step, - "Plan should fail", - {InjectFlightResponseResult.Failed}, - { - InjectFlightResponseResult.Planned: "Failure If Planned", - InjectFlightResponseResult.ConflictWithFlight: "Failure If Conflict", - InjectFlightResponseResult.Rejected: "Failure If Rejected", - }, - flight_planner, - flight_intent, - )[0] - - -class InvalidOpIntentSharingValidator(OpIntentValidator): - def expect_shared_with_invalid_data( - self, flight_intent: InjectFlightRequest, skip_if_not_found: bool = False - ) -> Optional[OperationalIntentReference]: - """Validate that operational intent information was shared with dss for a flight intent, but shared invalid data with USS. - - This function implements the test step described in validate_sharing_operational_intent_but_with_invalid_interuss_data. - - :param flight_intent: the flight intent that was supposed to have been shared. - :param skip_if_not_found: set to True to skip the execution of the checks if the operational intent was not found while it should have been modified. - - :returns: the shared operational intent reference. None if skipped because not found. - """ - self._begin_step() - - with self._scenario.check( - "Operational intent shared with DSS", [self._flight_planner.participant_id] - ) as check: - if self._orig_oi_ref is None: - # we expect a new op intent to have been created - if self._new_oi_ref is None: - check.record_failed( - summary="Operational intent reference not found in DSS", - severity=Severity.High, - details=f"USS {self._flight_planner.participant_id} was supposed to have shared a new operational intent with the DSS, but no matching operational intent references were found in the DSS in the area of the flight intent", - query_timestamps=[self._after_query.request.timestamp], - ) - oi_ref = self._new_oi_ref - - elif self._new_oi_ref is None: - # We expect the original op intent to have been either modified or left untouched, thus must be among - # the returned op intents. If additionally the op intent corresponds to an active flight, we fail a - # different appropriate check. Exception made if skip_if_not_found=True and op intent was deleted: step - # is skipped. - modified_oi_ref = self._find_after_oi(self._orig_oi_ref.id) - if modified_oi_ref is None: - if not skip_if_not_found: - if ( - flight_intent.operational_intent.state - == OperationalIntentState.Activated - ): - with self._scenario.check( - "Operational intent for active flight not deleted", - [self._flight_planner.participant_id], - ) as active_flight_check: - active_flight_check.record_failed( - summary="Operational intent reference for active flight not found in DSS", - severity=Severity.High, - details=f"USS {self._flight_planner.participant_id} was supposed to have shared with the DSS an updated operational intent by modifying it, but no matching operational intent references were found in the DSS in the area of the flight intent", - query_timestamps=[ - self._after_query.request.timestamp - ], - ) - else: - check.record_failed( - summary="Operational intent reference not found in DSS", - severity=Severity.High, - details=f"USS {self._flight_planner.participant_id} was supposed to have shared with the DSS an updated operational intent by modifying it, but no matching operational intent references were found in the DSS in the area of the flight intent", - query_timestamps=[self._after_query.request.timestamp], - ) - else: - self._scenario.record_note( - self._flight_planner.participant_id, - f"Operational intent reference with ID {self._orig_oi_ref.id} not found in DSS, instructed to skip test step.", - ) - self._scenario.end_test_step() - return None - oi_ref = modified_oi_ref - - else: - # we expect the original op intent to have been replaced with a new one, thus old one must NOT be among the returned op intents - if self._find_after_oi(self._orig_oi_ref.id) is not None: - check.record_failed( - summary="Operational intent reference found duplicated in DSS", - severity=Severity.High, - details=f"USS {self._flight_planner.participant_id} was supposed to have shared with the DSS an updated operational intent by replacing it, but it ended up duplicating the operational intent in the DSS", - query_timestamps=[self._after_query.request.timestamp], - ) - oi_ref = self._new_oi_ref - - goidr_json, oi_full_query = self._dss.get_full_op_intent_without_validation( - oi_ref - ) - self._scenario.record_query(oi_full_query) - 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 {oi_ref.id}", - query_timestamps=[oi_full_query.request.timestamp], - ) - - # if schema validation errors or standard req validation - with self._scenario.check( - "Invalid data in Operational intent details shared by Mock USS for negative test", - [self._flight_planner.participant_id], - ) as check: - - validation_errors = [] - # schema_validation_errors = schema_validation.validate( - # schema_validation.F3548_21.OpenAPIPath, - # schema_validation.F3548_21.GetOperationalIntentDetailsResponse, - # oi_full_query.response.json, - # ) - schema_validation_errors = None - logger.debug(f"Schema validation errors {schema_validation_errors}") - if schema_validation_errors: - details = ( - "The response received from querying operational intent details failed validation against the required OpenAPI schema:\n" - + "\n".join( - f"At {e.json_path} in the response: {e.message}" - for e in schema_validation_errors - ) - ) - validation_errors.append(details) - else: - oi_full = None - try: - goidr = ImplicitDict.parse( - goidr_json, GetOperationalIntentDetailsResponse - ) - oi_full = goidr.operational_intent - - if ( - oi_full.reference.state == OperationalIntentState.Accepted - or oi_full.reference.state == OperationalIntentState.Activated - ) and oi_full.details.get("off_nominal_volumes", None): - details = f"Operational intent {oi_full.reference.id} had {len(oi_full.details.off_nominal_volumes)} off-nominal volumes in wrong state - {oi_full.reference.state}" - validation_errors.append(details) - - def volume_vertices(v4): - if "outline_circle" in v4.volume: - return 1 - if "outline_polygon" in v4.volume: - return len(v4.volume.outline_polygon.vertices) - - all_volumes = oi_full.details.get( - "volumes", [] - ) + oi_full.details.get("off_nominal_volumes", []) - n_vertices = sum(volume_vertices(v) for v in all_volumes) - - if n_vertices > 10000: - details = ( - f"Operational intent {oi_full.reference.id} had too many total vertices - {n_vertices}", - ) - validation_errors.append(details) - except (KeyError, ValueError) as e: - logger.debug( - f"Validation error in GetOperationalIntentDetailsResponse. {e}" - ) - validation_errors.append(e) - - if not validation_errors: - check.record_failed( - summary="This negative test case requires invalid data shared with other USS in Operational intent details ", - severity=Severity.High, - details=f"Data shared by Mock USS with other USSes had no invalid data. This test case required invalid data for testing.", - query_timestamps=[oi_full_query.request.timestamp], - ) - - self._scenario.end_test_step() - return oi_ref diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/plan_flight_intent_expect_failed.md b/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/plan_flight_intent_expect_failed.md index 6581b593da..ce747a72de 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/plan_flight_intent_expect_failed.md +++ b/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/plan_flight_intent_expect_failed.md @@ -1,6 +1,6 @@ # Plan flight Expect Failed test step -This page describes the content of a common test case where a valid user flight intent fails in a flight planner, because of invalid data for a nearby flight shared by another USS. See `plan_flight_intent_expect_failed` in [test_steps.py](invalid_op_test_steps.py). +This page describes the content of a common test case where a valid user flight intent fails in a flight planner, because of invalid data shared for a nearby flight shared by another USS. See `plan_flight_intent_expect_failed` in invalid_op_test_steps.py. ## Plan should fail check diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/validate_sharing_operational_intent_but_with_invalid_interuss_data.md b/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/validate_sharing_operational_intent_but_with_invalid_interuss_data.md index e13318769e..8511b562ae 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/validate_sharing_operational_intent_but_with_invalid_interuss_data.md +++ b/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/test_steps/validate_sharing_operational_intent_but_with_invalid_interuss_data.md @@ -1,6 +1,6 @@ # Validate flight sharing invalid data test step -This step verifies that a created flight is shared properly per ASTM F3548-21 by querying the DSS for flights in the area of the flight intent, and then retrieving the details from the USS if the operational intent reference is found. See `validate_shared_operational_intent` in [test_steps.py](invalid_op_test_steps.py). +This step verifies that a created flight is shared properly per ASTM F3548-21 by querying the DSS for flights in the area of the flight intent, and then retrieving the details from the USS if the operational intent reference is found. See `expect_shared_with_invalid_data` in invalid_op_test_steps.py. ## DSS responses check