From 75574c614856250ace6eda81aa05b646429a11f6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micka=C3=ABl=20Misbach?= Date: Wed, 6 Nov 2024 15:09:24 +0100 Subject: [PATCH] [uss_qualifier/scenarios/utm/data_exchange_validation] Fix interpolation of operation path (#829) --- .../get_op_data_validation.py | 8 ++++++-- .../test_steps/expected_interactions_test_steps.py | 14 ++++++++------ 2 files changed, 14 insertions(+), 8 deletions(-) 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 90d1e4d2b6..75e47c2af1 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 @@ -215,7 +215,9 @@ def _plan_successfully_test_case(self, times: Dict[TimeDuringTest, Time]): op_id=OperationID.GetOperationalIntentDetails, direction=QueryDirection.Incoming, since=flight_1_planning_time, - query_params={"entity_id": flight_2_oi_ref.id}, + query_params=dict( + entityid=flight_2_oi_ref.id, + ), ) with self.check( "Expect GET request when no notification", @@ -337,7 +339,9 @@ def _plan_unsuccessfully_test_case(self, times: Dict[TimeDuringTest, Time]): op_id=OperationID.GetOperationalIntentDetails, direction=QueryDirection.Incoming, since=flight_1_planning_time, - query_params={"entity_id": flight_2_oi_ref.id}, + query_params=dict( + entityid=flight_2_oi_ref.id, + ), ) with self.check( "Expect GET request when no notification", 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 6868ff7e1e..6e78b97edf 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 @@ -123,8 +123,13 @@ def mock_uss_interactions( query_params: Optional[Dict[str, str]] = None, is_applicable: Optional[Callable[[Interaction], bool]] = None, ) -> Tuple[List[Interaction], Query]: - """Determine if mock_uss recorded an interaction for the specified operation in the specified direction.""" + """ + Determine if mock_uss recorded an interaction for the specified operation in the specified direction. + Raises: + KeyError: if query_params contains a non-existing parameter + IndexError: if query_params is missing a parameter + """ with scenario.check( "Mock USS interactions logs retrievable", [mock_uss.participant_id] ) as check: @@ -142,12 +147,9 @@ def mock_uss_interactions( op = api.OPERATIONS[op_id] - op_path = op.path if query_params is None: - query_params = {} - for m in re.findall(r"\{[^}]+\}", op_path): - param_name = m[1:-1] - op_path = op_path.replace(m, query_params.get(param_name, r"[^/]+")) + query_params = {} # avoid linting error due to immutable default argument + op_path = op.path.format(**query_params) # raises KeyError, IndexError if is_applicable is None: is_applicable = lambda i: True