Newer
Older
from http import HTTPStatus
class Checks:
def __init__(self):
self.checks = {
'Check Response Status Code':
Checks.check_response_status_code,
'Check Response Body Containing Array Of URIs set to':
Checks.check_response_body_containing_array_of_uris_set_to,
'Check Created Resources Set To':
Checks.check_created_resources_set_to,
'Check Response Headers Containing Content-Type set to':
Checks.check_response_headers_containing_content_type_set_to,
'Check Response Headers Link Not Empty':
Checks.check_response_headers_link_not_empty,
'Check Response Headers Containing URI set to':
Checks.check_response_headers_containing_uri_set_to,
'Check Response Headers ID Not Empty':
Checks.check_response_headers_id_not_empty,
'Check Response Body Containing an Attribute set to':
Checks.check_response_body_containing_an_attribute_set_to,
'Check Response Body Containing Entity element':
Checks.check_response_body_containing_entity_element,
'Check Response Body Containing List Containing Entity Elements':
Checks.check_response_body_containing_list_containing_entity_elements,
'Check Response Body Containing List Containing Entity Elements With Different Types':
Checks.check_response_body_containing_list_containing_entity_elements_with_different_types,
'Check Response Body Containing EntityTemporal element':
Checks.check_response_body_containing_entitytemporal_element,
'Check Response Body Containing List Containing EntityTemporal elements':
Checks.check_response_body_containing_list_containing_entitytemporal_elements,
'Check Response Body Containing Subscription element':
Checks.check_response_body_containing_subscription_element,
'Check Response Body Containing List Containing Subscription elements':
Checks.check_response_body_containing_list_containing_subscription_elements,
'Check Response Body Containing Number Of Entities':
Checks.check_response_body_containing_number_of_entities,
'Check Response Body Containing Context Source Registration element':
Checks.check_response_body_containing_context_source_registration_element,
'Check Response Body Containing EntityTypeList element':
Checks.check_response_body_containing_entitytypelist_element,
'Check Response Body Containing EntityType element':
Checks.check_response_body_containing_entitytype_element,
'Check Response Body Containing EntityTypeInfo element':
Checks.check_response_body_containing_entitytypeinfo_element,
'Check Response Body Containing AttributeList element':
Checks.check_response_body_containing_attributelist_element,
'Check Response Body Containing Attribute element':
Checks.check_response_body_containing_attribute_element,
'Check Response Body Containing List Containing Context Source Registrations elements':
Checks.check_response_body_containing_list_containing_context_source_registrations_elements,
'Check Response Body Type When Using Session Request':
Checks.check_response_body_type_when_using_session_request,
'Check Response Body Containing ProblemDetails Element Containing Type Element set to':
Checks.check_response_body_containing_problemdetails_element_containing_type_element_set_to,
'Check Response Body Title When Using Session Request':
Checks.check_response_body_title_when_using_session_request,
'Check Response Body Containing ProblemDetails Element Containing Title Element':
Checks.check_response_body_containing_problemdetails_element_containing_title_element,
'Check Response Body Containing ProblemDetails Element':
Checks.check_response_body_containing_problemdetails_element,
'Check RL Response Body Containing ProblemDetails Element Containing Type Element set to':
Checks.check_rl_response_body_containing_problemdetails_element_containing_type_element_set_to,
'Check RL Response Body Containing ProblemDetails Element Containing Title Element':
Checks.check_rl_response_body_containing_problemdetails_element_containing_title_element,
'Check JSON Value In Response Body':
Checks.check_json_value_in_response_body,
'Check Pagination Prev And Next Headers':
Checks.check_pagination_prev_and_next_headers,
'Check Created Resource Set To':
Checks.check_created_resource_set_to,
'Check Updated Resource Set To':
Checks.check_updated_resource_set_to,
'Check Updated Resources Set To':
Checks.check_updated_resources_set_to,
Checks.check_sut_not_containing_resource,
'Check SUT Not Containing Resources':
Checks.check_sut_not_containing_resources,
'Check NotificationParams':
Checks.check_notificationparams,
'Check Response Body Containing Batch Operation Result':
Checks.check_response_body_containing_batch_operation_result,
'Wait for no notification':
Checks.wait_for_no_notification,
'Wait for notification and validate it':
Checks.wait_for_notification_and_validate_it,
'Dictionary Should Contain Key':
Checks.dictionary_should_contain_key,
'Should Not Be Empty':
Checks.should_not_be_empty,
'Should be True':
Checks.should_be_true,
Checks.check_response_body_content,
'Check Retrieving Context Source Registration':
lopezaguilar
committed
Checks.check_retrieving_context_source_registration,
'Check JSON Value Not In Response Body':
Checks.check_json_value_not_in_response_body,
'Check Response Reason set to':
Checks.check_response_reason_set_to,
'Check Response Does Not Contain Body':
Checks.check_response_does_not_contain_body,
'Check Response Headers Containing NGSILD-Results-Count Equals To' :
Checks.check_response_header_contains_ngsild_results_count_equals_to,
'Check Response Status Code': {
'params': ['status_code'],
'Check Response Body Containing Array Of URIs set to': {
'params': ['expected_entities_ids', 'response_body'],
'position': [0, 1]
},
'Check Response Body Containing ProblemDetails Element Containing Type Element set to': {
'params': ['type'],
'Check Response Body Containing ProblemDetails Element': {
'params': ['problem_type'],
'position': [1]
},
'Check Response Headers Containing Content-Type set to': {
'params': ['content_type'],
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
'Check Updated Resource Set To': {
'params': ['updated_resource', 'response_body', 'ignored_keys'],
'position': [0]
},
'Check Response Body Containing ProblemDetails Element Containing Title Element': {
'params': ['response_body'],
'position': [0]
},
'Check Response Headers Containing URI set to': {
'params': ['expected_entity_id', 'response_headers'],
'position': [0, 1]
},
'Check Response Body Title When Using Session Request': {
'params': ['response_body'],
'position': [0]
},
'Check Response Body Containing EntityTypeInfo element': {
'params': ['expectation_filename', 'response_body'],
'position': [0, 1]
},
'Check Response Body Containing List Containing Entity Elements': {
'params': ['expectation_filename', 'entities_ids', 'response_body', 'ignore_core_context_version'],
'position': []
},
'Check Response Body Containing List Containing Entity Elements With Different Types': {
'params': ['filename', 'entities_representation_ids', 'response_body', 'ignore_core_context_version'],
'position': []
},
'Check Response Body Containing an Attribute set to': {
'params': ['expected_attribute_name', 'response_body', 'expected_attribute_value'],
'position': []
'Check Response Body Containing Attribute element': {
'params': ['expectation_filename', 'response_body'],
'position': [0, 1]
},
'Check Response Body Containing EntityTemporal element': {
'params': ['filename', 'temporal_entity_representation_id', 'response_body'],
'position': [0, 1, 2]
},
'Check SUT Not Containing Resources': {
'params': ['response_body'],
'position': [0]
},
'Check Response Body Containing List Containing EntityTemporal elements': {
'params': ['filename', 'entity_ids'],
'Check RL Response Body Containing ProblemDetails Element Containing Title Element': {
'params': ['response_body'],
'position': [0]
},
'Check Response Body Containing List Containing Subscription elements': {
},
'Check Response Body Containing Number Of Entities': {
'params': ['entity_type', 'number_entities', 'response'],
'position': [0, 1, 2]
},
'Check Response Body Containing Context Source Registration element': {
},
'Check Response Body Containing EntityTypeList element': {
},
'Check Response Body Containing EntityType element': {
},
'Check Response Body Type When Using Session Request': {
'params': ['type'],
'Check Created Resource Set To': {
'params': ['created_resource', 'response_body', 'ignored_keys'],
'position': []
},
'Check Created Resources Set To': {
'params': ['expected_resources', 'response_body', 'ignored_keys'],
'position': []
},
'Check RL Response Body Containing ProblemDetails Element Containing Type Element set to': {
},
'Check JSON Value In Response Body': {
},
'Check Pagination Prev And Next Headers': {
},
'Check Updated Resources Set To': {
'params': ['number_entities'],
},
'Check SUT Not Containing Resource': {
},
'Check NotificationParams': {
'params': ['filename', 'expected_additional_members', 'response_body'],
'position': [0, 1, 2]
},
'Check Response Body Containing Batch Operation Result': {
'params': ['operation'],
},
'Should be Equal': {
'params': ['expected_value', 'obtained_value'],
},
'Check Response Body Containing Subscription element': {
'params': ['filename', 'subscription_id', 'response_body'],
},
'Wait for notification': {
'params': ['timeout'],
'position': []
'Wait for notification and validate it': {
'params': ['expected_subscription_id', 'expected_context_source_registration_ids',
'expected_trigger_reason', 'expected_notification_data_entities',
'timeout'],
'position': []
},
'Wait for no notification': {
'params': ['timeout'],
'position': []
},
'Check Response Body Containing AttributeList element': {
'params': ['filename', 'response'],
},
'Check Response Body Containing Entity element': {
'params': ['filename', 'id', 'response'],
'Check Response Body Content': {
'params': ['expectation_filename', 'response_body', 'additional_ignored_path'],
'position': []
},
'Dictionary Should Contain Key': {
'params': ['dictionary', 'key'],
'position': [0, 1]
},
'Should Not Be Empty': {
'params': ['variable'],
'position': [0]
},
'Should be True': {
'params': ['expression'],
'position': [0]
},
'Check Retrieving Context Source Registration': {
'params': ['registration_id', 'context', 'accept', 'registration_payload'],
'position': [0, 1, 2, 3]
lopezaguilar
committed
},
'Check JSON Value Not In Response Body': {
'params': ['json_path_expr'],
'position': [0]
},
'Check Response Reason set to': {
'params': ['reason'],
'position': [1]
},
'Check Response Does Not Contain Body': {
'params': ['response'],
'position': [0]
,
'Check Response Headers Containing NGSILD-Results-Count Equals To': {
'params': ['expected_result_count' , 'response_headers'],
'position': [0, 1]
}
}
@staticmethod
def check_response_status_code(kwargs: list) -> str:
if "status_code" in kwargs:
status_code = kwargs['status_code']
try:
return f'Response Status Code set to {status_code} ({HTTPStatus(status_code).phrase})'
except ValueError:
return f'Response Status Code set to {status_code}'
else:
raise Exception(f'ERROR, Expected status_code parameter but received: {kwargs}')
@staticmethod
def wait_for_notification(kwargs: list) -> str:
if 'timeout' in kwargs and kwargs['timeout'] != '':
result = f"After waiting '{kwargs['timeout']}' seconds"
else:
result = f"After waiting '5' seconds"
return result
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
@staticmethod
def wait_for_notification_and_validate_it(kwargs: list) -> str:
expected_parameters = ['expected_subscription_id', 'expected_context_source_registration_ids',
'expected_trigger_reason', 'expected_notification_data_entities',
'timeout']
if 'expected_notification_data_entities' not in kwargs:
kwargs['expected_notification_data_entities'] = "${EMPTY}"
if 'timeout' not in kwargs:
kwargs['timeout'] = '5'
result = [x for x in kwargs if x not in expected_parameters]
response = "Waiting for Notification and validate it"
for key, value in kwargs.items():
match key:
case 'expected_subscription_id':
response = f"{response} and\n Query Parameter: expected_subscription_id set to '{value}'"
case 'expected_context_source_registration_ids':
response = f"{response} and\n Query Parameter: expected_context_source_registration_ids set to '{value}'"
case 'expected_trigger_reason':
response = f"{response} and\n Query Parameter: expected_trigger_reason set to '{value}'"
case 'expected_notification_data_entities':
response = f"{response} and\n Query Parameter: expected_notification_data_entities set to '{value}'"
case 'timeout':
response = f"{response} and\n Query Parameter: timeout set to '{value}'"
# If an exact match is not confirmed, this last case will be used if provided
case 'checks':
pass
case _:
raise Exception(f"ERROR, unexpected attribute '{result}', the attributes expected are "
f"'{expected_parameters}', but received: {kwargs}")
return response
@staticmethod
def wait_for_no_notification(kwargs: list) -> str:
expected_parameters = ['timeout']
if 'timeout' not in kwargs:
kwargs['timeout'] = '5'
result = [x for x in kwargs if x not in expected_parameters]
response = "Waiting for no Notification data"
for key, value in kwargs.items():
match key:
case 'timeout':
response = f"{response} and\n Query Parameter: timeout set to '{value}'"
# If an exact match is not confirmed, this last case will be used if provided
case 'checks':
pass
case _:
raise Exception(f"ERROR, unexpected attribute '{result}', the attributes expected are "
f"'{expected_parameters}', but received: {kwargs}")
return response
@staticmethod
def check_response_reason_set_to(kwargs: list) -> str:
return f"Response reason set to '{kwargs['reason']}'"
@staticmethod
def check_response_does_not_contain_body(kwargs: list) -> str:
return f"Response does not contain a body"
@staticmethod
def check_response_body_containing_array_of_uris_set_to(kwargs: list) -> str:
return 'Response Body set to an array of created entities ids'
@staticmethod
def check_created_resources_set_to(kwargs: list) -> str:
return 'Created resources set to ${entities}'
@staticmethod
def dictionary_should_contain_key(kwargs: list) -> str:
if 'dictionary' in kwargs and 'key' in kwargs:
return f"The dictionary `{kwargs['dictionary']}' should contain the key '{kwargs['key']}'"
else:
raise Exception(f'ERROR, Expected dictionary and key parameters but received: {kwargs}')
@staticmethod
def should_not_be_empty(kwargs: list) -> str:
if 'variable' in kwargs:
return f"The variable `{kwargs['variable']}' should not be '${{EMPTY}}'"
else:
raise Exception(f'ERROR, Expected dictionary and key parameters but received: {kwargs}')
@staticmethod
def should_be_true(kwargs: list) -> str:
if 'expression' in kwargs:
return f"The expression `{kwargs['expression']}' should be True"
else:
raise Exception(f'ERROR, Expected dictionary and key parameters but received: {kwargs}')
@staticmethod
def check_response_headers_containing_content_type_set_to(kwargs: list) -> str:
if "content_type" in kwargs:
content_type = kwargs['content_type']
return f'Response Header: Content-Type set to {content_type}'
else:
raise Exception(f'ERROR, Expected status_code parameter but received: {kwargs}')
@staticmethod
def check_response_headers_link_not_empty(kwargs: list) -> str:
return f'Response Header: Link is not Empty'
@staticmethod
def check_response_headers_containing_uri_set_to(kwargs: list) -> str:
if 'expected_entity_id' in kwargs and 'response_headers' in kwargs:
return f"Response Header: Location containing ${kwargs['expected_entity_id']}"
else:
raise Exception(f'ERROR, Expected expected_entity_id and response_headers parameters '
f'but received: {kwargs}')
@staticmethod
def check_response_headers_id_not_empty(kwargs: list) -> str:
return 'Response Header: Location is not Empty'
@staticmethod
def check_response_body_containing_an_attribute_set_to(kwargs: list) -> str:
expected_parameters = ['checks', 'expected_attribute_name', 'response_body', 'expected_attribute_value']
result = [x for x in kwargs if x not in expected_parameters]
response = "Check Response Body containing an Attribute set to"
for key, value in kwargs.items():
match key:
case 'expected_attribute_name':
response = f"{response} and\n Query Parameter: expected_attribute_name set to '{value}'"
case 'response_body':
response = f"{response} and\n Query Parameter: response_body set to '{value}'"
case 'expected_attribute_value':
response = f"{response} and\n Query Parameter: expected_attribute_value set to '{value}'"
# If an exact match is not confirmed, this last case will be used if provided
case 'checks':
pass
case _:
raise Exception(f"ERROR, unexpected attribute '{result}', the attributes expected are "
f"'{expected_parameters}', but received: {kwargs}")
return response
@staticmethod
def check_response_body_containing_entity_element(kwargs: list) -> str:
if 'filename' in kwargs and 'id' in kwargs and 'response' in kwargs:
return f"Response Body containing en entity element with id set to '{kwargs['id']}' and body content set to '{kwargs['filename']}'"
@staticmethod
def check_response_body_containing_list_containing_entity_elements(kwargs: list) -> str:
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
expected_parameters = ['expectation_filename', 'entities_ids', 'response_body', 'ignore_core_context_version']
if 'ignore_core_context_version' not in kwargs:
kwargs['ignore_core_context_version'] = False
result = [x for x in kwargs if x not in expected_parameters]
response = "Response Body containing a list containing Entity Elements"
for key, value in kwargs.items():
match key:
case 'expectation_filename':
response = f"{response} and\n Query Parameter: {key} set to '{value}'"
case 'entities_ids':
response = f"{response} and\n Query Parameter: {key} set to '{value}'"
case 'response_body':
response = f"{response} and\n Query Parameter: {key} set to '{value}'"
case 'ignore_core_context_version':
response = f"{response} and\n Query Parameter: {key} set to '{value}'"
case 'checks':
pass
case _:
raise Exception(f"ERROR, unexpected attribute '{result}', the attributes expected are "
f"'{expected_parameters}', but received: {kwargs}")
return response
@staticmethod
def check_response_body_containing_list_containing_entity_elements_with_different_types(kwargs: list) -> str:
expected_parameters = ['filename', 'entities_representation_ids', 'response_body', 'ignore_core_context_version']
if 'ignore_core_context_version' not in kwargs:
kwargs['ignore_core_context_version'] = False
result = [x for x in kwargs if x not in expected_parameters]
response = "Response body containing a list containing entity elements with different types"
for key, value in kwargs.items():
match key:
case 'filename':
response = f"{response} and\n Query Parameter: {key} set to '{value}'"
case 'entities_representation_ids':
response = f"{response} and\n Query Parameter: {key} set to '{value}'"
case 'response_body':
response = f"{response} and\n Query Parameter: {key} set to '{value}'"
case 'ignore_core_context_version':
response = f"{response} and\n Query Parameter: {key} set to '{value}'"
case 'checks':
pass
case _:
raise Exception(f"ERROR, unexpected attribute '{result}', the attributes expected are "
f"'{expected_parameters}', but received: {kwargs}")
return response
@staticmethod
def check_retrieving_context_source_registration(kwargs: list) -> str:
expected_parameters = ['registration_id', 'context', 'accept', 'registration_payload']
result = [x for x in kwargs if x not in expected_parameters]
response = "Check Retrieving Context Source Registration"
for key, value in kwargs.items():
match key:
case 'registration_id':
response = f"{response} and\n Query Parameter: registration_id set to '{value}'"
case 'context':
response = f"{response} and\n Query Parameter: context set to '{value}'"
case 'accept':
response = f"{response} and\n Query Parameter: accept set to '{value}'"
case 'registration_payload':
response = f"{response} and\n Query Parameter: registration_payload set to '{value}'"
case 'checks':
pass
case _:
raise Exception(f"ERROR, unexpected attribute '{result}', the attributes expected are "
f"'{expected_parameters}', but received: {kwargs}")
return response
@staticmethod
def check_response_body_containing_entitytemporal_element(kwargs: list) -> str:
if 'filename' in kwargs and 'temporal_entity_representation_id' in kwargs and 'response_body' in kwargs:
return (f"Response Body containing EntityTemporal element containing attribute instances in the time range"
f" specified by the NGSI-LD temporal query:\n"
f" * the payload is defined in the file set to '{kwargs['filename']}'\n"
f" * the id was changed to '{kwargs['temporal_entity_representation_id']}'\n"
f" * response body to be checked set to '{kwargs['response_body']}'")
else:
raise Exception(f"ERROR, expected 'filename', 'temporal_entity_representation_id', and 'response_body' "
f"attributes, received: '{kwargs}'")
@staticmethod
def check_response_body_containing_list_containing_entitytemporal_elements(kwargs: list) -> str:
if 'filename' in kwargs and 'entity_ids' in kwargs:
return (f"Request response body containing a list that contains Entity Temporal Elements\n"
f" compared with file '{kwargs['filename']}'\n"
f" and using the list of entity ids define in '{kwargs['entity_ids']}'")
raise Exception(f"ERROR, expected parameters 'filename' and 'entity_ids', but received '{kwargs}'")
@staticmethod
def check_response_body_containing_subscription_element(kwargs: list) -> str:
if 'filename' in kwargs and 'subscription_id' in kwargs and 'response_body' in kwargs:
return (f"Response Body containing the same content defined in file '{kwargs['filename']}'"
f" with subscription id '{kwargs['subscription_id']}'")
else:
raise Exception(f"ERROR, expected parameters 'filename' and 'entity_ids', but received '{kwargs}'")
def check_response_body_containing_list_containing_subscription_elements(kwargs: list) -> str:
if 'file' in kwargs and 'id' in kwargs and 'response' in kwargs:
return (f"Response containing:\n"
f" * file set to '{kwargs['file']}'\n"
f" * id set to '{kwargs['id']}'\n"
f" * response set to '{kwargs['response']}'")
raise Exception(f"ERROR, expected 'file', 'id', and 'response' attributes, received: '{kwargs}'")
@staticmethod
def check_response_body_containing_number_of_entities(kwargs: list) -> str:
if "entity_type" in kwargs and 'number_entities' in kwargs and 'response' in kwargs:
number_entities = kwargs['number_entities']
response = kwargs['response']
return (f"Response Body containing a list of entities equal to '{number_entities}' of type '{entity_type}' "
f"with response set to '{response}'")
else:
raise Exception(f'ERROR, expected entity_type and number_entities attributes, but received: {kwargs}')
@staticmethod
def check_response_body_containing_context_source_registration_element(kwargs: list) -> str:
if 'file' in kwargs and 'id' in kwargs and 'response' in kwargs:
return (f"Response containing:\n"
f" * file set to '{kwargs['file']}'\n"
f" * id set to '{kwargs['id']}'\n"
f" * response set to '{kwargs['response']}'")
raise Exception(f"ERROR, expected 'file', 'id', and 'response' attributes, received: '{kwargs}'")
# if 'csr_description' in kwargs:
# csr = kwargs['csr_description']
# return f"Response body containing a '{csr}'"
# else:
# raise Exception(f"ERROR, expected csr_description attribute, but received: {kwargs}")
@staticmethod
def check_response_body_containing_entitytypelist_element(kwargs: list) -> str:
if 'filename' in kwargs and 'response' in kwargs:
return f"Response Body containing an Entity Type List with expectation body equal to file: '{kwargs['filename']}'"
raise Exception(f"ERROR, expected filename and response attributes, but received: {kwargs}")
@staticmethod
def check_response_body_containing_entitytype_element(kwargs: list) -> str:
if 'filename' in kwargs and 'response' in kwargs:
return f"Response Body containing an Entity Type Element with expectation body equal to file: '{kwargs['filename']}'"
raise Exception(f"ERROR, expected filename and response attributes, but received: {kwargs}")
def check_response_body_containing_entitytypeinfo_element(kwargs: list) -> str:
if 'expectation_filename' in kwargs and 'response_body' in kwargs:
return (f"Response body containing an Entity Type Info "
f"with expectation body set to file '{kwargs['expectation_filename']}' "
f"and response body to be checked set to '{kwargs['response_body']}'")
else:
raise Exception(f"ERROR, expected filename and response attributes, but received: {kwargs}")
def check_response_body_containing_attributelist_element(kwargs: list) -> str:
if 'filename' in kwargs and 'response' in kwargs:
return (f"Response Body containing an Attribute List element"
f"\n * with filename set to '{kwargs['filename']}'"
f"\n * response set to '{kwargs['response']}'")
@staticmethod
def check_response_body_containing_attribute_element(kwargs: list) -> str:
if 'expectation_filename' in kwargs and 'response_body' in kwargs:
return (f"Response body containing an array of attributes"
f"\n * with the expected payload defined in the file '{kwargs['expectation_filename']}'"
f"\n * and response body set to '{kwargs['response_body']}'")
@staticmethod
def check_response_body_containing_list_containing_context_source_registrations_elements(kwargs: list) -> str:
return 'Response body set to list of all matching Context Source Registrations resolved against the default JSON-LD context'
@staticmethod
def check_response_body_type_when_using_session_request(kwargs: list) -> str:
if 'type' in kwargs:
type = kwargs['type']
return f"Response Body containing the type '{type}'"
else:
raise Exception(f"ERROR, expected type attribute, but received: {kwargs}")
@staticmethod
def check_response_body_containing_problemdetails_element_containing_type_element_set_to(kwargs: list) -> str:
if 'type' in kwargs:
type = kwargs['type']
return f"Response Body containing the type '{type}'"
else:
raise Exception(f"ERROR, expected type attribute, but received: {kwargs}")
@staticmethod
def check_response_body_containing_problemdetails_element(kwargs: list) -> str:
if 'problem_type' in kwargs:
problem_type = kwargs['problem_type']
return (f"Response Body containing the type '{problem_type}' and "
f"Response Body Title is a string and is not ${{EMPTY}} and "
f"Response Body Detail is a string and is not ${{EMPTY}} and "
f"Response Boty Title is not equal to Response Body Detail")
else:
raise Exception(f"ERROR, expected type attribute, but received: {kwargs}")
@staticmethod
def check_response_body_title_when_using_session_request(kwargs: list) -> str:
return "Response body containing 'title' element"
@staticmethod
def check_response_body_containing_problemdetails_element_containing_title_element(kwargs: list) -> str:
return "Response body containing 'title' element"
def check_rl_response_body_containing_problemdetails_element_containing_type_element_set_to(kwargs: list) -> str:
if 'type' in kwargs:
type = kwargs['type']
return f"Response Body containing the type '{type}'"
else:
raise Exception(f"ERROR, expected type attribute, but received: {kwargs}")
@staticmethod
def check_rl_response_body_containing_problemdetails_element_containing_title_element(kwargs: list) -> str:
return "Response body containing 'title' element"
@staticmethod
def check_json_value_in_response_body(kwargs: list) -> str:
if 'key' in kwargs and 'value' in kwargs:
key = kwargs['key']
value = kwargs['value']
return f"Response Body containing the key '{key}', with the value '{value}'"
else:
raise Exception(f"ERROR, expected key and value attributes, but received: {kwargs}")
lopezaguilar
committed
@staticmethod
def check_json_value_not_in_response_body(kwargs: list) -> str:
if 'json_path_expr' in kwargs:
key = kwargs['json_path_expr']
return f"Check that response body does not contain the key '{key}'"
else:
raise Exception(f"ERROR, expected key and value attributes, but received: {kwargs}")
@staticmethod
def check_pagination_prev_and_next_headers(kwargs: list) -> str:
previous = None
next = None
if 'previous' in kwargs:
previous = kwargs['previous']
previous_text = f"with 'Prev' header equal to '{previous}'"
if 'next' in kwargs:
next = kwargs['next']
next_text = f"with 'Next' header equal to '{next}'"
if previous is None and next is None:
raise Exception(f"ERROR, expected previous or next attributes, but received: {kwargs}")
elif previous is not None and next is None:
result = f"Response header {previous_text}"
elif previous is None and next is not None:
result = f"Response header {next_text}"
else:
return result
@staticmethod
def check_created_resource_set_to(kwargs: list) -> str:
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
expected_parameters = ['created_resource', 'response_body', 'ignored_keys']
if 'ignored_keys' not in kwargs:
kwargs['ignored_keys'] = '${None}'
result = [x for x in kwargs if x not in expected_parameters]
response = "Check Created Resource Set To"
for key, value in kwargs.items():
match key:
case 'created_resource':
response = f"{response} and\n Query Parameter: '{key}' set to '{value}'"
case 'response_body':
response = f"{response} and\n Query Parameter: '{key}' set to '{value}'"
case 'ignored_keys':
response = f"{response} and\n Query Parameter: '{key}' set to '{value}'"
# If an exact match is not confirmed, this last case will be used if provided
case 'checks':
pass
case _:
raise Exception(f"ERROR, unexpected attribute '{result}', the attributes expected are "
f"'{expected_parameters}', but received: {kwargs}")
return response
@staticmethod
def check_created_resources_set_to(kwargs: list) -> str:
expected_parameters = ['expected_resources', 'response_body', 'ignored_keys']
if 'ignored_keys' not in kwargs:
kwargs['ignored_keys'] = '${None}'
result = [x for x in kwargs if x not in expected_parameters]
response = "Check Created Resource Set To"
for key, value in kwargs.items():
match key:
case 'expected_resources':
response = f"{response} and\n Query Parameter: '{key}' set to '{value}'"
case 'response_body':
response = f"{response} and\n Query Parameter: '{key}' set to '{value}'"
case 'ignored_keys':
response = f"{response} and\n Query Parameter: '{key}' set to '{value}'"
# If an exact match is not confirmed, this last case will be used if provided
case 'checks':
pass
case _:
raise Exception(f"ERROR, unexpected attribute '{result}', the attributes expected are "
f"'{expected_parameters}', but received: {kwargs}")
return response
def check_updated_resource_set_to(kwargs: list) -> str:
return "Updated Entity set to ${entity}"
def check_updated_resources_set_to(kwargs: list) -> str:
if 'number_entities' in kwargs:
number_entities = kwargs['number_entities']
return f"Updated Entities set to '{number_entities}' valid entities"
def check_sut_not_containing_resource(kwargs: list) -> str:
if "status_code" in kwargs:
status_code = kwargs['status_code']
try:
return f'Response Status Code set to {status_code} ({HTTPStatus(status_code).phrase})'
except ValueError:
return f'Response Status Code set to {status_code}'
else:
raise Exception(f'ERROR, Expected status_code parameter but received: {kwargs}')
def check_sut_not_containing_resources(kwargs: list) -> str:
return f'Response body is empty'
def check_notificationparams(kwargs: list) -> str:
if 'filename' in kwargs and 'expected_additional_members' in kwargs and 'response_body' in kwargs:
return (f"Response containing:\n"
f" * Notification expectation file path set to '{kwargs['filename']}'\n"
f" * Expected Additional Members set to '{kwargs['expected_additional_members']}'\n"
f" * Response Body set to '{kwargs['response_body']}'\n")
raise Exception(f"ERROR, expected 'filename', 'expected_additional_members', and 'response_body', "
f"but received '{kwargs}'")
@staticmethod
def check_response_body_containing_batch_operation_result(kwargs: list) -> str:
if "operation" in kwargs:
return f"Response body containing batch operation result set to '{kwargs['operation']}'"
else:
raise Exception(f'ERROR, Expected operation parameter but received: {kwargs}')
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
@staticmethod
def check_response_body_content(kwargs: list) -> str:
expected_parameters = ['expectation_filename', 'response_body', 'additional_ignored_path']
if 'additional_ignored_path' not in kwargs:
kwargs['additional_ignored_path'] = '${EMPTY}'
result = [x for x in kwargs if x not in expected_parameters]
response = "Check Response Body Content"
for key, value in kwargs.items():
match key:
case 'expectation_filename':
response = f"{response} and\n Query Parameter: expectation_filename set to '{value}'"
case 'response_body':
response = f"{response} and\n Query Parameter: response_body set to '{value}'"
case 'additional_ignored_path':
response = f"{response} and\n Query Parameter: additional_ignored_path set to '{value}'"
# If an exact match is not confirmed, this last case will be used if provided
case 'checks':
pass
case _:
raise Exception(f"ERROR, unexpected attribute '{result}', the attributes expected are "
f"'{expected_parameters}', but received: {kwargs}")
return response
@staticmethod
def should_be_equal(kwargs: list) -> str:
if 'expected_value' in kwargs and 'obtained_value' in kwargs:
return f"Notification data: '{kwargs['obtained_value']}' equal to '{kwargs['expected_value']}'"
else:
raise Exception(f"ERROR, Expected 'expected_value' and 'obtained_value' parameters but received: '{kwargs}'")
@staticmethod
def check_response_header_contains_ngsild_results_count_equals_to(kwargs: list) -> str:
if 'expected_result_count' in kwargs:
expected_result_count = kwargs['expected_result_count']
return f'Response Header: NGSILD-Results-Count equals to {expected_result_count}'
else:
raise Exception(f"ERROR, Expected 'expected_result_count' but received: '{kwargs}'")
def get_checks(self, **kwargs) -> str:
checking = None
if "checks" in kwargs:
checking = kwargs["checks"]
else:
raise Exception(f'ERROR, the attribute checks is mandatory, but received: {kwargs}')
if isinstance(checking, str):
result = self.checks[checking](kwargs)
elif isinstance(checking, list):
result = [self.checks[x](kwargs) for x in checking]
result = " and\n".join(result)
else:
raise Exception(f"ERROR, checks type not supported: {checking}")
return result
if __name__ == "__main__":
data = Checks()
print(data.get_checks(checks='Check Response Status Code',
status_code=201))
print(data.get_checks(checks='Check Response Body Containing Array Of URIs set to'))
print(data.get_checks(checks='Check Created Resources Set To'))
print(data.get_checks(checks='Check Response Headers Containing Content-Type set to',
content_type='application/json'))
print(data.get_checks(checks='Check Response Headers Link Not Empty'))
print(data.get_checks(checks='Check Response Headers Containing URI set to'))
print(data.get_checks(checks='Check Response Headers ID Not Empty'))
print(data.get_checks(checks='Check Response Body Containing an Attribute set to',
attribute_name='status'))
print(data.get_checks(checks='Check Response Body Containing an Attribute set to',
attribute_name='status',
attribute_value='active'))
print(data.get_checks(checks='Check Response Body Containing Entity element'))
print(data.get_checks(checks='Check Response Body Containing List Containing Entity Elements'))
print(data.get_checks(checks='Check Response Body Containing List Containing Entity Elements With Different Types'))
print(data.get_checks(checks='Check Response Body Containing EntityTemporal element'))
print(data.get_checks(checks='Check Response Body Containing List Containing EntityTemporal elements',
timeRel='after',
timeAt='2020-07-01T12:05:00Z'))
print(data.get_checks(checks='Check Response Body Containing Subscription element'))
print(data.get_checks(checks='Check Response Body Containing List Containing Subscription elements',
number=2))
print(data.get_checks(checks='Check Response Body Containing List Containing Subscription elements',
number=1))
print(data.get_checks(checks='Check Response Body Containing Number Of Entities',
entity_type="Vehicle",
number_entities=3))
print(data.get_checks(checks='Check Response Body Containing Context Source Registration element',
csr_description='Context Source Registration'))
print(data.get_checks(checks='Check Response Body Containing EntityTypeList element',
description='Json object with list of entity types with context'))
print(data.get_checks(checks='Check Response Body Containing EntityType element',
description='Json object with an entity type with context'))
print(data.get_checks(checks='Check Response Body Containing EntityTypeInfo element'))
print(data.get_checks(checks='Check Response Body Containing AttributeList element'))
print(data.get_checks(checks='Check Response Body Containing Attribute element'))
print(data.get_checks(checks='Check Response Body Containing List Containing Context Source Registrations elements'))
print(data.get_checks(checks='Check Response Body Type When Using Session Request',
type='https://uri.etsi.org/ngsi-ld/errors/BadRequestData'))
print(data.get_checks(checks='Check Response Body Containing ProblemDetails Element Containing Type Element set to',
type='https://uri.etsi.org/ngsi-ld/errors/BadRequestData'))
print(data.get_checks(checks='Check Response Body Title When Using Session Request'))
print(data.get_checks(checks='Check Response Body Containing ProblemDetails Element Containing Title Element'))
print(data.get_checks(checks='Check RL Response Body Containing ProblemDetails Element Containing Type Element set to',
type='https://uri.etsi.org/ngsi-ld/errors/BadRequestData'))
print(data.get_checks(checks='Check RL Response Body Containing ProblemDetails Element Containing Title Element'))
print(data.get_checks(checks='Check JSON Value In Response Body',
key="['information']['entities'][0]['type']",
value="Building"))
print(data.get_checks(checks='Check Pagination Prev And Next Headers',
previous='</ngsi-ld/v1/csourceSubscriptions?limit=1&page=1>;rel="prev";type="application/ld+json"',
next='</ngsi-ld/v1/csourceSubscriptions?limit=1&page=3>;rel="next";type="application/ld+json"'))
print(data.get_checks(checks='Check Pagination Prev And Next Headers',
previous='',
next='</ngsi-ld/v1/csourceSubscriptions?limit=1&page=3>;rel="next";type="application/ld+json"'))
print(data.get_checks(checks='Check Pagination Prev And Next Headers',
previous='</ngsi-ld/v1/csourceSubscriptions?limit=1&page=1>;rel="prev";type="application/ld+json"',
next=''))
print(data.get_checks(checks='Check Pagination Prev And Next Headers',
previous='',
next=''))
print(data.get_checks(checks='Check Created Resource Set To'))
print(data.get_checks(checks='Check Updated Resource Set To'))
print(data.get_checks(checks='Check Updated Resources Set To',
number_entities=2))
print(data.get_checks(checks='Check SUT Not Containing Resource',
status_code=404))
print(data.get_checks(checks='Check SUT Not Containing Resources'))
print(data.get_checks(checks='Check NotificationParams',
format="keyValues",
uri="http://my.endpoint.org/notify",
accept="application/json",
status="ok",
timesSent="1"))
print()
print(data.get_checks(checks=
['Check Response Status Code',
'Check Response Body Containing Array Of URIs set to',
'Check Created Resources Set To']
, status_code=201))
print(data.get_checks(checks='Check Response Headers Containing NGSILD-Results-Count Equals To'))
# Check exceptions
try:
print(data.get_checks(checks='Check Response Body Containing an Attribute set to'))
except Exception as e: