diff --git a/monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py b/monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py index 692dcf9c8f..2f9f3699c6 100644 --- a/monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py +++ b/monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py @@ -83,14 +83,9 @@ 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}" - query = fetch.query_and_describe( - self.client, - "GET", - url, - QueryType.F3548v21USSGetOperationalIntentDetails, + result, query = self.get_full_op_intent_without_validation( + op_intent_ref, uss_participant_id, - scope=SCOPE_SC, ) if query.status_code != 200: result = None @@ -101,19 +96,26 @@ def get_full_op_intent( return result, query def get_full_op_intent_without_validation( - self, op_intent_ref: OperationalIntentReference + self, + op_intent_ref: OperationalIntentReference, + uss_participant_id: Optional[str] = None, ) -> Tuple[Dict, fetch.Query]: """ GET OperationalIntent without validating, as invalid data expected for negative tests Args: op_intent_ref: - + uss_participant_id: Returns: returns the response json when query is successful """ url = f"{op_intent_ref.uss_base_url}/uss/v1/operational_intents/{op_intent_ref.id}" query = fetch.query_and_describe( - self.client, "GET", url, scope=SCOPE_SC, participant_id=self.participant_id + self.client, + "GET", + url, + QueryType.F3548v21USSGetOperationalIntentDetails, + uss_participant_id, + scope=SCOPE_SC, ) result = None if query.status_code == 200: diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/get_op_data_validation.py b/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/get_op_data_validation.py index 4ff7a59e54..8f9611abfa 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/get_op_data_validation.py +++ b/monitoring/uss_qualifier/scenarios/astm/utm/data_exchange_validation/get_op_data_validation.py @@ -1,13 +1,11 @@ from typing import Optional, Dict - from monitoring.monitorlib.clients.flight_planning.flight_info_template import ( FlightInfoTemplate, ) -from monitoring.monitorlib.temporal import TimeDuringTest, Time +from monitoring.monitorlib.temporal import TimeDuringTest from urllib.parse import urlsplit import arrow from implicitdict import StringBasedDateTime -from datetime import datetime from monitoring.monitorlib.temporal import Time from monitoring.monitorlib.clients.flight_planning.client import FlightPlannerClient @@ -26,7 +24,6 @@ MockUSSResource, ) from monitoring.uss_qualifier.scenarios.astm.utm.data_exchange_validation.test_steps.invalid_op_test_steps import ( - InvalidOpIntentSharingValidator, plan_flight_intent_expect_failed, ) from monitoring.uss_qualifier.scenarios.astm.utm.test_steps import OpIntentValidator @@ -148,6 +145,8 @@ def _tested_uss_plans_deconflicted_flight_near_existing_flight( ): times[TimeDuringTest.TimeOfEvaluation] = Time(arrow.utcnow().datetime) flight_2 = self.flight_2.resolve(times) + + planning_time = Time(arrow.utcnow().datetime) with OpIntentValidator( self, self.control_uss_client, @@ -164,17 +163,17 @@ def _tested_uss_plans_deconflicted_flight_near_existing_flight( flight_2_oi_ref = validator.expect_shared(flight_2) - precondition_no_post_interaction( - self, + self._precondition_no_post_interaction( self.control_uss, - times[TimeDuringTest.TimeOfEvaluation], - self.tested_uss_client.get_base_url(), + planning_time, + self._get_domain(self.tested_uss_client.get_base_url()), "Precondition - check tested_uss has no subscription in flight 2 area", ) times[TimeDuringTest.TimeOfEvaluation] = Time(arrow.utcnow().datetime) flight_1 = self.flight_1.resolve(times) + planning_time = Time(arrow.utcnow().datetime) with OpIntentValidator( self, self.tested_uss_client, @@ -192,13 +191,11 @@ def _tested_uss_plans_deconflicted_flight_near_existing_flight( flight_1, ) - control_uss_domain = "{0.scheme}://{0.netloc}/".format( - urlsplit(self.control_uss.base_url) - ) + control_uss_domain = self.control_uss.base_url expect_interuss_get_interactions( self, self.control_uss, - times[TimeDuringTest.TimeOfEvaluation], + planning_time, control_uss_domain, flight_2_oi_ref.id, "Validate flight2 GET interaction", @@ -206,7 +203,7 @@ def _tested_uss_plans_deconflicted_flight_near_existing_flight( expect_interuss_post_interactions( self, self.control_uss, - times[TimeDuringTest.TimeOfEvaluation], + planning_time, control_uss_domain, "Validate flight1 Notification sent to Control_uss", ) @@ -235,7 +232,8 @@ def _tested_uss_unable_to_plan_flight_near_invalid_shared_existing_flight( additional_fields = {"behavior": behavior} - with InvalidOpIntentSharingValidator( + planning_time = Time(arrow.utcnow().datetime) + with OpIntentValidator( self, self.control_uss_client, self.dss, @@ -251,17 +249,17 @@ def _tested_uss_unable_to_plan_flight_near_invalid_shared_existing_flight( ) flight_2_oi_ref = validator.expect_shared_with_invalid_data(flight_info) - precondition_no_post_interaction( - self, + self._precondition_no_post_interaction( self.control_uss, - times[TimeDuringTest.TimeOfEvaluation], - self.tested_uss_client.get_base_url(), + planning_time, + self._get_domain(self.tested_uss_client.get_base_url()), "Precondition - check tested_uss has no subscription in flight 2 area", ) times[TimeDuringTest.TimeOfEvaluation] = Time(arrow.utcnow().datetime) flight_1 = self.flight_1.resolve(times) - with InvalidOpIntentSharingValidator( + planning_time = Time(arrow.utcnow().datetime) + with OpIntentValidator( self, self.tested_uss_client, self.dss, @@ -276,13 +274,11 @@ def _tested_uss_unable_to_plan_flight_near_invalid_shared_existing_flight( ) validator.expect_not_shared() - control_uss_domain = "{0.scheme}://{0.netloc}/".format( - urlsplit(self.control_uss.base_url) - ) + control_uss_domain = self.control_uss.base_url expect_interuss_get_interactions( self, self.control_uss, - times[TimeDuringTest.TimeOfEvaluation], + planning_time, control_uss_domain, flight_2_oi_ref.id, "Validate flight 2 GET interaction", @@ -290,7 +286,7 @@ def _tested_uss_unable_to_plan_flight_near_invalid_shared_existing_flight( expect_no_interuss_post_interactions( self, self.control_uss, - times[TimeDuringTest.TimeOfEvaluation], + planning_time, control_uss_domain, "Validate flight 1 Notification not sent to Control_uss", ) @@ -299,6 +295,26 @@ def _tested_uss_unable_to_plan_flight_near_invalid_shared_existing_flight( self, "Delete Control_uss flight", self.control_uss_client, self.flight_2_id ) + def _precondition_no_post_interaction( + self, + mock_uss: MockUSSClient, + st: StringBasedDateTime, + posted_to_url: str, + test_step: str, + ): + self.begin_test_step(test_step) + + mock_uss_sent_notification = precondition_no_post_interaction( + self, mock_uss, st, posted_to_url + ) + if mock_uss_sent_notification: + msg = f"As a precondition for the scenario tests, there should have been no post made to {posted_to_url}" + raise ScenarioCannotContinueError(msg) + self.end_test_step() + + def _get_domain(self, url): + return "{0.scheme}://{0.netloc}/".format(urlsplit(url)) + def cleanup(self): self.begin_cleanup() cleanup_flights_fp_client( 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 index 455c277e4b..d0083de576 100644 --- 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 @@ -11,9 +11,6 @@ from implicitdict import StringBasedDateTime from loguru import logger from monitoring.monitorlib.clients.mock_uss.interactions import Interaction -from monitoring.uss_qualifier.scenarios.scenario import ( - ScenarioCannotContinueError, -) def expect_interuss_post_interactions( @@ -23,17 +20,30 @@ def expect_interuss_post_interactions( posted_to_url: str, test_step: str, ): + """ + This step checks if a notification was sent to a subscribed USS, from time 'st' to now + Args: + scenario: + mock_uss: + st: + posted_to_url: url of the subscribed USS + test_step: + + Returns: + + """ scenario.begin_test_step(test_step) + interactions, query = _get_interuss_interactions_with_check(scenario, mock_uss, st) + logger.debug(f"Checking for POST request to {posted_to_url}") + found = any_post_interactions_to_url(interactions, posted_to_url) with scenario.check("Expect Notification sent") as check: - interactions = _get_interuss_interactions_with_check(scenario, mock_uss, st) - logger.debug(f"Checking for POST request to {posted_to_url}") - found = any_post_interactions_to_url(interactions, posted_to_url) - if found == False: + if not found: check.record_failed( - summary=f"Notification to {posted_to_url} not received", + summary=f"Notification to {posted_to_url} not sent", severity=Severity.Medium, - details=f"Notification to {posted_to_url} not received", - requirements="SCDxxxx", + details=f"Notification to {posted_to_url} not sent", + requirements="SCD0085", + query_timestamps=[query.request.timestamp], ) scenario.end_test_step() @@ -45,17 +55,30 @@ def expect_no_interuss_post_interactions( posted_to_url: str, test_step: str, ): + """ + This step checks no notification was sent to any USS as no DSS entity was created, from time 'st' to now + Args: + scenario: + mock_uss: + st: + posted_to_url: + test_step: + + Returns: + + """ scenario.begin_test_step(test_step) + interactions, query = _get_interuss_interactions_with_check(scenario, mock_uss, st) + logger.debug(f"Checking for POST request to {posted_to_url}") + found = any_post_interactions_to_url(interactions, posted_to_url) with scenario.check("Expect Notification not sent") as check: - interactions = _get_interuss_interactions_with_check(scenario, mock_uss, st) - logger.debug(f"Checking for POST request to {posted_to_url}") - found = any_post_interactions_to_url(interactions, posted_to_url) - if found == True: + if found: check.record_failed( - summary=f"Notification to {posted_to_url} wrongly sent", + summary=f"Notification to {posted_to_url} wrongly sent for an entity not created.", severity=Severity.Medium, - details=f"Notification to {posted_to_url} wrongly sent", - requirements="SCDxxxx", + details=f"Notification to {posted_to_url} wrongly sent for an entity not created.", + requirements="interuss.f3548.notification_requirements.NoDssEntityNoNotification", + query_timestamps=[query.request.timestamp], ) scenario.end_test_step() @@ -68,27 +91,37 @@ def expect_interuss_get_interactions( id: str, test_step: str, ): + """ + This step checks a GET request to a USS was made for an existing entity, from time 'st' to now + Args: + scenario: + mock_uss: + st: + get_from_url: USS managing the entity + id: entity id + test_step: + + Returns: + + """ scenario.begin_test_step(test_step) - if mock_uss is None: - raise ScenarioCannotContinueError("mock_uss should not be None.") - - interactions = _get_interuss_interactions_with_check(scenario, mock_uss, st) + interactions, query = _get_interuss_interactions_with_check(scenario, mock_uss, st) logger.debug(f"Checking for GET request to {get_from_url} for id {id}") + found = False + for interaction in interactions: + method = interaction.query.request.method + url = interaction.query.request.url + if method == "GET" and url.startswith(get_from_url) and id in url: + found = True 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: + if not found: 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", + requirements="SCD0035", + query_timestamps=[query.request.timestamp], ) - scenario.end_test_step() @@ -96,12 +129,22 @@ def _get_interuss_interactions_with_check( scenario: TestScenarioType, mock_uss: MockUSSClient, st: StringBasedDateTime, -) -> List[Interaction]: +) -> Tuple[List[Interaction], Query]: + """ + Method to get interuss interactions with a scenario check from mock_uss from time 'st' to now. + Args: + scenario: + mock_uss: + st: + + Returns: + """ with scenario.check("MockUSS interactions request") as check: try: interactions, query = _get_interuss_interactions(mock_uss, st) - return interactions + scenario.record_query(query) + return interactions, query except QueryError as e: for q in e.queries: scenario.record_query(q) @@ -117,24 +160,34 @@ def _get_interuss_interactions( mock_uss: MockUSSClient, st: StringBasedDateTime, ) -> Tuple[List[Interaction], Query]: + """ + Method to get interuss interactions from mock_uss from time 'st' to now. + Args: + mock_uss: + st: + + Returns: + + """ + # Wait - To make sure that interuss interactions are received and recorded + # Using a guess value of 5 seconds time.sleep(5) - if mock_uss is None: - raise ScenarioCannotContinueError("mock_uss should not be None.") all_interactions, query = mock_uss.get_interactions(st) exclude_sub = mock_uss.session.auth_adapter.get_sub() + def get_client_sub(headers): + token = headers.get("Authorization").split(" ")[1] + payload = jwt.decode( + token, algorithms="RS256", options={"verify_signature": False} + ) + return payload["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}") + sub = get_client_sub(headers) if sub == excl_sub: - logger.debug(f"Excluding interaction with sub: {sub} ") return False else: return True @@ -159,17 +212,10 @@ def precondition_no_post_interaction( mock_uss: MockUSSClient, st: StringBasedDateTime, posted_to_url: str, - test_step: str, -): - scenario.begin_test_step(test_step) - interactions, _ = _get_interuss_interactions(mock_uss, st) - logger.debug(f"Checking for POST request to {posted_to_url}") - found = any_post_interactions_to_url(interactions, posted_to_url) - if found: - msg = f"As a precondition for the scenario tests, there should have been no post made to {posted_to_url}" - raise ScenarioCannotContinueError(msg) - - scenario.end_test_step() +) -> bool: + interactions, query = _get_interuss_interactions(mock_uss, st) + scenario.record_query(query) + return any_post_interactions_to_url(interactions, posted_to_url) def any_post_interactions_to_url( @@ -179,6 +225,6 @@ def any_post_interactions_to_url( 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 method == "POST" and url.startswith(posted_to_url): + return True return found 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 index ec02dc9da1..ed65613779 100644 --- 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 @@ -1,23 +1,9 @@ from typing import Optional, Tuple -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, - InjectFlightResponse, -) -from monitoring.uss_qualifier.common_data_definitions import Severity from monitoring.uss_qualifier.resources.flight_planning.flight_planner import ( FlightPlannerClient, ) -from monitoring.uss_qualifier.scenarios.astm.utm.test_steps import OpIntentValidator from monitoring.uss_qualifier.scenarios.flight_planning.test_steps import ( submit_flight, - expect_flight_intent_state, ) from monitoring.uss_qualifier.scenarios.scenario import TestScenarioType from monitoring.monitorlib.clients.flight_planning.flight_info import FlightInfo @@ -51,134 +37,3 @@ def plan_flight_intent_expect_failed( flight_planner, flight_intent, ) - - -class InvalidOpIntentSharingValidator(OpIntentValidator): - def expect_shared_with_invalid_data( - self, 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 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: - 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 standard req validation errors - 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 = [] - 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/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 38473df15c..9657bf47ef 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 @@ -6,7 +6,7 @@ This step verifies that a created flight is shared properly per ASTM F3548-21 by **[astm.f3548.v21.DSS0005](../../../../../requirements/astm/f3548/v21.md)** -## Operational intent shared with DSS check +## Operational intent shared correctly check If a reference to the operational intent for the flight is not found in the DSS, this check will fail per **[astm.f3548.v21.USS0005](../../../../../requirements/astm/f3548/v21.md)** and **[astm.f3548.v21.OPIN0025](../../../../../requirements/astm/f3548/v21.md)**. diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/prep_planners.py b/monitoring/uss_qualifier/scenarios/astm/utm/prep_planners.py index a36ccbc96b..2b8ad5fa79 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/prep_planners.py +++ b/monitoring/uss_qualifier/scenarios/astm/utm/prep_planners.py @@ -30,8 +30,8 @@ def __init__( ): super(PrepareFlightPlanners, self).__init__( flight_planners, - mock_uss, flight_intents, + mock_uss, flight_intents2, flight_intents3, flight_intents4, diff --git a/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py b/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py index df805e9704..ca2baadc3e 100644 --- a/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py +++ b/monitoring/uss_qualifier/scenarios/astm/utm/test_steps.py @@ -1,7 +1,7 @@ from __future__ import annotations from typing import List, Optional, Union - +from implicitdict import ImplicitDict from monitoring.monitorlib import schema_validation, fetch from monitoring.monitorlib.clients.flight_planning.client import FlightPlannerClient from monitoring.monitorlib.geotemporal import Volume4DCollection @@ -9,8 +9,8 @@ OperationalIntentState, Volume4D, OperationalIntentReference, + GetOperationalIntentDetailsResponse, ) - from monitoring.monitorlib.clients.flight_planning.flight_info import ( UasState, AirspaceUsageState, @@ -159,103 +159,13 @@ def expect_shared( :returns: the shared operational intent reference. None if skipped because not found. """ - self._begin_step() - - with self._scenario.check( - "Operational intent shared correctly", [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 ( - (isinstance(flight_intent, InjectFlightRequest)) - and ( - flight_intent.operational_intent.state - == OperationalIntentState.Activated - ) - ) or ( - isinstance(flight_intent, FlightInfo) - and ( - ( - flight_intent.basic_information.uas_state - == UasState.Nominal - ) - and ( - flight_intent.basic_information.usage_state - == AirspaceUsageState.InUse - ) - ) - ): - 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 + oi_ref = self._operational_intent_shared_check(flight_intent, skip_if_not_found) 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) - 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], - ) + self._operational_intent_retrievable_check(oi_full_query) with self._scenario.check( "Operational intent details data format", @@ -346,4 +256,179 @@ def volume_vertices(v4): ) self._scenario.end_test_step() + return oi_full.reference + + def expect_shared_with_invalid_data( + self, + flight_intent: Union[InjectFlightRequest, FlightInfo], + 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 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. + """ + + oi_ref = self._operational_intent_shared_check(flight_intent, skip_if_not_found) + + 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) + + # validation errors expected check + 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 = [] + 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: + 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 + + def _operational_intent_retrievable_check(self, 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], + ) + + def _operational_intent_shared_check( + self, + flight_intent: Union[InjectFlightRequest | FlightInfo], + skip_if_not_found: bool, + ) -> OperationalIntentReference: + + self._begin_step() + + with self._scenario.check( + "Operational intent shared correctly", [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 ( + (isinstance(flight_intent, InjectFlightRequest)) + and ( + flight_intent.operational_intent.state + == OperationalIntentState.Activated + ) + ) or ( + isinstance(flight_intent, FlightInfo) + and ( + ( + flight_intent.basic_information.uas_state + == UasState.Nominal + ) + and ( + flight_intent.basic_information.usage_state + == AirspaceUsageState.InUse + ) + ) + ): + 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 + return oi_ref diff --git a/monitoring/uss_qualifier/scenarios/flight_planning/prep_planners.py b/monitoring/uss_qualifier/scenarios/flight_planning/prep_planners.py index 59e893be32..f767574feb 100644 --- a/monitoring/uss_qualifier/scenarios/flight_planning/prep_planners.py +++ b/monitoring/uss_qualifier/scenarios/flight_planning/prep_planners.py @@ -31,8 +31,8 @@ class PrepareFlightPlanners(TestScenario): def __init__( self, flight_planners: FlightPlannersResource, - mock_uss: MockUSSResource, flight_intents: FlightIntentsResource, + mock_uss: Optional[MockUSSResource] = None, flight_intents2: Optional[FlightIntentsResource] = None, flight_intents3: Optional[FlightIntentsResource] = None, flight_intents4: Optional[FlightIntentsResource] = None, @@ -63,9 +63,10 @@ def __init__( self.flight_planners = { fp.participant_id: fp.client for fp in flight_planners.flight_planners } - self.flight_planners.update( - {mock_uss.mock_uss.participant_id: mock_uss.mock_uss.flight_planner} - ) + if mock_uss is not None: + self.flight_planners.update( + {mock_uss.mock_uss.participant_id: mock_uss.mock_uss.flight_planner} + ) def run(self, context): self.begin_test_scenario(context) diff --git a/monitoring/uss_qualifier/suites/uspace/flight_auth.yaml b/monitoring/uss_qualifier/suites/uspace/flight_auth.yaml index d4d4fd04cc..95ff367c56 100644 --- a/monitoring/uss_qualifier/suites/uspace/flight_auth.yaml +++ b/monitoring/uss_qualifier/suites/uspace/flight_auth.yaml @@ -27,8 +27,8 @@ actions: scenario_type: scenarios.flight_planning.PrepareFlightPlanners resources: flight_planners: flight_planners - mock_uss: mock_uss flight_intents: invalid_flight_auth_flights + mock_uss: mock_uss on_failure: Abort - action_generator: generator_type: action_generators.flight_planning.FlightPlannerCombinations