Skip to content

Commit

Permalink
Return error if response is not json
Browse files Browse the repository at this point in the history
  • Loading branch information
@otto-ifak
otto-ifak committed Sep 4, 2024
1 parent d97d984 commit 2afc432
Showing 1 changed file with 25 additions and 45 deletions.
70 changes: 25 additions & 45 deletions aas_test_engines/api.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -383,6 +383,22 @@ def _make_invoke_result(request: Request) -> AasTestResult:
return AasTestResult(f"Invoke: {request.operation.method.upper()} {request.make_path()}")


def _invoke(parent_result: AasTestResult, request: Request, server: str) -> dict:
parent_result.append(_make_invoke_result(request))
response = request.execute(server)
if response.status_code != 200:
raise ApiTestSuiteException(f"Expected status code 200, but got {response.status_code}")
try:
data = response.json()
except requests.exceptions.JSONDecodeError as e:
raise ApiTestSuiteException(f"Cannot decode as JSON: {e}")

if not isinstance(data, dict):
raise ApiTestSuiteException(f"Expected an object, got {type(data)}")

return data


class ApiTestSuiteException(Exception):
pass

Expand Down Expand Up @@ -441,11 +457,7 @@ def teardown(self):
class GetAllAasTestSuite(ApiTestSuite):
def setup(self, result: AasTestResult) -> Dict[str, List[any]]:
request = generate_one_valid(self.open_api.operations["GetAllAssetAdministrationShells"], self.sample_cache, {'limit': 1})
result.append(_make_invoke_result(request))
response = request.execute(self.server)
if response.status_code != 200:
raise ApiTestSuiteException(f"Cannot look up idShort, got status {response.status_code}")
data = response.json()
data = _invoke(result, request, self.server)
self.valid_values = {
'limit': [1],
'idShort': [_lookup(data, ['result', 0, 'idShort'])],
Expand All @@ -462,11 +474,7 @@ def setup(self, result: AasTestResult) -> Dict[str, List[any]]:
class GetAasById(ApiTestSuite):
def setup(self, result: AasTestResult) -> Dict[str, List[any]]:
request = generate_one_valid(self.open_api.operations["GetAllAssetAdministrationShells"], self.sample_cache, {'limit': 1})
result.append(_make_invoke_result(request))
response = request.execute(self.server)
if response.status_code != 200:
raise ApiTestSuiteException(f"Cannot look up aasIdentifier, got status {response.status_code}")
data = response.json()
data = _invoke(result, request, self.server)
valid_id = _lookup(data, ['result', 0, 'id'])
self.valid_values = {
'aasIdentifier': [b64urlsafe(valid_id)]
Expand All @@ -476,11 +484,7 @@ def setup(self, result: AasTestResult) -> Dict[str, List[any]]:
class AasBySuperpathSuite(ApiTestSuite):
def setup(self, result: AasTestResult) -> Dict[str, List[any]]:
request = generate_one_valid(self.open_api.operations["GetAllAssetAdministrationShells"], self.sample_cache, {'limit': 1})
result.append(_make_invoke_result(request))
response = request.execute(self.server)
if response.status_code != 200:
raise ApiTestSuiteException(f"Cannot look up aasIdentifier, got status {response.status_code}")
data = response.json()
data = _invoke(result, request, self.server)
valid_id = _lookup(data, ['result', 0, 'id'])
self.valid_values = {
'aasIdentifier': [b64urlsafe(valid_id)]
Expand All @@ -490,11 +494,7 @@ def setup(self, result: AasTestResult) -> Dict[str, List[any]]:
class SubmodelBySuperpathSuite(ApiTestSuite):
def setup(self, result: AasTestResult) -> Dict[str, List[any]]:
request = generate_one_valid(self.open_api.operations["GetAllAssetAdministrationShells"], self.sample_cache, {'limit': 1})
result.append(_make_invoke_result(request))
response = request.execute(self.server)
if response.status_code != 200:
raise ApiTestSuiteException(f"Cannot look up submodelIdentifier, got status {response.status_code}")
data = response.json()
data = _invoke(result, request, self.server)
valid_id = _lookup(data, ['result', 0, 'id'])
valid_submodel_id = _lookup(data, ['result', 0, 'submodels', 0, 'keys', 0, 'value'])
self.valid_values = {
Expand Down Expand Up @@ -536,23 +536,15 @@ class SubmodelElementBySuperpathSuite(ApiTestSuite):

def setup(self, result: AasTestResult) -> Dict[str, List[any]]:
request = generate_one_valid(self.open_api.operations["GetAllAssetAdministrationShells"], self.sample_cache, {'limit': 1})
result.append(_make_invoke_result(request))
response = request.execute(self.server)
if response.status_code != 200:
raise ApiTestSuiteException(f"Cannot look up submodelIdentifier, got status {response.status_code}")
data = response.json()
data = _invoke(result, request, self.server)
valid_id = _lookup(data, ['result', 0, 'id'])
valid_submodel_id = _lookup(data, ['result', 0, 'submodels', 0, 'keys', 0, 'value'])
overwrites = {
'aasIdentifier': [b64urlsafe(valid_id)],
'submodelIdentifier': [b64urlsafe(valid_submodel_id)],
}
request = generate_one_valid(self.open_api.operations["GetAllSubmodelElements_AasRepository"], self.sample_cache, overwrites)
result.append(_make_invoke_result(request))
response = request.execute(self.server)
if response.status_code != 200:
raise ApiTestSuiteException(f"Cannot look up idShortPath, got status {response.status_code}")
data = response.json()
data = _invoke(result, request, self.server)
elements = _lookup(data, ['result'])
self.paths = {}
_collect_submodel_elements(elements, self.paths, '')
Expand Down Expand Up @@ -596,23 +588,15 @@ def execute(self, result_positive: AasTestResult, result_negative: AasTestResult
class GetFileByPathSuperpathSuite(ApiTestSuite):
def setup(self, result: AasTestResult) -> Dict[str, List[any]]:
request = generate_one_valid(self.open_api.operations["GetAllAssetAdministrationShells"], self.sample_cache, {'limit': 1})
result.append(_make_invoke_result(request))
response = request.execute(self.server)
if response.status_code != 200:
raise ApiTestSuiteException(f"Cannot look up submodelIdentifier, got status {response.status_code}")
data = response.json()
data = _invoke(result, request, self.server)
valid_id = _lookup(data, ['result', 0, 'id'])
valid_submodel_id = _lookup(data, ['result', 0, 'submodels', 0, 'keys', 0, 'value'])
overwrites = {
'aasIdentifier': [b64urlsafe(valid_id)],
'submodelIdentifier': [b64urlsafe(valid_submodel_id)],
}
request = generate_one_valid(self.open_api.operations["GetAllSubmodelElements_AasRepository"], self.sample_cache, overwrites)
result.append(_make_invoke_result(request))
response = request.execute(self.server)
if response.status_code != 200:
raise ApiTestSuiteException(f"Cannot look up idShortPath, got status {response.status_code}")
data = response.json()
data = _invoke(result, request, self.server)
paths = {}
_collect_submodel_elements(_lookup(data, ['result']), paths, '')
try:
Expand All @@ -625,11 +609,7 @@ def setup(self, result: AasTestResult) -> Dict[str, List[any]]:
class GenerateSerializationSuite(ApiTestSuite):
def setup(self, result: AasTestResult) -> Dict[str, List[any]]:
request = generate_one_valid(self.open_api.operations["GetAllAssetAdministrationShells"], self.sample_cache, {'limit': 1})
result.append(_make_invoke_result(request))
response = request.execute(self.server)
if response.status_code != 200:
raise ApiTestSuiteException(f"Cannot look up submodelIdentifier, got status {response.status_code}")
data = response.json()
data = _invoke(result, request, self.server)
valid_id = _lookup(data, ['result', 0, 'id'])
valid_submodel_id = _lookup(data, ['result', 0, 'submodels', 0, 'keys', 0, 'value'])
self.valid_values = {
Expand Down

0 comments on commit 2afc432

Please sign in to comment.