From cafabe8bd03eb86121e84cb63a05acd7bef4a902 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Otto?= Date: Mon, 13 Jan 2025 19:26:28 +0100 Subject: [PATCH] Stop test execution if api response does not conform to schema (fixes #78) --- aas_test_engines/result.py | 12 +++-- aas_test_engines/test_cases/v3_0/api.py | 66 ++++++++++++------------- 2 files changed, 42 insertions(+), 36 deletions(-) diff --git a/aas_test_engines/result.py b/aas_test_engines/result.py index 6507b37..f373b45 100644 --- a/aas_test_engines/result.py +++ b/aas_test_engines/result.py @@ -110,8 +110,9 @@ def from_json(self, data: dict) -> "AasTestResult": class ContextManager: - def __init__(self, result: AasTestResult): + def __init__(self, result: AasTestResult, catch_all_exceptions: bool): self.result = result + self.catch_all_exceptions = catch_all_exceptions def __enter__(self) -> AasTestResult: managers.append(self) @@ -128,6 +129,11 @@ def __exit__(self, exc_type, exc_val, traceback): if managers: managers[-1].result.append(self.result) return True + elif self.catch_all_exceptions: + self.result.append(AasTestResult(f"Internal error: {exc_val}", Level.CRITICAL)) + if managers: + managers[-1].result.append(self.result) + return True else: return False @@ -151,9 +157,9 @@ def write(message: Union[str, AasTestResult]): managers[-1].result.append(message) -def start(message: Union[str, AasTestResult]) -> ContextManager: +def start(message: Union[str, AasTestResult], catch_all_exceptions: bool = False) -> ContextManager: result = _as_result(message, Level.INFO) - return ContextManager(result) + return ContextManager(result, catch_all_exceptions) def abort(message: Union[str, AasTestResult]): diff --git a/aas_test_engines/test_cases/v3_0/api.py b/aas_test_engines/test_cases/v3_0/api.py index 4af50b7..8b6b3a1 100644 --- a/aas_test_engines/test_cases/v3_0/api.py +++ b/aas_test_engines/test_cases/v3_0/api.py @@ -508,39 +508,39 @@ def _invoke(request: Request, conf: ExecConf, positive_test: bool) -> requests.m def _invoke_and_decode(request: Request, conf: ExecConf, positive_test: bool) -> dict: - with start(f"Invoke: {request.operation.method.upper()} {request.make_path()}"): - response = _execute(request, conf, positive_test) - expected_responses = [] - expected_responses += [i for i in request.operation.responses if i.code == response.status_code] - expected_responses += [i for i in request.operation.responses if i.code is None] - if not expected_responses: - abort(f"Invalid status code {response.status_code}") - data = _get_json(response) - ref = expected_responses[0].schema.get('$ref') - if ref == '#/components/schemas/AssetAdministrationShell': - result, aas = parse_and_check_json(AssetAdministrationShell, data) - write(result) - return data - if ref == '#/components/schemas/Environment': - result, aas = parse_and_check_json(Environment, data) - write(result) - return data - if ref == '#/components/schemas/GetAssetAdministrationShellsResult': - entries = data.get('result', None) - if entries: - result, aas = parse_and_check_json(List[AssetAdministrationShell], entries) - write(result) - # Fallthrough to check against schema for paging_metadata - - validator = parse_schema({**expected_responses[0].schema, '$schema': 'https://json-schema.org/draft/2020-12/schema'}, ParseConfig(raise_on_unknown_format=False)) - validation_result = validator.validate(data) - if validation_result.ok: - write("Response conforms to schema") - else: - result = AasTestResult(f"Invalid response for schema", level=Level.ERROR) - _map_error(result, validation_result) - raise ResultException(result) + write(f"Invoke: {request.operation.method.upper()} {request.make_path()}") + response = _execute(request, conf, positive_test) + expected_responses = [] + expected_responses += [i for i in request.operation.responses if i.code == response.status_code] + expected_responses += [i for i in request.operation.responses if i.code is None] + if not expected_responses: + abort(f"Invalid status code {response.status_code}") + data = _get_json(response) + ref = expected_responses[0].schema.get('$ref') + if ref == '#/components/schemas/AssetAdministrationShell': + result, aas = parse_and_check_json(AssetAdministrationShell, data) + write(result) + return data + if ref == '#/components/schemas/Environment': + result, aas = parse_and_check_json(Environment, data) + write(result) return data + if ref == '#/components/schemas/GetAssetAdministrationShellsResult': + entries = data.get('result', None) + if entries: + result, aas = parse_and_check_json(List[AssetAdministrationShell], entries) + write(result) + # Fallthrough to check against schema for paging_metadata + + validator = parse_schema({**expected_responses[0].schema, '$schema': 'https://json-schema.org/draft/2020-12/schema'}, ParseConfig(raise_on_unknown_format=False)) + validation_result = validator.validate(data) + if validation_result.ok: + write("Response conforms to schema") + else: + result = AasTestResult(f"Invalid response for schema", level=Level.ERROR) + _map_error(result, validation_result) + raise ResultException(result) + return data class ApiTestSuite: @@ -1940,7 +1940,7 @@ def execute_tests(conf: ExecConf, suite: str) -> Tuple[AasTestResult, ConfusionM for operation in _spec.open_api.operations.values(): if operation.operation_id not in operation_ids: continue - with start(f"Checking {operation.method.upper()} {operation.path} ({operation.operation_id})"): + with start(f"Checking {operation.method.upper()} {operation.path} ({operation.operation_id})", True): if conf.dry: continue