Check for constraints

aas_test_engines/result.py

@@ -1,4 +1,4 @@
-from typing import List, TypeVar
+from typing import List, TypeVar, Union
 from enum import Enum
 import os
 import html
@@ -137,18 +137,25 @@ class ResultException(Exception):
     def __init__(self, result) -> None:
         self.result = result
 
+def _as_result(message: Union[str, AasTestResult], level: Level):
+    if isinstance(message, AasTestResult):
+        return message
+    assert isinstance(message, str)
+    return AasTestResult(message, level=level)
 
-def write(message: str, level=Level.INFO):
+
+def write(message: Union[str, AasTestResult]):
     if not managers:
         raise RuntimeError("No open context")
-    result = AasTestResult(message, level=level)
-    managers[-1].result.append(result)
+    message = _as_result(message, Level.INFO)
+    managers[-1].result.append(message)
 
 
-def start(message: str, level=Level.INFO) -> ContextManager:
-    result = AasTestResult(message, level=level)
+def start(message: Union[str, AasTestResult]) -> ContextManager:
+    result = _as_result(message, Level.INFO)
     return ContextManager(result)
 
 
-def abort(message: str, level=Level.ERROR):
-    raise ResultException(AasTestResult(message, level=level))
+def abort(message: Union[str, AasTestResult]):
+    result = _as_result(message, Level.ERROR)
+    raise ResultException(result)

aas_test_engines/test_cases/v3_0/__init__.py

@@ -1,29 +1,14 @@
 from typing import Tuple, Optional
 from aas_test_engines.result import AasTestResult
 
-from .parse import parse_concrete_object, check_constraints
+from .parse import parse_and_check_json, parse_and_check_xml
 from .adapter import JsonAdapter, XmlAdapter, AdapterPath
 from .model import Environment
 
-def _check_constraints(result_root: AasTestResult, env: Environment):
-    if result_root.ok():
-        result_constraints = AasTestResult("Check constraints")
-        check_constraints(env, result_constraints)
-        result_root.append(result_constraints)
 
 def json_to_env(value: any) -> Tuple[AasTestResult, Optional[Environment]]:
-    result_root = AasTestResult("Check")
-    result_meta_model = AasTestResult("Check meta model")
-    env: Environment = parse_concrete_object(Environment, JsonAdapter(value, AdapterPath()), result_meta_model)
-    result_root.append(result_meta_model)
-    _check_constraints(result_root, env)
-    return result_root, env
+    return parse_and_check_json(Environment, value)
 
 
 def xml_to_env(value: any) -> Tuple[AasTestResult, Optional[Environment]]:
-    result_root = AasTestResult("Check")
-    result_meta_model = AasTestResult("Check meta model")
-    env: Environment = parse_concrete_object(Environment, XmlAdapter(value, AdapterPath()), result_meta_model)
-    result_root.append(result_meta_model)
-    _check_constraints(result_root, env)
-    return result_root, env
+    return parse_and_check_xml(Environment, value)

aas_test_engines/test_cases/v3_0/api.py

@@ -20,6 +20,9 @@
 
 import base64
 
+from .model import AssetAdministrationShell, Environment
+from .parse import parse_and_check_json
+
 
 def b64urlsafe(value: str) -> str:
     return base64.urlsafe_b64encode(value.encode()).decode()
@@ -39,7 +42,7 @@ def _assert(predicate: bool, message, level: Level = Level.ERROR):
     if predicate:
         write(f'{message}: OK')
     else:
-        abort(f'{message}: Fail', level)
+        abort(AasTestResult('{message}: Fail', level))
 
 
 def _stringify_path(path: List[Union[str, int]]) -> str:
@@ -427,15 +430,15 @@ class ExecConf:
 def _check_server(exec_conf: ExecConf) -> bool:
     with start(f'Trying to reach {exec_conf.server}'):
         if exec_conf.dry:
-            write("Skipped due to dry run", Level.WARNING)
+            write(AasTestResult("Skipped due to dry run", Level.WARNING))
             return True
 
         try:
             requests.get(exec_conf.server, verify=exec_conf.verify)
             write('OK')
             return True
         except requests.exceptions.RequestException as e:
-            write('Failed to reach: {}'.format(e), Level.CRITICAL)
+            write(AasTestResult('Failed to reach: {}'.format(e), Level.CRITICAL))
             return False
 
 
@@ -456,7 +459,6 @@ def _load_spec() -> AasSpec:
 _spec = _load_spec()
 
 
-
 def _shorten(content: bytes, max_len: int = 300) -> str:
     try:
         content = content.decode()
@@ -471,15 +473,15 @@ def _get_json(response: requests.models.Response) -> dict:
     try:
         return response.json()
     except requests.exceptions.JSONDecodeError as e:
-        abort(f"Cannot decode as JSON: {e}", Level.CRITICAL)
+        abort(AasTestResult(f"Cannot decode as JSON: {e}", Level.CRITICAL))
 
 
 def _invoke(request: Request, conf: ExecConf, positive_test) -> requests.models.Response:
     prepared_request = request.build(conf.server).prepare()
     response = requests.Session().send(prepared_request, verify=conf.verify)
     write(f"Response: ({response.status_code}): {_shorten(response.content)}")
     if response.status_code >= 500:
