# Copyright (c) 2018 Cisco and/or its affiliates. # # This file is part of Ansible # # Ansible is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # Ansible is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with Ansible. If not, see . # import json import unittest import pytest from ansible_collections.community.general.tests.unit.compat import mock from ansible_collections.community.general.tests.unit.compat.mock import call, patch from ansible_collections.community.general.plugins.module_utils.network.ftd.common import HTTPMethod, FtdUnexpectedResponse from ansible_collections.community.general.plugins.module_utils.network.ftd.configuration import iterate_over_pageable_resource, BaseConfigurationResource, \ OperationChecker, OperationNamePrefix, ParamName, QueryParams from ansible_collections.community.general.plugins.module_utils.network.ftd.fdm_swagger_client import ValidationError, OperationField class TestBaseConfigurationResource(object): @pytest.fixture def connection_mock(self, mocker): connection_class_mock = mocker.patch('ansible_collections.community.general.plugins.modules.network.ftd.ftd_configuration.Connection') connection_instance = connection_class_mock.return_value connection_instance.validate_data.return_value = True, None connection_instance.validate_query_params.return_value = True, None connection_instance.validate_path_params.return_value = True, None return connection_instance @patch.object(BaseConfigurationResource, '_fetch_system_info') @patch.object(BaseConfigurationResource, '_send_request') def test_get_objects_by_filter_with_multiple_filters(self, send_request_mock, fetch_system_info_mock, connection_mock): objects = [ {'name': 'obj1', 'type': 1, 'foo': {'bar': 'buzz'}}, {'name': 'obj2', 'type': 1, 'foo': {'bar': 'buz'}}, {'name': 'obj3', 'type': 2, 'foo': {'bar': 'buzz'}} ] fetch_system_info_mock.return_value = { 'databaseInfo': { 'buildVersion': '6.3.0' } } connection_mock.get_operation_spec.return_value = { 'method': HTTPMethod.GET, 'url': '/object/' } resource = BaseConfigurationResource(connection_mock, False) send_request_mock.side_effect = [{'items': objects}, {'items': []}] # resource.get_objects_by_filter returns generator so to be able compare generated list with expected list # we need evaluate it. assert objects == list(resource.get_objects_by_filter('test', {})) send_request_mock.assert_has_calls( [ mock.call('/object/', 'get', {}, {}, {'limit': 10, 'offset': 0}) ] ) send_request_mock.reset_mock() send_request_mock.side_effect = [{'items': objects}, {'items': []}] # resource.get_objects_by_filter returns generator so to be able compare generated list with expected list # we need evaluate it. assert [objects[0]] == list(resource.get_objects_by_filter('test', {ParamName.FILTERS: {'name': 'obj1'}})) send_request_mock.assert_has_calls( [ mock.call('/object/', 'get', {}, {}, {QueryParams.FILTER: 'name:obj1', 'limit': 10, 'offset': 0}) ] ) send_request_mock.reset_mock() send_request_mock.side_effect = [{'items': objects}, {'items': []}] # resource.get_objects_by_filter returns generator so to be able compare generated list with expected list # we need evaluate it. assert [objects[1]] == list(resource.get_objects_by_filter( 'test', {ParamName.FILTERS: {'name': 'obj2', 'type': 1, 'foo': {'bar': 'buz'}}})) send_request_mock.assert_has_calls( [ mock.call('/object/', 'get', {}, {}, {QueryParams.FILTER: 'name:obj2', 'limit': 10, 'offset': 0}) ] ) @patch.object(BaseConfigurationResource, '_fetch_system_info') @patch.object(BaseConfigurationResource, '_send_request') def test_get_objects_by_filter_with_multiple_responses(self, send_request_mock, fetch_system_info_mock, connection_mock): send_request_mock.side_effect = [ {'items': [ {'name': 'obj1', 'type': 'foo'}, {'name': 'obj2', 'type': 'bar'} ]}, {'items': [ {'name': 'obj3', 'type': 'foo'} ]}, {'items': []} ] fetch_system_info_mock.return_value = { 'databaseInfo': { 'buildVersion': '6.3.0' } } connection_mock.get_operation_spec.return_value = { 'method': HTTPMethod.GET, 'url': '/object/' } resource = BaseConfigurationResource(connection_mock, False) assert [{'name': 'obj1', 'type': 'foo'}] == list(resource.get_objects_by_filter( 'test', {ParamName.FILTERS: {'type': 'foo'}})) send_request_mock.assert_has_calls( [ mock.call('/object/', 'get', {}, {}, {'limit': 10, 'offset': 0}) ] ) send_request_mock.reset_mock() send_request_mock.side_effect = [ {'items': [ {'name': 'obj1', 'type': 'foo'}, {'name': 'obj2', 'type': 'bar'} ]}, {'items': [ {'name': 'obj3', 'type': 'foo'} ]}, {'items': []} ] resp = list(resource.get_objects_by_filter( 'test', { ParamName.FILTERS: {'type': 'foo'}, ParamName.QUERY_PARAMS: {'limit': 2} })) assert [{'name': 'obj1', 'type': 'foo'}, {'name': 'obj3', 'type': 'foo'}] == resp send_request_mock.assert_has_calls( [ mock.call('/object/', 'get', {}, {}, {'limit': 2, 'offset': 0}), mock.call('/object/', 'get', {}, {}, {'limit': 2, 'offset': 2}) ] ) def test_module_should_fail_if_validation_error_in_data(self, connection_mock): connection_mock.get_operation_spec.return_value = {'method': HTTPMethod.POST, 'url': '/test'} report = { 'required': ['objects[0].type'], 'invalid_type': [ { 'path': 'objects[3].id', 'expected_type': 'string', 'actually_value': 1 } ] } connection_mock.validate_data.return_value = (False, json.dumps(report, sort_keys=True, indent=4)) with pytest.raises(ValidationError) as e_info: resource = BaseConfigurationResource(connection_mock, False) resource.crud_operation('addTest', {'data': {}}) result = e_info.value.args[0] key = 'Invalid data provided' assert result[key] result[key] = json.loads(result[key]) assert result == {key: { 'invalid_type': [{'actually_value': 1, 'expected_type': 'string', 'path': 'objects[3].id'}], 'required': ['objects[0].type'] }} def test_module_should_fail_if_validation_error_in_query_params(self, connection_mock): connection_mock.get_operation_spec.return_value = {'method': HTTPMethod.GET, 'url': '/test', 'returnMultipleItems': False} report = { 'required': ['objects[0].type'], 'invalid_type': [ { 'path': 'objects[3].id', 'expected_type': 'string', 'actually_value': 1 } ] } connection_mock.validate_query_params.return_value = (False, json.dumps(report, sort_keys=True, indent=4)) with pytest.raises(ValidationError) as e_info: resource = BaseConfigurationResource(connection_mock, False) resource.crud_operation('getTestList', {'data': {}}) result = e_info.value.args[0] key = 'Invalid query_params provided' assert result[key] result[key] = json.loads(result[key]) assert result == {key: { 'invalid_type': [{'actually_value': 1, 'expected_type': 'string', 'path': 'objects[3].id'}], 'required': ['objects[0].type']}} def test_module_should_fail_if_validation_error_in_path_params(self, connection_mock): connection_mock.get_operation_spec.return_value = {'method': HTTPMethod.GET, 'url': '/test', 'returnMultipleItems': False} report = { 'path_params': { 'required': ['objects[0].type'], 'invalid_type': [ { 'path': 'objects[3].id', 'expected_type': 'string', 'actually_value': 1 } ] } } connection_mock.validate_path_params.return_value = (False, json.dumps(report, sort_keys=True, indent=4)) with pytest.raises(ValidationError) as e_info: resource = BaseConfigurationResource(connection_mock, False) resource.crud_operation('putTest', {'data': {}}) result = e_info.value.args[0] key = 'Invalid path_params provided' assert result[key] result[key] = json.loads(result[key]) assert result == {key: { 'path_params': { 'invalid_type': [{'actually_value': 1, 'expected_type': 'string', 'path': 'objects[3].id'}], 'required': ['objects[0].type']}}} def test_module_should_fail_if_validation_error_in_all_params(self, connection_mock): connection_mock.get_operation_spec.return_value = {'method': HTTPMethod.POST, 'url': '/test'} report = { 'data': { 'required': ['objects[0].type'], 'invalid_type': [ { 'path': 'objects[3].id', 'expected_type': 'string', 'actually_value': 1 } ] }, 'path_params': { 'required': ['some_param'], 'invalid_type': [ { 'path': 'name', 'expected_type': 'string', 'actually_value': True } ] }, 'query_params': { 'required': ['other_param'], 'invalid_type': [ { 'path': 'f_integer', 'expected_type': 'integer', 'actually_value': "test" } ] } } connection_mock.validate_data.return_value = (False, json.dumps(report['data'], sort_keys=True, indent=4)) connection_mock.validate_query_params.return_value = (False, json.dumps(report['query_params'], sort_keys=True, indent=4)) connection_mock.validate_path_params.return_value = (False, json.dumps(report['path_params'], sort_keys=True, indent=4)) with pytest.raises(ValidationError) as e_info: resource = BaseConfigurationResource(connection_mock, False) resource.crud_operation('putTest', {'data': {}}) result = e_info.value.args[0] key_data = 'Invalid data provided' assert result[key_data] result[key_data] = json.loads(result[key_data]) key_path_params = 'Invalid path_params provided' assert result[key_path_params] result[key_path_params] = json.loads(result[key_path_params]) key_query_params = 'Invalid query_params provided' assert result[key_query_params] result[key_query_params] = json.loads(result[key_query_params]) assert result == { key_data: {'invalid_type': [{'actually_value': 1, 'expected_type': 'string', 'path': 'objects[3].id'}], 'required': ['objects[0].type']}, key_path_params: {'invalid_type': [{'actually_value': True, 'expected_type': 'string', 'path': 'name'}], 'required': ['some_param']}, key_query_params: { 'invalid_type': [{'actually_value': 'test', 'expected_type': 'integer', 'path': 'f_integer'}], 'required': ['other_param']}} @pytest.mark.parametrize("test_api_version, expected_result", [ ("6.2.3", "name:object_name"), ("6.3.0", "name:object_name"), ("6.4.0", "fts~object_name") ] ) def test_stringify_name_filter(self, test_api_version, expected_result, connection_mock): filters = {"name": "object_name"} with patch.object(BaseConfigurationResource, '_fetch_system_info') as fetch_system_info_mock: fetch_system_info_mock.return_value = { 'databaseInfo': { 'buildVersion': test_api_version } } resource = BaseConfigurationResource(connection_mock, False) assert resource._stringify_name_filter(filters) == expected_result, "Unexpected result for version %s" % ( test_api_version) class TestIterateOverPageableResource(object): def test_iterate_over_pageable_resource_with_no_items(self): resource_func = mock.Mock(return_value={'items': []}) items = iterate_over_pageable_resource(resource_func, {'query_params': {}}) assert [] == list(items) def test_iterate_over_pageable_resource_with_one_page(self): resource_func = mock.Mock(side_effect=[ {'items': ['foo', 'bar']}, {'items': []}, ]) items = iterate_over_pageable_resource(resource_func, {'query_params': {}}) assert ['foo', 'bar'] == list(items) resource_func.assert_has_calls([ call(params={'query_params': {'offset': 0, 'limit': 10}}) ]) def test_iterate_over_pageable_resource_with_multiple_pages(self): objects = [ {'items': ['foo']}, {'items': ['bar']}, {'items': ['buzz']}, {'items': []}, ] resource_func = mock.Mock(side_effect=objects) items = iterate_over_pageable_resource(resource_func, {'query_params': {}}) assert ['foo'] == list(items) resource_func.reset_mock() resource_func = mock.Mock(side_effect=objects) items = iterate_over_pageable_resource(resource_func, {'query_params': {'limit': 1}}) assert ['foo', 'bar', 'buzz'] == list(items) def test_iterate_over_pageable_resource_should_preserve_query_params(self): resource_func = mock.Mock(return_value={'items': []}) items = iterate_over_pageable_resource(resource_func, {'query_params': {'filter': 'name:123'}}) assert [] == list(items) resource_func.assert_called_once_with(params={'query_params': {'filter': 'name:123', 'offset': 0, 'limit': 10}}) def test_iterate_over_pageable_resource_should_preserve_limit(self): resource_func = mock.Mock(side_effect=[ {'items': ['foo']}, {'items': []}, ]) items = iterate_over_pageable_resource(resource_func, {'query_params': {'limit': 1}}) assert ['foo'] == list(items) resource_func.assert_has_calls([ call(params={'query_params': {'offset': 0, 'limit': 1}}) ]) def test_iterate_over_pageable_resource_should_preserve_offset(self): resource_func = mock.Mock(side_effect=[ {'items': ['foo']}, {'items': []}, ]) items = iterate_over_pageable_resource(resource_func, {'query_params': {'offset': 3}}) assert ['foo'] == list(items) resource_func.assert_has_calls([ call(params={'query_params': {'offset': 3, 'limit': 10}}), ]) def test_iterate_over_pageable_resource_should_pass_with_string_offset_and_limit(self): resource_func = mock.Mock(side_effect=[ {'items': ['foo']}, {'items': []}, ]) items = iterate_over_pageable_resource(resource_func, {'query_params': {'offset': '1', 'limit': '1'}}) assert ['foo'] == list(items) resource_func.assert_has_calls([ call(params={'query_params': {'offset': '1', 'limit': '1'}}), call(params={'query_params': {'offset': 2, 'limit': '1'}}) ]) def test_iterate_over_pageable_resource_raises_exception_when_server_returned_more_items_than_requested(self): resource_func = mock.Mock(side_effect=[ {'items': ['foo', 'redundant_bar']}, {'items': []}, ]) with pytest.raises(FtdUnexpectedResponse): list(iterate_over_pageable_resource(resource_func, {'query_params': {'offset': '1', 'limit': '1'}})) resource_func.assert_has_calls([ call(params={'query_params': {'offset': '1', 'limit': '1'}}) ]) class TestOperationCheckerClass(unittest.TestCase): def setUp(self): self._checker = OperationChecker def test_is_add_operation_positive(self): operation_name = OperationNamePrefix.ADD + "Object" operation_spec = {OperationField.METHOD: HTTPMethod.POST} assert self._checker.is_add_operation(operation_name, operation_spec) def test_is_add_operation_wrong_method_in_spec(self): operation_name = OperationNamePrefix.ADD + "Object" operation_spec = {OperationField.METHOD: HTTPMethod.GET} assert not self._checker.is_add_operation(operation_name, operation_spec) def test_is_add_operation_negative_wrong_operation_name(self): operation_name = OperationNamePrefix.GET + "Object" operation_spec = {OperationField.METHOD: HTTPMethod.POST} assert not self._checker.is_add_operation(operation_name, operation_spec) def test_is_edit_operation_positive(self): operation_name = OperationNamePrefix.EDIT + "Object" operation_spec = {OperationField.METHOD: HTTPMethod.PUT} assert self._checker.is_edit_operation(operation_name, operation_spec) def test_is_edit_operation_wrong_method_in_spec(self): operation_name = OperationNamePrefix.EDIT + "Object" operation_spec = {OperationField.METHOD: HTTPMethod.GET} assert not self._checker.is_edit_operation(operation_name, operation_spec) def test_is_edit_operation_negative_wrong_operation_name(self): operation_name = OperationNamePrefix.GET + "Object" operation_spec = {OperationField.METHOD: HTTPMethod.PUT} assert not self._checker.is_edit_operation(operation_name, operation_spec) def test_is_delete_operation_positive(self): operation_name = OperationNamePrefix.DELETE + "Object" operation_spec = {OperationField.METHOD: HTTPMethod.DELETE} self.assertTrue( self._checker.is_delete_operation(operation_name, operation_spec) ) def test_is_delete_operation_wrong_method_in_spec(self): operation_name = OperationNamePrefix.DELETE + "Object" operation_spec = {OperationField.METHOD: HTTPMethod.GET} assert not self._checker.is_delete_operation(operation_name, operation_spec) def test_is_delete_operation_negative_wrong_operation_name(self): operation_name = OperationNamePrefix.GET + "Object" operation_spec = {OperationField.METHOD: HTTPMethod.DELETE} assert not self._checker.is_delete_operation(operation_name, operation_spec) def test_is_get_list_operation_positive(self): operation_name = OperationNamePrefix.GET + "Object" operation_spec = { OperationField.METHOD: HTTPMethod.GET, OperationField.RETURN_MULTIPLE_ITEMS: True } assert self._checker.is_get_list_operation(operation_name, operation_spec) def test_is_get_list_operation_wrong_method_in_spec(self): operation_name = OperationNamePrefix.GET + "Object" operation_spec = { OperationField.METHOD: HTTPMethod.POST, OperationField.RETURN_MULTIPLE_ITEMS: True } assert not self._checker.is_get_list_operation(operation_name, operation_spec) def test_is_get_list_operation_does_not_return_list(self): operation_name = OperationNamePrefix.GET + "Object" operation_spec = { OperationField.METHOD: HTTPMethod.GET, OperationField.RETURN_MULTIPLE_ITEMS: False } assert not self._checker.is_get_list_operation(operation_name, operation_spec) def test_is_get_operation_positive(self): operation_name = OperationNamePrefix.GET + "Object" operation_spec = { OperationField.METHOD: HTTPMethod.GET, OperationField.RETURN_MULTIPLE_ITEMS: False } self.assertTrue( self._checker.is_get_operation(operation_name, operation_spec) ) def test_is_get_operation_wrong_method_in_spec(self): operation_name = OperationNamePrefix.ADD + "Object" operation_spec = { OperationField.METHOD: HTTPMethod.POST, OperationField.RETURN_MULTIPLE_ITEMS: False } assert not self._checker.is_get_operation(operation_name, operation_spec) def test_is_get_operation_negative_when_returns_multiple(self): operation_name = OperationNamePrefix.GET + "Object" operation_spec = { OperationField.METHOD: HTTPMethod.GET, OperationField.RETURN_MULTIPLE_ITEMS: True } assert not self._checker.is_get_operation(operation_name, operation_spec) def test_is_upsert_operation_positive(self): operation_name = OperationNamePrefix.UPSERT + "Object" assert self._checker.is_upsert_operation(operation_name) def test_is_upsert_operation_with_wrong_operation_name(self): for op_type in [OperationNamePrefix.ADD, OperationNamePrefix.GET, OperationNamePrefix.EDIT, OperationNamePrefix.DELETE]: operation_name = op_type + "Object" assert not self._checker.is_upsert_operation(operation_name) def test_is_find_by_filter_operation(self): operation_name = OperationNamePrefix.GET + "Object" operation_spec = { OperationField.METHOD: HTTPMethod.GET, OperationField.RETURN_MULTIPLE_ITEMS: True } params = {ParamName.FILTERS: 1} self.assertTrue( self._checker.is_find_by_filter_operation( operation_name, params, operation_spec ) ) def test_is_find_by_filter_operation_negative_when_filters_empty(self): operation_name = OperationNamePrefix.GET + "Object" operation_spec = { OperationField.METHOD: HTTPMethod.GET, OperationField.RETURN_MULTIPLE_ITEMS: True } params = {ParamName.FILTERS: None} assert not self._checker.is_find_by_filter_operation( operation_name, params, operation_spec ) params = {} assert not self._checker.is_find_by_filter_operation( operation_name, params, operation_spec ) def test_is_upsert_operation_supported_operation(self): get_list_op_spec = {OperationField.METHOD: HTTPMethod.GET, OperationField.RETURN_MULTIPLE_ITEMS: True} add_op_spec = {OperationField.METHOD: HTTPMethod.POST} edit_op_spec = {OperationField.METHOD: HTTPMethod.PUT} assert self._checker.is_upsert_operation_supported({'getList': get_list_op_spec, 'edit': edit_op_spec}) assert self._checker.is_upsert_operation_supported( {'add': add_op_spec, 'getList': get_list_op_spec, 'edit': edit_op_spec}) assert not self._checker.is_upsert_operation_supported({'getList': get_list_op_spec}) assert not self._checker.is_upsert_operation_supported({'edit': edit_op_spec}) assert not self._checker.is_upsert_operation_supported({'getList': get_list_op_spec, 'add': add_op_spec})