Skip to content
checks.py 53.2 KiB
Newer Older
from http import HTTPStatus


class Checks:
    def __init__(self):
        self.checks = {
lopezaguilar's avatar
lopezaguilar committed
            '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':
lopezaguilar's avatar
lopezaguilar committed
                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':
lopezaguilar's avatar
lopezaguilar committed
                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,
lopezaguilar's avatar
lopezaguilar committed
            'Check Response Body Containing a Boolean Attribute set to':
                Checks.check_response_body_containing_a_boolean_attribute_set_to,
            'Check Response Body Contains DateTime Value':
                Checks.check_response_body_contains_datetime_value,
lopezaguilar's avatar
lopezaguilar committed
            'Check Response Body Containing ProblemDetails Element Containing Title Element':
                Checks.check_response_body_containing_problemdetails_element_containing_title_element,
            'Check Dictionary Might Contain Additional Members of the NotificationParams':
lopezaguilar's avatar
lopezaguilar committed
                Checks.check_dictionary_might_contain_additional_members,
            'Check Response Body Containing ProblemDetails Element':
                Checks.check_response_body_containing_problemdetails_element,
lopezaguilar's avatar
lopezaguilar committed
            'Check RL Response Body Containing ProblemDetails Element Containing Type Element set to':
lopezaguilar's avatar
lopezaguilar committed
                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,
lopezaguilar's avatar
lopezaguilar committed
            'Check SUT Not Containing Resource':
                Checks.check_sut_not_containing_resource,
            'Check SUT Not Containing Resources':
                Checks.check_sut_not_containing_resources,
            'Check NotificationParams':
lopezaguilar's avatar
lopezaguilar committed
                Checks.check_notificationparams,
            'Check Response Body Containing Batch Operation Result':
lopezaguilar's avatar
lopezaguilar committed
                Checks.check_response_body_containing_batch_operation_result,
lopezaguilar's avatar
lopezaguilar committed
            'Wait for notification':
                Checks.wait_for_notification,
            'Wait for no notification':
                Checks.wait_for_no_notification,
            'Wait for notification and validate it':
                Checks.wait_for_notification_and_validate_it,
lopezaguilar's avatar
lopezaguilar committed
            'Should be Equal':
lopezaguilar's avatar
lopezaguilar committed
                Checks.should_be_equal,
            '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,
lopezaguilar's avatar
lopezaguilar committed
            'Check Response Body Content':
lopezaguilar's avatar
lopezaguilar committed
                Checks.check_response_body_content,
            'Check Retrieving Context Source Registration':
                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,
lopezaguilar's avatar
lopezaguilar committed
            'Check Response Header is Empty':
                Checks.check_response_header_is_empty,
            'Check Response Headers Link set to':
                Checks.check_response_headers_link_set_to,
            'Check Response Headers Containing NGSILD-Results-Count Equals To' :
                Checks.check_response_header_contains_ngsild_results_count_equals_to,
lopezaguilar's avatar
lopezaguilar committed
        }

        self.args = {
            'Check Response Status Code': {
                'params': ['status_code'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [0]
            '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'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [1]
            'Check Response Body Containing ProblemDetails Element': {
                'params': ['problem_type'],
                'position': [1]
            },
            'Check Response Headers Containing Content-Type set to': {
                'params': ['content_type'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [0]
            '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 Dictionary Might Contain Additional Members of the NotificationParams': {
lopezaguilar's avatar
lopezaguilar committed
                'params': ['dictionary', 'key'],
                'position': [0, 1]
            },
            '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': {
lopezaguilar's avatar
lopezaguilar committed
                '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'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [0, 1]
            'Check RL Response Body Containing ProblemDetails Element Containing Title Element': {
                'params': ['response_body'],
                'position': [0]
            },
lopezaguilar's avatar
lopezaguilar committed
            'Check Response Body Containing a Boolean Attribute set to': {
                'params': ['expected_attribute_name', 'response_body', 'expected_attribute_value'],
                'position': [0, 1, 2]
            },
            'Check Response Body Contains DateTime Value': {
                'params': ['dictionary', 'key', 'expected value'],
                'position': [0, 1, 2]
            },
            'Check Response Body Containing List Containing Subscription elements': {
lopezaguilar's avatar
lopezaguilar committed
                'params': ['file', 'id', 'response'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [0, 1, 2]
            },
            'Check Response Body Containing Number Of Entities': {
lopezaguilar's avatar
lopezaguilar committed
                'params': ['entity_type', 'number_entities', 'response'],
                'position': [0, 1, 2]
            },
            'Check Response Body Containing Context Source Registration element': {
lopezaguilar's avatar
lopezaguilar committed
                'params': ['file', 'id', 'response'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [0, 1, 2]
            },
            'Check Response Body Containing EntityTypeList element': {
                'params': ['filename', 'response'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [0, 1]
            },
            'Check Response Body Containing EntityType element': {
                'params': ['filename', 'response'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [0, 1]
            },
            'Check Response Body Type When Using Session Request': {
                'params': ['type'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [1]
            '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': {
                'params': ['type'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [1]
            },
            'Check JSON Value In Response Body': {
                'params': ['key', 'value'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [0, 1]
            },
            'Check Pagination Prev And Next Headers': {
                'params': ['previous', 'next'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [1, 2]
            },
            'Check Updated Resources Set To': {
                'params': ['number_entities'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [0]
            },
            'Check SUT Not Containing Resource': {
                'params': ['status_code'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [0]
            },
            'Check NotificationParams': {
lopezaguilar's avatar
lopezaguilar committed
                'params': ['filename', 'expected_additional_members', 'response_body'],
lopezaguilar's avatar
lopezaguilar committed
            },
            'Check Response Body Containing Batch Operation Result': {
                'params': ['operation'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [0]
lopezaguilar's avatar
lopezaguilar committed
            },
            'Should be Equal': {
                'params': ['expected_value', 'obtained_value'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [0, 1]
lopezaguilar's avatar
lopezaguilar committed
            },
            'Check Response Body Containing Subscription element': {
                'params': ['filename', 'subscription_id', 'response_body'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [0, 1, 2]
lopezaguilar's avatar
lopezaguilar committed
            },
            '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'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [0, 1]
            },
            'Check Response Body Containing Entity element': {
                'params': ['filename', 'id', 'response'],
lopezaguilar's avatar
lopezaguilar committed
                'position': [0, 1, 2]
lopezaguilar's avatar
lopezaguilar committed
            '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]
lopezaguilar's avatar
lopezaguilar committed
            },
            'Check Retrieving Context Source Registration': {
                'params': ['registration_id', 'context', 'accept', 'registration_payload'],
                'position': [0, 1, 2, 3]
            },
            '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]
lopezaguilar's avatar
lopezaguilar committed
            },
            'Check Response Header is Empty': {
                'params': ['response_headers'],
                'position': [0]
            },
            'Check Response Headers Link set to': {
                'params': ['response_headers', 'expected_link_header'],
                'position': [0, 1]
lopezaguilar's avatar
lopezaguilar committed
            }
            ,
            '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}')

lopezaguilar's avatar
lopezaguilar committed
    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

    @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

lopezaguilar's avatar
lopezaguilar committed
    @staticmethod
    def check_response_header_is_empty(kwargs: list) -> str:
        return "The response header contains an empty dictionary"

    @staticmethod
    def check_response_headers_link_set_to(kwargs: list) -> str:
        return f"The response headers Link is set to '{kwargs['expected_link_header']}'"

    @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}')

lopezaguilar's avatar
lopezaguilar committed
    @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}')

lopezaguilar's avatar
lopezaguilar committed
    @staticmethod
    def check_response_body_containing_a_boolean_attribute_set_to(kwargs: list) -> str:
        return (f"Check that the payload body contains a boolean attribute with name "
                f"'{kwargs['expected_attribute_name']}' and value '{kwargs['expected_attribute_value']}")

lopezaguilar's avatar
lopezaguilar committed
    @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}')
lopezaguilar's avatar
lopezaguilar committed

    @staticmethod
    def check_response_headers_id_not_empty(kwargs: list) -> str:
        return 'Response Header: Location is not Empty'

lopezaguilar's avatar
lopezaguilar committed
    @staticmethod
    def check_dictionary_might_contain_additional_members(kwargs: list) -> str:
        if 'dictionary' in kwargs and 'key' in kwargs:
            return f"The dictionary `{kwargs['dictionary']}' might contain the key '{kwargs['key']}'"
        else:
            raise Exception(f"ERROR, Expected 'dictionary' and 'key' parameters but received: {kwargs}")

lopezaguilar's avatar
lopezaguilar committed
    @staticmethod
    def check_response_body_containing_an_attribute_set_to(kwargs: list) -> str:
lopezaguilar's avatar
lopezaguilar committed
        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
lopezaguilar's avatar
lopezaguilar committed

lopezaguilar's avatar
lopezaguilar committed
    @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']}'"
lopezaguilar's avatar
lopezaguilar committed

    @staticmethod
    def check_response_body_containing_list_containing_entity_elements(kwargs: list) -> str:
        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
lopezaguilar's avatar
lopezaguilar committed
    @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

lopezaguilar's avatar
lopezaguilar committed
    @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}'")
lopezaguilar's avatar
lopezaguilar committed

    @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']}'")
lopezaguilar's avatar
lopezaguilar committed
        else:
            raise Exception(f"ERROR, expected parameters 'filename' and 'entity_ids', but received '{kwargs}'")
lopezaguilar's avatar
lopezaguilar committed
    @staticmethod
    def check_response_body_contains_datetime_value(kwargs: list) -> str:
        # ${dictionary}    ${key}    ${expected value}
        return (f"Check that the dictionary contains the key '{kwargs['key']}' with type DateTime and value set "
                f"to '{kwargs['expected value']}'")

lopezaguilar's avatar
lopezaguilar committed
    @staticmethod
    def check_response_body_containing_subscription_element(kwargs: list) -> str:
lopezaguilar's avatar
lopezaguilar committed
        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}'")
lopezaguilar's avatar
lopezaguilar committed
    @staticmethod
lopezaguilar's avatar
lopezaguilar committed
    def check_response_body_containing_list_containing_subscription_elements(kwargs: list) -> str:
lopezaguilar's avatar
lopezaguilar committed
        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']}'")
lopezaguilar's avatar
lopezaguilar committed
        else:
lopezaguilar's avatar
lopezaguilar committed
            raise Exception(f"ERROR, expected 'file', 'id', and 'response' attributes, received: '{kwargs}'")
lopezaguilar's avatar
lopezaguilar committed
    @staticmethod
    def check_response_body_containing_number_of_entities(kwargs: list) -> str:
lopezaguilar's avatar
lopezaguilar committed
        if "entity_type" in kwargs and 'number_entities' in kwargs and 'response' in kwargs:
            number_entities = kwargs['number_entities']
lopezaguilar's avatar
lopezaguilar committed
            entity_type = kwargs['entity_type']
lopezaguilar's avatar
lopezaguilar committed
            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}'")
lopezaguilar's avatar
lopezaguilar committed
        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:
lopezaguilar's avatar
lopezaguilar committed
        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']}'")
lopezaguilar's avatar
lopezaguilar committed
        else:
lopezaguilar's avatar
lopezaguilar committed
            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}")
lopezaguilar's avatar
lopezaguilar committed

    @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']}'"
lopezaguilar's avatar
lopezaguilar committed
        else:
            raise Exception(f"ERROR, expected filename and response attributes, but received: {kwargs}")
lopezaguilar's avatar
lopezaguilar committed

    @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']}'"
lopezaguilar's avatar
lopezaguilar committed
        else:
            raise Exception(f"ERROR, expected filename and response attributes, but received: {kwargs}")
lopezaguilar's avatar
lopezaguilar committed
    @staticmethod
lopezaguilar's avatar
lopezaguilar committed
    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}")
lopezaguilar's avatar
lopezaguilar committed
    @staticmethod
lopezaguilar's avatar
lopezaguilar committed
    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']}'")
lopezaguilar's avatar
lopezaguilar committed
    @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']}'")
lopezaguilar's avatar
lopezaguilar committed

    @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}")

lopezaguilar's avatar
lopezaguilar committed
    @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"

lopezaguilar's avatar
lopezaguilar committed
    @staticmethod
lopezaguilar's avatar
lopezaguilar committed
    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}")

lopezaguilar's avatar
lopezaguilar committed
    @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}")

    @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}")

lopezaguilar's avatar
lopezaguilar committed
    @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:
lopezaguilar's avatar
lopezaguilar committed
            result = f"Response header {previous_text} and {next_text}"
lopezaguilar's avatar
lopezaguilar committed

        return result

    @staticmethod
    def check_created_resource_set_to(kwargs: list) -> str:
        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':
lopezaguilar's avatar
lopezaguilar committed
                    response = f"{response} and\n    Query Parameter: '{key}' set to '{value}' list of keys"
                # 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
    @staticmethod
lopezaguilar's avatar
lopezaguilar committed
    def check_updated_resource_set_to(kwargs: list) -> str:
            return "Updated Entity set to ${entity}"

    @staticmethod
    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"

    @staticmethod
lopezaguilar's avatar
lopezaguilar committed
    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}'
lopezaguilar's avatar
lopezaguilar committed
        else:
            raise Exception(f'ERROR, Expected status_code parameter but received: {kwargs}')

    @staticmethod
    def check_sut_not_containing_resources(kwargs: list) -> str:
        return f'Response body is empty'

    @staticmethod
    def check_notificationparams(kwargs: list) -> str:
lopezaguilar's avatar
lopezaguilar committed
        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")
lopezaguilar's avatar
lopezaguilar committed
            raise Exception(f"ERROR, expected 'filename', 'expected_additional_members', and 'response_body', "
                            f"but received '{kwargs}'")

lopezaguilar's avatar
lopezaguilar committed
    @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']}'"
lopezaguilar's avatar
lopezaguilar committed
        else:
            raise Exception(f'ERROR, Expected operation parameter but received: {kwargs}')

lopezaguilar's avatar
lopezaguilar committed
    @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


lopezaguilar's avatar
lopezaguilar committed
    @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()

lopezaguilar's avatar
lopezaguilar committed
    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'))
lopezaguilar's avatar
lopezaguilar committed
    print(data.get_checks(checks='Check Response Headers Containing Content-Type set to',
                          content_type='application/json'))
lopezaguilar's avatar
lopezaguilar committed
    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'))
lopezaguilar's avatar
lopezaguilar committed
    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))
lopezaguilar's avatar
lopezaguilar committed
    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',