-        abort(f"Got status code {response.status_code}", Level.CRITICAL)
+        abort(AasTestResult(f"Got status code {response.status_code}", Level.CRITICAL))
     if positive_test:
         if response.status_code < 200 or response.status_code > 299:
             abort(f"Expected status code 2xx, but got {response.status_code}")
@@ -498,6 +500,22 @@ def _invoke_and_decode(request: Request, conf: ExecConf, positive_test: bool) ->
         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:
@@ -596,7 +614,7 @@ def test_pagination(self):
         Test pagination
         """
         if self.cursor is None:
-            abort("Cannot check pagination, there must be at least 2 shells", level=Level.WARNING)
+            abort(AasTestResult("Cannot check pagination, there must be at least 2 shells", level=Level.WARNING))
         request = generate_one_valid(self.operation, self.sample_cache, {'cursor': self.cursor, 'limit': 1})
         data = _invoke_and_decode(request, self.conf, True)
         data = _lookup(data, ['result'])
@@ -711,7 +729,7 @@ def test_pagination(self):
         Check pagination
         """
         if self.cursor is None:
-            abort("Cannot check pagination, there must be at least 2 shells", level=Level.WARNING)
+            abort(AasTestResult("Cannot check pagination, there must be at least 2 shells", level=Level.WARNING))
         request = generate_one_valid(self.operation, self.sample_cache, {'aasIdentifier': b64urlsafe(self.valid_id), 'cursor': self.cursor, 'limit': 1})
         data = _invoke_and_decode(request, self.conf, True)
         data = _lookup(data, ['result'])
@@ -986,7 +1004,7 @@ class SubmodelElementBySuperpathSuite(SubmodelElementBySuperpathSuiteBase):
 
     def check_type(self, model_type: str, level: str, extent: str):
         if model_type not in self.paths:
-            abort("No such element present", level=Level.WARNING)
+            abort(AasTestResult("No such element present", level=Level.WARNING))
         id_short_path = self.paths[model_type][0]
         valid_values = self.valid_values.copy()
         valid_values['idShortPath'] = [id_short_path]
@@ -1041,7 +1059,7 @@ class SubmodelElementValueOnlyBySuperpathSuite(SubmodelElementBySuperpathSuiteBa
 
     def check_type(self, model_type: str, level: str, extent: str):
         if model_type not in self.paths:
-            abort("No such element present", level=Level.WARNING)
+            abort(AasTestResult("No such element present", level=Level.WARNING))
         id_short_path = self.paths[model_type][0]
         valid_values = self.valid_values.copy()
         valid_values['idShortPath'] = [id_short_path]
@@ -1087,7 +1105,7 @@ class SubmodelElementPathBySuperpathSuite(SubmodelElementBySuperpathSuiteBase):
 
     def check_type(self, model_type: str, level: str):
         if model_type not in self.paths:
-            abort("No such element present", level=Level.WARNING)
+            abort(AasTestResult("No such element present", level=Level.WARNING))
         id_short_path = self.paths[model_type][0]
         valid_values = self.valid_values.copy()
         valid_values['idShortPath'] = [id_short_path]
@@ -1133,7 +1151,7 @@ class SubmodelElementReferenceBySuperpathSuite(SubmodelElementBySuperpathSuiteBa
 
     def check_type(self, model_type: str):
         if model_type not in self.paths:
-            abort("No such element present", level=Level.WARNING)
+            abort(AasTestResult("No such element present", level=Level.WARNING))
         id_short_path = self.paths[model_type][0]
         valid_values = self.valid_values.copy()
         valid_values['idShortPath'] = [id_short_path]
@@ -1165,7 +1183,7 @@ class SubmodelElementReferenceBySuperpathSuite(SubmodelElementBySuperpathSuiteBa
 
     def check_type(self, model_type: str):
         if model_type not in self.paths:
-            abort("No such element present", level=Level.WARNING)
+            abort(AasTestResult("No such element present", level=Level.WARNING))
         id_short_path = self.paths[model_type][0]
         valid_values = self.valid_values.copy()
         valid_values['idShortPath'] = [id_short_path]

aas_test_engines/test_cases/v3_0/parse.py

@@ -5,7 +5,7 @@
 from aas_test_engines.result import AasTestResult, Level
 from enum import Enum
 import re
-from .adapter import AdapterPath
+from .adapter import AdapterPath, JsonAdapter, XmlAdapter
 
 
 class CheckConstraintException(Exception):
@@ -233,3 +233,21 @@ def check_constraints(obj, result: AasTestResult, path: AdapterPath = AdapterPat
                 check_constraints(i, result, path + field.name + idx)
         else:
             check_constraints(value, result, path + field.name)
+
+
+def _parse_and_check(cls, adapter: Adapter) -> Tuple[object, AasTestResult]:
+    result_root = AasTestResult("Check")
+    result_meta_model = AasTestResult("Check meta model")
+    env = parse(cls, adapter, result_meta_model)
+    result_root.append(result_meta_model)
+    if result_root.ok():
+        result_constraints = AasTestResult("Check constraints")
+        check_constraints(env, result_constraints)
+        result_root.append(result_constraints)
+    return result_root, env
+
+def parse_and_check_json(cls, value: any) -> Tuple[object, AasTestResult]:
+    return _parse_and_check(cls, JsonAdapter(value, AdapterPath()))
+
+def parse_and_check_xml(cls, value: any) -> Tuple[object, AasTestResult]:
+    return _parse_and_check(cls, XmlAdapter(value, AdapterPath()))