ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2026-03-29 08:27:34 +00:00
Code Activity

Move modules and module_utils unit tests to correct place (#81)

Browse source 
* Move modules and module_utils unit tests to correct place.

* Update ignore.txt

* Fix imports.

* Fix typos.

* Fix more typos.
This commit is contained in:
Felix Fontein 2020-03-31 10:42:38 +02:00 • committed by GitHub
parent ab3c2120fb
commit be191cce6c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1170 changed files with 732 additions and 751 deletions
Download patch file Download diff file Expand all files Collapse all files

0
tests/unit/plugins/module_utils/network/ftd/__init__.py Normal file
View file

446
tests/unit/plugins/module_utils/network/ftd/test_common.py Normal file
View file

@ -0,0 +1,446 @@
# 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 <http://www.gnu.org/licenses/>.
#
from ansible_collections.community.general.plugins.module_utils.network.ftd.common import equal_objects, delete_ref_duplicates, construct_ansible_facts
# simple objects
def test_equal_objects_return_false_with_different_length():
assert not equal_objects(
{'foo': 1},
{'foo': 1, 'bar': 2}
)
def test_equal_objects_return_false_with_different_fields():
assert not equal_objects(
{'foo': 1},
{'bar': 1}
)
def test_equal_objects_return_false_with_different_value_types():
assert not equal_objects(
{'foo': 1},
{'foo': '1'}
)
def test_equal_objects_return_false_with_different_values():
assert not equal_objects(
{'foo': 1},
{'foo': 2}
)
def test_equal_objects_return_false_with_different_nested_values():
assert not equal_objects(
{'foo': {'bar': 1}},
{'foo': {'bar': 2}}
)
def test_equal_objects_return_false_with_different_list_length():
assert not equal_objects(
{'foo': []},
{'foo': ['bar']}
)
def test_equal_objects_return_true_with_equal_objects():
assert equal_objects(
{'foo': 1, 'bar': 2},
{'bar': 2, 'foo': 1}
)
def test_equal_objects_return_true_with_equal_str_like_values():
assert equal_objects(
{'foo': b'bar'},
{'foo': u'bar'}
)
def test_equal_objects_return_true_with_equal_nested_dicts():
assert equal_objects(
{'foo': {'bar': 1, 'buz': 2}},
{'foo': {'buz': 2, 'bar': 1}}
)
def test_equal_objects_return_true_with_equal_lists():
assert equal_objects(
{'foo': ['bar']},
{'foo': ['bar']}
)
def test_equal_objects_return_true_with_ignored_fields():
assert equal_objects(
{'foo': 1, 'version': '123', 'id': '123123'},
{'foo': 1}
)
# objects with object references
def test_equal_objects_return_true_with_different_ref_ids():
assert not equal_objects(
{'foo': {'id': '1', 'type': 'network', 'ignored_field': 'foo'}},
{'foo': {'id': '2', 'type': 'network', 'ignored_field': 'bar'}}
)
def test_equal_objects_return_true_with_different_ref_types():
assert not equal_objects(
{'foo': {'id': '1', 'type': 'network', 'ignored_field': 'foo'}},
{'foo': {'id': '1', 'type': 'accessRule', 'ignored_field': 'bar'}}
)
def test_equal_objects_return_true_with_same_object_refs():
assert equal_objects(
{'foo': {'id': '1', 'type': 'network', 'ignored_field': 'foo'}},
{'foo': {'id': '1', 'type': 'network', 'ignored_field': 'bar'}}
)
# objects with array of object references
def test_equal_objects_return_false_with_different_array_length():
assert not equal_objects(
{'foo': [
{'id': '1', 'type': 'network', 'ignored_field': 'foo'}
]},
{'foo': []}
)
def test_equal_objects_return_false_with_different_array_order():
assert not equal_objects(
{'foo': [
{'id': '1', 'type': 'network', 'ignored_field': 'foo'},
{'id': '2', 'type': 'network', 'ignored_field': 'bar'}
]},
{'foo': [
{'id': '2', 'type': 'network', 'ignored_field': 'foo'},
{'id': '1', 'type': 'network', 'ignored_field': 'bar'}
]}
)
def test_equal_objects_return_true_with_equal_ref_arrays():
assert equal_objects(
{'foo': [
{'id': '1', 'type': 'network', 'ignored_field': 'foo'}
]},
{'foo': [
{'id': '1', 'type': 'network', 'ignored_field': 'bar'}
]}
)
# objects with nested structures and object references
def test_equal_objects_return_true_with_equal_nested_object_references():
assert equal_objects(
{
'name': 'foo',
'config': {
'version': '1',
'port': {
'name': 'oldPortName',
'type': 'port',
'id': '123'
}
}
},
{
'name': 'foo',
'config': {
'version': '1',
'port': {
'name': 'newPortName',
'type': 'port',
'id': '123'
}
}
}
)
def test_equal_objects_return_false_with_different_nested_object_references():
assert not equal_objects(
{
'name': 'foo',
'config': {
'version': '1',
'port': {
'name': 'oldPortName',
'type': 'port',
'id': '123'
}
}
},
{
'name': 'foo',
'config': {
'version': '1',
'port': {
'name': 'oldPortName',
'type': 'port',
'id': '234'
}
}
}
)
def test_equal_objects_return_true_with_equal_nested_list_of_object_references():
assert equal_objects(
{
'name': 'foo',
'config': {
'version': '1',
'ports': [{
'name': 'oldPortName',
'type': 'port',
'id': '123'
}, {
'name': 'oldPortName2',
'type': 'port',
'id': '234'
}]
}
},
{
'name': 'foo',
'config': {
'version': '1',
'ports': [{
'name': 'newPortName',
'type': 'port',
'id': '123'
}, {
'name': 'newPortName2',
'type': 'port',
'id': '234',
'extraField': 'foo'
}]
}
}
)
def test_equal_objects_return_true_with_reference_list_containing_duplicates():
assert equal_objects(
{
'name': 'foo',
'config': {
'version': '1',
'ports': [{
'name': 'oldPortName',
'type': 'port',
'id': '123'
}, {
'name': 'oldPortName',
'type': 'port',
'id': '123'
}, {
'name': 'oldPortName2',
'type': 'port',
'id': '234'
}]
}
},
{
'name': 'foo',
'config': {
'version': '1',
'ports': [{
'name': 'newPortName',
'type': 'port',
'id': '123'
}, {
'name': 'newPortName2',
'type': 'port',
'id': '234',
'extraField': 'foo'
}]
}
}
)
def test_delete_ref_duplicates_with_none():
assert delete_ref_duplicates(None) is None
def test_delete_ref_duplicates_with_empty_dict():
assert {} == delete_ref_duplicates({})
def test_delete_ref_duplicates_with_simple_object():
data = {
'id': '123',
'name': 'foo',
'type': 'bar',
'values': ['a', 'b']
}
assert data == delete_ref_duplicates(data)
def test_delete_ref_duplicates_with_object_containing_refs():
data = {
'id': '123',
'name': 'foo',
'type': 'bar',
'refs': [
{'id': '123', 'type': 'baz'},
{'id': '234', 'type': 'baz'},
{'id': '234', 'type': 'foo'}
]
}
assert data == delete_ref_duplicates(data)
def test_delete_ref_duplicates_with_object_containing_duplicate_refs():
data = {
'id': '123',
'name': 'foo',
'type': 'bar',
'refs': [
{'id': '123', 'type': 'baz'},
{'id': '123', 'type': 'baz'},
{'id': '234', 'type': 'baz'},
{'id': '234', 'type': 'baz'},
{'id': '234', 'type': 'foo'}
]
}
assert {
'id': '123',
'name': 'foo',
'type': 'bar',
'refs': [
{'id': '123', 'type': 'baz'},
{'id': '234', 'type': 'baz'},
{'id': '234', 'type': 'foo'}
]
} == delete_ref_duplicates(data)
def test_delete_ref_duplicates_with_object_containing_duplicate_refs_in_nested_object():
data = {
'id': '123',
'name': 'foo',
'type': 'bar',
'children': {
'refs': [
{'id': '123', 'type': 'baz'},
{'id': '123', 'type': 'baz'},
{'id': '234', 'type': 'baz'},
{'id': '234', 'type': 'baz'},
{'id': '234', 'type': 'foo'}
]
}
}
assert {
'id': '123',
'name': 'foo',
'type': 'bar',
'children': {
'refs': [
{'id': '123', 'type': 'baz'},
{'id': '234', 'type': 'baz'},
{'id': '234', 'type': 'foo'}
]
}
} == delete_ref_duplicates(data)
def test_construct_ansible_facts_should_make_default_fact_with_name_and_type():
response = {
'id': '123',
'name': 'foo',
'type': 'bar'
}
assert {'bar_foo': response} == construct_ansible_facts(response, {})
def test_construct_ansible_facts_should_not_make_default_fact_with_no_name():
response = {
'id': '123',
'name': 'foo'
}
assert {} == construct_ansible_facts(response, {})
def test_construct_ansible_facts_should_not_make_default_fact_with_no_type():
response = {
'id': '123',
'type': 'bar'
}
assert {} == construct_ansible_facts(response, {})
def test_construct_ansible_facts_should_use_register_as_when_given():
response = {
'id': '123',
'name': 'foo',
'type': 'bar'
}
params = {'register_as': 'fact_name'}
assert {'fact_name': response} == construct_ansible_facts(response, params)
def test_construct_ansible_facts_should_extract_items():
response = {'items': [
{
'id': '123',
'name': 'foo',
'type': 'bar'
}, {
'id': '123',
'name': 'foo',
'type': 'bar'
}
]}
params = {'register_as': 'fact_name'}
assert {'fact_name': response['items']} == construct_ansible_facts(response, params)
def test_construct_ansible_facts_should_ignore_items_with_no_register_as():
response = {'items': [
{
'id': '123',
'name': 'foo',
'type': 'bar'
}, {
'id': '123',
'name': 'foo',
'type': 'bar'
}
]}
assert {} == construct_ansible_facts(response, {})

588
tests/unit/plugins/module_utils/network/ftd/test_configuration.py Normal file
View file

@ -0,0 +1,588 @@
# 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 <http://www.gnu.org/licenses/>.
#
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})

1
tests/unit/plugins/module_utils/network/ftd/test_data/ngfw_with_ex.json Normal file
View file

File diff suppressed because one or more lines are too long

145
tests/unit/plugins/module_utils/network/ftd/test_device.py Normal file
View file

@ -0,0 +1,145 @@
# Copyright (c) 2019 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 <http://www.gnu.org/licenses/>.
#
import pytest
pytest.importorskip("kick")
from ansible_collections.community.general.plugins.module_utils.network.ftd.device import FtdPlatformFactory, FtdModel, FtdAsa5500xPlatform, \
Ftd2100Platform, AbstractFtdPlatform
from ansible_collections.community.general.tests.unit.plugins.modules.network.ftd.test_ftd_install import DEFAULT_MODULE_PARAMS
class TestFtdModel(object):
def test_has_value_should_return_true_for_existing_models(self):
assert FtdModel.FTD_2120 in FtdModel.supported_models()
assert FtdModel.FTD_ASA5516_X in FtdModel.supported_models()
def test_has_value_should_return_false_for_non_existing_models(self):
assert 'nonExistingModel' not in FtdModel.supported_models()
assert None not in FtdModel.supported_models()
class TestFtdPlatformFactory(object):
@pytest.fixture(autouse=True)
def mock_devices(self, mocker):
mocker.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.device.Kp')
mocker.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.device.Ftd5500x')
def test_factory_should_return_corresponding_platform(self):
ftd_platform = FtdPlatformFactory.create(FtdModel.FTD_ASA5508_X, dict(DEFAULT_MODULE_PARAMS))
assert type(ftd_platform) is FtdAsa5500xPlatform
ftd_platform = FtdPlatformFactory.create(FtdModel.FTD_2130, dict(DEFAULT_MODULE_PARAMS))
assert type(ftd_platform) is Ftd2100Platform
def test_factory_should_raise_error_with_not_supported_model(self):
with pytest.raises(ValueError) as ex:
FtdPlatformFactory.create('nonExistingModel', dict(DEFAULT_MODULE_PARAMS))
assert "FTD model 'nonExistingModel' is not supported by this module." == ex.value.args[0]
class TestAbstractFtdPlatform(object):
def test_install_ftd_image_raise_error_on_abstract_class(self):
with pytest.raises(NotImplementedError):
AbstractFtdPlatform().install_ftd_image(dict(DEFAULT_MODULE_PARAMS))
def test_supports_ftd_model_should_return_true_for_supported_models(self):
assert Ftd2100Platform.supports_ftd_model(FtdModel.FTD_2120)
assert FtdAsa5500xPlatform.supports_ftd_model(FtdModel.FTD_ASA5516_X)
def test_supports_ftd_model_should_return_false_for_non_supported_models(self):
assert not AbstractFtdPlatform.supports_ftd_model(FtdModel.FTD_2120)
assert not Ftd2100Platform.supports_ftd_model(FtdModel.FTD_ASA5508_X)
assert not FtdAsa5500xPlatform.supports_ftd_model(FtdModel.FTD_2120)
def test_parse_rommon_file_location(self):
server, path = AbstractFtdPlatform.parse_rommon_file_location('tftp://1.2.3.4/boot/rommon-boot.foo')
assert '1.2.3.4' == server
assert '/boot/rommon-boot.foo' == path
def test_parse_rommon_file_location_should_fail_for_non_tftp_protocol(self):
with pytest.raises(ValueError) as ex:
AbstractFtdPlatform.parse_rommon_file_location('http://1.2.3.4/boot/rommon-boot.foo')
assert 'The ROMMON image must be downloaded from TFTP server' in str(ex.value)
class TestFtd2100Platform(object):
@pytest.fixture
def kp_mock(self, mocker):
return mocker.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.device.Kp')
@pytest.fixture
def module_params(self):
return dict(DEFAULT_MODULE_PARAMS)
def test_install_ftd_image_should_call_kp_module(self, kp_mock, module_params):
ftd = FtdPlatformFactory.create(FtdModel.FTD_2110, module_params)
ftd.install_ftd_image(module_params)
assert kp_mock.called
assert kp_mock.return_value.ssh_console.called
ftd_line = kp_mock.return_value.ssh_console.return_value
assert ftd_line.baseline_fp2k_ftd.called
assert ftd_line.disconnect.called
def test_install_ftd_image_should_call_disconnect_when_install_fails(self, kp_mock, module_params):
ftd_line = kp_mock.return_value.ssh_console.return_value
ftd_line.baseline_fp2k_ftd.side_effect = Exception('Something went wrong')
ftd = FtdPlatformFactory.create(FtdModel.FTD_2120, module_params)
with pytest.raises(Exception):
ftd.install_ftd_image(module_params)
assert ftd_line.baseline_fp2k_ftd.called
assert ftd_line.disconnect.called
class TestFtdAsa5500xPlatform(object):
@pytest.fixture
def asa5500x_mock(self, mocker):
return mocker.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.device.Ftd5500x')
@pytest.fixture
def module_params(self):
return dict(DEFAULT_MODULE_PARAMS)
def test_install_ftd_image_should_call_kp_module(self, asa5500x_mock, module_params):
ftd = FtdPlatformFactory.create(FtdModel.FTD_ASA5508_X, module_params)
ftd.install_ftd_image(module_params)
assert asa5500x_mock.called
assert asa5500x_mock.return_value.ssh_console.called
ftd_line = asa5500x_mock.return_value.ssh_console.return_value
assert ftd_line.rommon_to_new_image.called
assert ftd_line.disconnect.called
def test_install_ftd_image_should_call_disconnect_when_install_fails(self, asa5500x_mock, module_params):
ftd_line = asa5500x_mock.return_value.ssh_console.return_value
ftd_line.rommon_to_new_image.side_effect = Exception('Something went wrong')
ftd = FtdPlatformFactory.create(FtdModel.FTD_ASA5516_X, module_params)
with pytest.raises(Exception):
ftd.install_ftd_image(module_params)
assert ftd_line.rommon_to_new_image.called
assert ftd_line.disconnect.called

379
tests/unit/plugins/module_utils/network/ftd/test_fdm_swagger_parser.py Normal file
View file

@ -0,0 +1,379 @@
# 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 <http://www.gnu.org/licenses/>.
#
import copy
import os
import unittest
from ansible_collections.community.general.plugins.module_utils.network.ftd.common import HTTPMethod
from ansible_collections.community.general.plugins.module_utils.network.ftd.fdm_swagger_client import FdmSwaggerParser
DIR_PATH = os.path.dirname(os.path.realpath(__file__))
TEST_DATA_FOLDER = os.path.join(DIR_PATH, 'test_data')
base = {
'basePath': "/api/fdm/v2",
'definitions': {"NetworkObject": {"type": "object",
"properties": {"version": {"type": "string"}, "name": {"type": "string"},
"description": {"type": "string"},
"subType": {"type": "object",
"$ref": "#/definitions/NetworkObjectType"},
"value": {"type": "string"},
"isSystemDefined": {"type": "boolean"},
"dnsResolution": {"type": "object",
"$ref": "#/definitions/FQDNDNSResolution"},
"id": {"type": "string"},
"type": {"type": "string", "default": "networkobject"}},
"required": ["subType", "type", "value", "name"]},
"NetworkObjectWrapper": {
"allOf": [{"$ref": "#/definitions/NetworkObject"}, {"$ref": "#/definitions/LinksWrapper"}]}
},
'paths': {
"/object/networks": {
"get": {"tags": ["NetworkObject"],
"operationId": "getNetworkObjectList",
"responses": {
"200": {
"description": "",
"schema": {"type": "object",
"title": "NetworkObjectList",
"properties": {
"items": {
"type": "array",
"items": {"$ref": "#/definitions/NetworkObjectWrapper"}},
"paging": {
"$ref": "#/definitions/Paging"}},
"required": ["items", "paging"]}}},
"parameters": [
{"name": "offset", "in": "query", "required": False, "type": "integer"},
{"name": "limit", "in": "query", "required": False, "type": "integer"},
{"name": "sort", "in": "query", "required": False, "type": "string"},
{"name": "filter", "in": "query", "required": False, "type": "string"}]},
"post": {"tags": ["NetworkObject"], "operationId": "addNetworkObject",
"responses": {
"200": {"description": "",
"schema": {"type": "object",
"$ref": "#/definitions/NetworkObjectWrapper"}},
"422": {"description": "",
"schema": {"type": "object", "$ref": "#/definitions/ErrorWrapper"}}},
"parameters": [{"in": "body", "name": "body",
"required": True,
"schema": {"$ref": "#/definitions/NetworkObject"}}]}
},
"/object/networks/{objId}": {
"get": {"tags": ["NetworkObject"], "operationId": "getNetworkObject",
"responses": {"200": {"description": "",
"schema": {"type": "object",
"$ref": "#/definitions/NetworkObjectWrapper"}},
"404": {"description": "",
"schema": {"type": "object",
"$ref": "#/definitions/ErrorWrapper"}}},
"parameters": [{"name": "objId", "in": "path", "required": True,
"type": "string"}]},
"put": {"tags": ["NetworkObject"], "operationId": "editNetworkObject",
"responses": {"200": {"description": "",
"schema": {"type": "object",
"$ref": "#/definitions/NetworkObjectWrapper"}},
"422": {"description": "",
"schema": {"type": "object",
"$ref": "#/definitions/ErrorWrapper"}}},
"parameters": [{"name": "objId", "in": "path", "required": True,
"type": "string"},
{"in": "body", "name": "body", "required": True,
"schema": {"$ref": "#/definitions/NetworkObject"}}]},
"delete": {"tags": ["NetworkObject"], "operationId": "deleteNetworkObject",
"responses": {"204": {"description": ""},
"422": {"description": "",
"schema": {"type": "object",
"$ref": "#/definitions/ErrorWrapper"}}},
"parameters": [{"name": "objId", "in": "path", "required": True,
"type": "string"}]}}}
}
def _get_objects(base_object, key_names):
return dict((_key, base_object[_key]) for _key in key_names)
class TestFdmSwaggerParser(unittest.TestCase):
def test_simple_object(self):
self._data = copy.deepcopy(base)
self.fdm_data = FdmSwaggerParser().parse_spec(self._data)
expected_operations = {
'getNetworkObjectList': {
'method': HTTPMethod.GET,
'url': '/api/fdm/v2/object/networks',
'modelName': 'NetworkObject',
'parameters': {
'path': {},
'query': {
'offset': {
'required': False,
'type': 'integer'
},
'limit': {
'required': False,
'type': 'integer'
},
'sort': {
'required': False,
'type': 'string'
},
'filter': {
'required': False,
'type': 'string'
}
}
},
'returnMultipleItems': True,
"tags": ["NetworkObject"]
},
'addNetworkObject': {
'method': HTTPMethod.POST,
'url': '/api/fdm/v2/object/networks',
'modelName': 'NetworkObject',
'parameters': {'path': {},
'query': {}},
'returnMultipleItems': False,
"tags": ["NetworkObject"]
},
'getNetworkObject': {
'method': HTTPMethod.GET,
'url': '/api/fdm/v2/object/networks/{objId}',
'modelName': 'NetworkObject',
'parameters': {
'path': {
'objId': {
'required': True,
'type': "string"
}
},
'query': {}
},
'returnMultipleItems': False,
"tags": ["NetworkObject"]
},
'editNetworkObject': {
'method': HTTPMethod.PUT,
'url': '/api/fdm/v2/object/networks/{objId}',
'modelName': 'NetworkObject',
'parameters': {
'path': {
'objId': {
'required': True,
'type': "string"
}
},
'query': {}
},
'returnMultipleItems': False,
"tags": ["NetworkObject"]
},
'deleteNetworkObject': {
'method': HTTPMethod.DELETE,
'url': '/api/fdm/v2/object/networks/{objId}',
'modelName': 'NetworkObject',
'parameters': {
'path': {
'objId': {
'required': True,
'type': "string"
}
},
'query': {}
},
'returnMultipleItems': False,
"tags": ["NetworkObject"]
}
}
assert sorted(['NetworkObject', 'NetworkObjectWrapper']) == sorted(self.fdm_data['models'].keys())
assert expected_operations == self.fdm_data['operations']
assert {'NetworkObject': expected_operations} == self.fdm_data['model_operations']
def test_simple_object_with_documentation(self):
api_spec = copy.deepcopy(base)
docs = {
'definitions': {
'NetworkObject': {
'description': 'Description for Network Object',
'properties': {'name': 'Description for name field'}
}
},
'paths': {
'/object/networks': {
'get': {
'description': 'Description for getNetworkObjectList operation',
'parameters': [{'name': 'offset', 'description': 'Description for offset field'}]
},
'post': {'description': 'Description for addNetworkObject operation'}
}
}
}
self.fdm_data = FdmSwaggerParser().parse_spec(api_spec, docs)
assert 'Description for Network Object' == self.fdm_data['models']['NetworkObject']['description']
assert '' == self.fdm_data['models']['NetworkObjectWrapper']['description']
network_properties = self.fdm_data['models']['NetworkObject']['properties']
assert '' == network_properties['id']['description']
assert not network_properties['id']['required']
assert 'Description for name field' == network_properties['name']['description']
assert network_properties['name']['required']
ops = self.fdm_data['operations']
assert 'Description for getNetworkObjectList operation' == ops['getNetworkObjectList']['description']
assert 'Description for addNetworkObject operation' == ops['addNetworkObject']['description']
assert '' == ops['deleteNetworkObject']['description']
get_op_params = ops['getNetworkObjectList']['parameters']
assert 'Description for offset field' == get_op_params['query']['offset']['description']
assert '' == get_op_params['query']['limit']['description']
def test_model_operations_should_contain_all_operations(self):
data = {
'basePath': '/v2/',
'definitions': {
'Model1': {"type": "object"},
'Model2': {"type": "object"},
'Model3': {"type": "object"}
},
'paths': {
'path1': {
'get': {
'operationId': 'getSomeModelList',
"responses": {
"200": {"description": "",
"schema": {"type": "object",
"title": "NetworkObjectList",
"properties": {
"items": {
"type": "array",
"items": {
"$ref": "#/definitions/Model1"
}
}
}}
}
}
},
"post": {
"operationId": "addSomeModel",
"parameters": [{"in": "body",
"name": "body",
"schema": {"$ref": "#/definitions/Model2"}
}]}
},
'path2/{id}': {
"get": {"operationId": "getSomeModel",
"responses": {"200": {"description": "",
"schema": {"type": "object",
"$ref": "#/definitions/Model3"}},
}
},
"put": {"operationId": "editSomeModel",
"parameters": [{"in": "body",
"name": "body",
"schema": {"$ref": "#/definitions/Model1"}}
]},
"delete": {
"operationId": "deleteModel3",
}},
'path3': {
"delete": {
"operationId": "deleteNoneModel",
}
}
}
}
expected_operations = {
'getSomeModelList': {
'method': HTTPMethod.GET,
'url': '/v2/path1',
'modelName': 'Model1',
'returnMultipleItems': True,
'tags': []
},
'addSomeModel': {
'method': HTTPMethod.POST,
'url': '/v2/path1',
'modelName': 'Model2',
'parameters': {
'path': {},
'query': {}
},
'returnMultipleItems': False,
'tags': []
},
'getSomeModel': {
'method': HTTPMethod.GET,
'url': '/v2/path2/{id}',
'modelName': 'Model3',
'returnMultipleItems': False,
'tags': []
},
'editSomeModel': {
'method': HTTPMethod.PUT,
'url': '/v2/path2/{id}',
'modelName': 'Model1',
'parameters': {
'path': {},
'query': {}
},
'returnMultipleItems': False,
'tags': []
},
'deleteModel3': {
'method': HTTPMethod.DELETE,
'url': '/v2/path2/{id}',
'modelName': 'Model3',
'returnMultipleItems': False,
'tags': []
},
'deleteNoneModel': {
'method': HTTPMethod.DELETE,
'url': '/v2/path3',
'modelName': None,
'returnMultipleItems': False,
'tags': []
}
}
fdm_data = FdmSwaggerParser().parse_spec(data)
assert sorted(['Model1', 'Model2', 'Model3']) == sorted(fdm_data['models'].keys())
assert expected_operations == fdm_data['operations']
assert {
'Model1': {
'getSomeModelList': expected_operations['getSomeModelList'],
'editSomeModel': expected_operations['editSomeModel'],
},
'Model2': {
'addSomeModel': expected_operations['addSomeModel']
},
'Model3': {
'getSomeModel': expected_operations['getSomeModel'],
'deleteModel3': expected_operations['deleteModel3']
},
None: {
'deleteNoneModel': expected_operations['deleteNoneModel']
}
} == fdm_data['model_operations']

1120
tests/unit/plugins/module_utils/network/ftd/test_fdm_swagger_validator.py Normal file
View file

File diff suppressed because it is too large Load diff

75
tests/unit/plugins/module_utils/network/ftd/test_fdm_swagger_with_real_data.py Normal file
View file

@ -0,0 +1,75 @@
import json
import os
import unittest
from ansible_collections.community.general.plugins.module_utils.network.ftd.fdm_swagger_client import FdmSwaggerValidator, FdmSwaggerParser
DIR_PATH = os.path.dirname(os.path.realpath(__file__))
TEST_DATA_FOLDER = os.path.join(DIR_PATH, 'test_data')
class TestFdmSwagger(unittest.TestCase):
def setUp(self):
self.init_mock_data()
def init_mock_data(self):
with open(os.path.join(TEST_DATA_FOLDER, 'ngfw_with_ex.json'), 'rb') as f:
self.base_data = json.loads(f.read().decode('utf-8'))
def test_with_all_data(self):
fdm_data = FdmSwaggerParser().parse_spec(self.base_data)
validator = FdmSwaggerValidator(fdm_data)
models = fdm_data['models']
operations = fdm_data['operations']
invalid = set({})
for operation in operations:
model_name = operations[operation]['modelName']
method = operations[operation]['method']
if method != 'get' and model_name in models:
if 'example' in models[model_name]:
example = models[model_name]['example']
try:
valid, rez = validator.validate_data(operation, example)
assert valid
except Exception:
invalid.add(model_name)
assert invalid == set(['TCPPortObject',
'UDPPortObject',
'ICMPv4PortObject',
'ICMPv6PortObject',
'StandardAccessList',
'ExtendedAccessList',
'ASPathList',
'RouteMap',
'StandardCommunityList',
'ExpandedCommunityList',
'IPV4PrefixList',
'IPV6PrefixList',
'PolicyList',
'SyslogServer',
'HAConfiguration',
'TestIdentitySource'])
def test_parse_all_data(self):
self.fdm_data = FdmSwaggerParser().parse_spec(self.base_data)
operations = self.fdm_data['operations']
without_model_name = []
expected_operations_counter = 0
for key in self.base_data['paths']:
operation = self.base_data['paths'][key]
for dummy in operation:
expected_operations_counter += 1
for key in operations:
operation = operations[key]
if not operation['modelName']:
without_model_name.append(operation['url'])
if operation['modelName'] == '_File' and 'download' not in operation['url']:
self.fail('File type can be defined for download operation only')
assert sorted(['/api/fdm/v2/operational/deploy/{objId}', '/api/fdm/v2/action/upgrade']) == sorted(
without_model_name)
assert sorted(self.fdm_data['model_operations'][None].keys()) == sorted(['deleteDeployment', 'startUpgrade'])
assert expected_operations_counter == len(operations)

886
tests/unit/plugins/module_utils/network/ftd/test_upsert_functionality.py Normal file
View file

@ -0,0 +1,886 @@
# 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 <http://www.gnu.org/licenses/>.
#
from __future__ import absolute_import
import copy
import json
import unittest
import pytest
from ansible_collections.community.general.tests.unit.compat import mock
from ansible_collections.community.general.plugins.module_utils.network.ftd.common import FtdServerError, HTTPMethod, ResponseParams, FtdConfigurationError
from ansible_collections.community.general.plugins.module_utils.network.ftd.configuration import DUPLICATE_NAME_ERROR_MESSAGE, UNPROCESSABLE_ENTITY_STATUS, \
MULTIPLE_DUPLICATES_FOUND_ERROR, BaseConfigurationResource, FtdInvalidOperationNameError, QueryParams, \
ADD_OPERATION_NOT_SUPPORTED_ERROR, ParamName
from ansible_collections.community.general.plugins.module_utils.network.ftd.fdm_swagger_client import ValidationError
ADD_RESPONSE = {'status': 'Object added'}
EDIT_RESPONSE = {'status': 'Object edited'}
DELETE_RESPONSE = {'status': 'Object deleted'}
GET_BY_FILTER_RESPONSE = [{'name': 'foo', 'description': 'bar'}]
ARBITRARY_RESPONSE = {'status': 'Arbitrary request sent'}
class TestUpsertOperationUnitTests(unittest.TestCase):
@mock.patch.object(BaseConfigurationResource, '_fetch_system_info')
def setUp(self, fetch_system_info_mock):
self._conn = mock.MagicMock()
self._resource = BaseConfigurationResource(self._conn)
fetch_system_info_mock.return_value = {
'databaseInfo': {
'buildVersion': '6.3.0'
}
}
def test_get_operation_name(self):
operation_a = mock.MagicMock()
operation_b = mock.MagicMock()
def checker_wrapper(expected_object):
def checker(obj, *args, **kwargs):
return obj == expected_object
return checker
operations = {
operation_a: "spec",
operation_b: "spec"
}
assert operation_a == self._resource._get_operation_name(checker_wrapper(operation_a), operations)
assert operation_b == self._resource._get_operation_name(checker_wrapper(operation_b), operations)
assert self._resource._get_operation_name(checker_wrapper(None), operations) is None
@mock.patch.object(BaseConfigurationResource, "_get_operation_name")
@mock.patch.object(BaseConfigurationResource, "add_object")
def test_add_upserted_object(self, add_object_mock, get_operation_mock):
model_operations = mock.MagicMock()
params = mock.MagicMock()
add_op_name = get_operation_mock.return_value
assert add_object_mock.return_value == self._resource._add_upserted_object(model_operations, params)
get_operation_mock.assert_called_once_with(
self._resource._operation_checker.is_add_operation,
model_operations)
add_object_mock.assert_called_once_with(add_op_name, params)
@mock.patch.object(BaseConfigurationResource, "_get_operation_name")
@mock.patch.object(BaseConfigurationResource, "add_object")
def test_add_upserted_object_with_no_add_operation(self, add_object_mock, get_operation_mock):
model_operations = mock.MagicMock()
get_operation_mock.return_value = None
with pytest.raises(FtdConfigurationError) as exc_info:
self._resource._add_upserted_object(model_operations, mock.MagicMock())
assert ADD_OPERATION_NOT_SUPPORTED_ERROR in str(exc_info.value)
get_operation_mock.assert_called_once_with(self._resource._operation_checker.is_add_operation, model_operations)
add_object_mock.assert_not_called()
@mock.patch.object(BaseConfigurationResource, "_get_operation_name")
@mock.patch.object(BaseConfigurationResource, "edit_object")
@mock.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.configuration.copy_identity_properties')
@mock.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.configuration._set_default')
def test_edit_upserted_object(self, _set_default_mock, copy_properties_mock, edit_object_mock, get_operation_mock):
model_operations = mock.MagicMock()
existing_object = mock.MagicMock()
params = {
'path_params': {},
'data': {}
}
result = self._resource._edit_upserted_object(model_operations, existing_object, params)
assert result == edit_object_mock.return_value
_set_default_mock.assert_has_calls([
mock.call(params, 'path_params', {}),
mock.call(params, 'data', {})
])
get_operation_mock.assert_called_once_with(
self._resource._operation_checker.is_edit_operation,
model_operations
)
copy_properties_mock.assert_called_once_with(
existing_object,
params['data']
)
edit_object_mock.assert_called_once_with(
get_operation_mock.return_value,
params
)
@mock.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.configuration.OperationChecker.is_upsert_operation_supported')
@mock.patch.object(BaseConfigurationResource, "get_operation_specs_by_model_name")
@mock.patch.object(BaseConfigurationResource, "_find_object_matching_params")
@mock.patch.object(BaseConfigurationResource, "_add_upserted_object")
@mock.patch.object(BaseConfigurationResource, "_edit_upserted_object")
def test_upsert_object_successfully_added(self, edit_mock, add_mock, find_object, get_operation_mock,
is_upsert_supported_mock):
params = mock.MagicMock()
is_upsert_supported_mock.return_value = True
find_object.return_value = None
result = self._resource.upsert_object('upsertFoo', params)
assert result == add_mock.return_value
self._conn.get_model_spec.assert_called_once_with('Foo')
is_upsert_supported_mock.assert_called_once_with(get_operation_mock.return_value)
get_operation_mock.assert_called_once_with('Foo')
find_object.assert_called_once_with('Foo', params)
add_mock.assert_called_once_with(get_operation_mock.return_value, params)
edit_mock.assert_not_called()
@mock.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.configuration.equal_objects')
@mock.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.configuration.OperationChecker.is_upsert_operation_supported')
@mock.patch.object(BaseConfigurationResource, "get_operation_specs_by_model_name")
@mock.patch.object(BaseConfigurationResource, "_find_object_matching_params")
@mock.patch.object(BaseConfigurationResource, "_add_upserted_object")
@mock.patch.object(BaseConfigurationResource, "_edit_upserted_object")
def test_upsert_object_successfully_edited(self, edit_mock, add_mock, find_object, get_operation_mock,
is_upsert_supported_mock, equal_objects_mock):
params = mock.MagicMock()
existing_obj = mock.MagicMock()
is_upsert_supported_mock.return_value = True
find_object.return_value = existing_obj
equal_objects_mock.return_value = False
result = self._resource.upsert_object('upsertFoo', params)
assert result == edit_mock.return_value
self._conn.get_model_spec.assert_called_once_with('Foo')
get_operation_mock.assert_called_once_with('Foo')
is_upsert_supported_mock.assert_called_once_with(get_operation_mock.return_value)
add_mock.assert_not_called()
equal_objects_mock.assert_called_once_with(existing_obj, params[ParamName.DATA])
edit_mock.assert_called_once_with(get_operation_mock.return_value, existing_obj, params)
@mock.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.configuration.equal_objects')
@mock.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.configuration.OperationChecker.is_upsert_operation_supported')
@mock.patch.object(BaseConfigurationResource, "get_operation_specs_by_model_name")
@mock.patch.object(BaseConfigurationResource, "_find_object_matching_params")
@mock.patch.object(BaseConfigurationResource, "_add_upserted_object")
@mock.patch.object(BaseConfigurationResource, "_edit_upserted_object")
def test_upsert_object_returned_without_modifications(self, edit_mock, add_mock, find_object, get_operation_mock,
is_upsert_supported_mock, equal_objects_mock):
params = mock.MagicMock()
existing_obj = mock.MagicMock()
is_upsert_supported_mock.return_value = True
find_object.return_value = existing_obj
equal_objects_mock.return_value = True
result = self._resource.upsert_object('upsertFoo', params)
assert result == existing_obj
self._conn.get_model_spec.assert_called_once_with('Foo')
get_operation_mock.assert_called_once_with('Foo')
is_upsert_supported_mock.assert_called_once_with(get_operation_mock.return_value)
add_mock.assert_not_called()
equal_objects_mock.assert_called_once_with(existing_obj, params[ParamName.DATA])
edit_mock.assert_not_called()
@mock.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.configuration.OperationChecker.is_upsert_operation_supported')
@mock.patch.object(BaseConfigurationResource, "get_operation_specs_by_model_name")
@mock.patch.object(BaseConfigurationResource, "_find_object_matching_params")
@mock.patch.object(BaseConfigurationResource, "_add_upserted_object")
@mock.patch.object(BaseConfigurationResource, "_edit_upserted_object")
def test_upsert_object_not_supported(self, edit_mock, add_mock, find_object, get_operation_mock,
is_upsert_supported_mock):
params = mock.MagicMock()
is_upsert_supported_mock.return_value = False
self.assertRaises(
FtdInvalidOperationNameError,
self._resource.upsert_object, 'upsertFoo', params
)
self._conn.get_model_spec.assert_called_once_with('Foo')
get_operation_mock.assert_called_once_with('Foo')
is_upsert_supported_mock.assert_called_once_with(get_operation_mock.return_value)
find_object.assert_not_called()
add_mock.assert_not_called()
edit_mock.assert_not_called()
@mock.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.configuration.OperationChecker.is_upsert_operation_supported')
@mock.patch.object(BaseConfigurationResource, "get_operation_specs_by_model_name")
@mock.patch.object(BaseConfigurationResource, "_find_object_matching_params")
@mock.patch.object(BaseConfigurationResource, "_add_upserted_object")
@mock.patch.object(BaseConfigurationResource, "_edit_upserted_object")
def test_upsert_object_when_model_not_supported(self, edit_mock, add_mock, find_object, get_operation_mock,
is_upsert_supported_mock):
params = mock.MagicMock()
self._conn.get_model_spec.return_value = None
self.assertRaises(
FtdInvalidOperationNameError,
self._resource.upsert_object, 'upsertNonExisting', params
)
self._conn.get_model_spec.assert_called_once_with('NonExisting')
get_operation_mock.assert_not_called()
is_upsert_supported_mock.assert_not_called()
find_object.assert_not_called()
add_mock.assert_not_called()
edit_mock.assert_not_called()
@mock.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.configuration.equal_objects')
@mock.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.configuration.OperationChecker.is_upsert_operation_supported')
@mock.patch.object(BaseConfigurationResource, "get_operation_specs_by_model_name")
@mock.patch.object(BaseConfigurationResource, "_find_object_matching_params")
@mock.patch.object(BaseConfigurationResource, "_add_upserted_object")
@mock.patch.object(BaseConfigurationResource, "_edit_upserted_object")
def test_upsert_object_with_fatal_error_during_edit(self, edit_mock, add_mock, find_object, get_operation_mock,
is_upsert_supported_mock, equal_objects_mock):
params = mock.MagicMock()
existing_obj = mock.MagicMock()
is_upsert_supported_mock.return_value = True
find_object.return_value = existing_obj
equal_objects_mock.return_value = False
edit_mock.side_effect = FtdConfigurationError("Some object edit error")
self.assertRaises(
FtdConfigurationError,
self._resource.upsert_object, 'upsertFoo', params
)
is_upsert_supported_mock.assert_called_once_with(get_operation_mock.return_value)
self._conn.get_model_spec.assert_called_once_with('Foo')
get_operation_mock.assert_called_once_with('Foo')
find_object.assert_called_once_with('Foo', params)
add_mock.assert_not_called()
edit_mock.assert_called_once_with(get_operation_mock.return_value, existing_obj, params)
@mock.patch('ansible_collections.community.general.plugins.module_utils.network.ftd.configuration.OperationChecker.is_upsert_operation_supported')
@mock.patch.object(BaseConfigurationResource, "get_operation_specs_by_model_name")
@mock.patch.object(BaseConfigurationResource, "_find_object_matching_params")
@mock.patch.object(BaseConfigurationResource, "_add_upserted_object")
@mock.patch.object(BaseConfigurationResource, "_edit_upserted_object")
def test_upsert_object_with_fatal_error_during_add(self, edit_mock, add_mock, find_object, get_operation_mock,
is_upsert_supported_mock):
params = mock.MagicMock()
is_upsert_supported_mock.return_value = True
find_object.return_value = None
error = FtdConfigurationError("Obj duplication error")
add_mock.side_effect = error
self.assertRaises(
FtdConfigurationError,
self._resource.upsert_object, 'upsertFoo', params
)
is_upsert_supported_mock.assert_called_once_with(get_operation_mock.return_value)
self._conn.get_model_spec.assert_called_once_with('Foo')
get_operation_mock.assert_called_once_with('Foo')
find_object.assert_called_once_with('Foo', params)
add_mock.assert_called_once_with(get_operation_mock.return_value, params)
edit_mock.assert_not_called()
# functional tests below
class TestUpsertOperationFunctionalTests(object):
@pytest.fixture(autouse=True)
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
def test_module_should_create_object_when_upsert_operation_and_object_does_not_exist(self, connection_mock):
url = '/test'
operations = {
'getObjectList': {
'method': HTTPMethod.GET,
'url': url,
'modelName': 'Object',
'returnMultipleItems': True},
'addObject': {
'method': HTTPMethod.POST,
'modelName': 'Object',
'url': url},
'editObject': {
'method': HTTPMethod.PUT,
'modelName': 'Object',
'url': '/test/{objId}'},
'otherObjectOperation': {
'method': HTTPMethod.GET,
'modelName': 'Object',
'url': '/test/{objId}',
'returnMultipleItems': False
}
}
def get_operation_spec(name):
return operations[name]
def request_handler(url_path=None, http_method=None, body_params=None, path_params=None, query_params=None):
if http_method == HTTPMethod.POST:
assert url_path == url
assert body_params == params['data']
assert query_params == {}
assert path_params == params['path_params']
return {
ResponseParams.SUCCESS: True,
ResponseParams.RESPONSE: ADD_RESPONSE
}
elif http_method == HTTPMethod.GET:
return {
ResponseParams.SUCCESS: True,
ResponseParams.RESPONSE: {'items': []}
}
else:
assert False
connection_mock.get_operation_spec = get_operation_spec
connection_mock.get_operation_specs_by_model_name.return_value = operations
connection_mock.send_request = request_handler
params = {
'operation': 'upsertObject',
'data': {'id': '123', 'name': 'testObject', 'type': 'object'},
'path_params': {'objId': '123'},
'register_as': 'test_var'
}
result = self._resource_execute_operation(params, connection=connection_mock)
assert ADD_RESPONSE == result
def test_module_should_fail_when_no_model(self, connection_mock):
connection_mock.get_model_spec.return_value = None
params = {
'operation': 'upsertObject',
'data': {'id': '123', 'name': 'testObject', 'type': 'object'},
'path_params': {'objId': '123'},
'register_as': 'test_var'
}
with pytest.raises(FtdInvalidOperationNameError) as exc_info:
self._resource_execute_operation(params, connection=connection_mock)
assert 'upsertObject' == exc_info.value.operation_name
def test_module_should_fail_when_no_add_operation_and_no_object(self, connection_mock):
url = '/test'
operations = {
'getObjectList': {
'method': HTTPMethod.GET,
'url': url,
'modelName': 'Object',
'returnMultipleItems': True},
'editObject': {
'method': HTTPMethod.PUT,
'modelName': 'Object',
'url': '/test/{objId}'},
'otherObjectOperation': {
'method': HTTPMethod.GET,
'modelName': 'Object',
'url': '/test/{objId}',
'returnMultipleItems': False
}}
def get_operation_spec(name):
return operations[name]
connection_mock.get_operation_spec = get_operation_spec
connection_mock.get_operation_specs_by_model_name.return_value = operations
connection_mock.send_request.return_value = {
ResponseParams.SUCCESS: True,
ResponseParams.RESPONSE: {'items': []}
}
params = {
'operation': 'upsertObject',
'data': {'id': '123', 'name': 'testObject', 'type': 'object'},
'path_params': {'objId': '123'},
'register_as': 'test_var'
}
with pytest.raises(FtdConfigurationError) as exc_info:
self._resource_execute_operation(params, connection=connection_mock)
assert ADD_OPERATION_NOT_SUPPORTED_ERROR in str(exc_info.value)
# test when object exists but with different fields(except id)
def test_module_should_update_object_when_upsert_operation_and_object_exists(self, connection_mock):
url = '/test'
obj_id = '456'
version = 'test_version'
url_with_id_templ = '{0}/{1}'.format(url, '{objId}')
new_value = '0000'
old_value = '1111'
params = {
'operation': 'upsertObject',
'data': {'name': 'testObject', 'value': new_value, 'type': 'object'},
'register_as': 'test_var'
}
def request_handler(url_path=None, http_method=None, body_params=None, path_params=None, query_params=None):
if http_method == HTTPMethod.POST:
assert url_path == url
assert body_params == params['data']
assert query_params == {}
assert path_params == {}
return {
ResponseParams.SUCCESS: False,
ResponseParams.RESPONSE: DUPLICATE_NAME_ERROR_MESSAGE,
ResponseParams.STATUS_CODE: UNPROCESSABLE_ENTITY_STATUS
}
elif http_method == HTTPMethod.GET:
is_get_list_req = url_path == url
is_get_req = url_path == url_with_id_templ
assert is_get_req or is_get_list_req
if is_get_list_req:
assert body_params == {}
assert query_params == {QueryParams.FILTER: 'name:testObject', 'limit': 10, 'offset': 0}
assert path_params == {}
elif is_get_req:
assert body_params == {}
assert query_params == {}
assert path_params == {'objId': obj_id}
return {
ResponseParams.SUCCESS: True,
ResponseParams.RESPONSE: {
'items': [
{'name': 'testObject', 'value': old_value, 'type': 'object', 'id': obj_id,
'version': version}
]
}
}
elif http_method == HTTPMethod.PUT:
assert url_path == url_with_id_templ
return {
ResponseParams.SUCCESS: True,
ResponseParams.RESPONSE: body_params
}
else:
assert False
operations = {
'getObjectList': {'method': HTTPMethod.GET, 'url': url, 'modelName': 'Object', 'returnMultipleItems': True},
'addObject': {'method': HTTPMethod.POST, 'modelName': 'Object', 'url': url},
'editObject': {'method': HTTPMethod.PUT, 'modelName': 'Object', 'url': url_with_id_templ},
'otherObjectOperation': {
'method': HTTPMethod.GET,
'modelName': 'Object',
'url': url_with_id_templ,
'returnMultipleItems': False}
}
def get_operation_spec(name):
return operations[name]
connection_mock.get_operation_spec = get_operation_spec
connection_mock.get_operation_specs_by_model_name.return_value = operations
connection_mock.send_request = request_handler
expected_val = {'name': 'testObject', 'value': new_value, 'type': 'object', 'id': obj_id, 'version': version}
result = self._resource_execute_operation(params, connection=connection_mock)
assert expected_val == result
# test when object exists and all fields have the same value
def test_module_should_not_update_object_when_upsert_operation_and_object_exists_with_the_same_fields(
self, connection_mock):
url = '/test'
url_with_id_templ = '{0}/{1}'.format(url, '{objId}')
params = {
'operation': 'upsertObject',
'data': {'name': 'testObject', 'value': '3333', 'type': 'object'},
'register_as': 'test_var'
}
expected_val = copy.deepcopy(params['data'])
expected_val['version'] = 'test_version'
expected_val['id'] = 'test_id'
def request_handler(url_path=None, http_method=None, body_params=None, path_params=None, query_params=None):
if http_method == HTTPMethod.POST:
assert url_path == url
assert body_params == params['data']
assert query_params == {}
assert path_params == {}
return {
ResponseParams.SUCCESS: False,
ResponseParams.RESPONSE: DUPLICATE_NAME_ERROR_MESSAGE,
ResponseParams.STATUS_CODE: UNPROCESSABLE_ENTITY_STATUS
}
elif http_method == HTTPMethod.GET:
assert url_path == url
assert body_params == {}
assert query_params == {QueryParams.FILTER: 'name:testObject', 'limit': 10, 'offset': 0}
assert path_params == {}
return {
ResponseParams.SUCCESS: True,
ResponseParams.RESPONSE: {
'items': [expected_val]
}
}
else:
assert False
operations = {
'getObjectList': {'method': HTTPMethod.GET, 'modelName': 'Object', 'url': url, 'returnMultipleItems': True},
'addObject': {'method': HTTPMethod.POST, 'modelName': 'Object', 'url': url},
'editObject': {'method': HTTPMethod.PUT, 'modelName': 'Object', 'url': url_with_id_templ},
'otherObjectOperation': {
'method': HTTPMethod.GET,
'modelName': 'Object',
'url': url_with_id_templ,
'returnMultipleItems': False}
}
def get_operation_spec(name):
return operations[name]
connection_mock.get_operation_spec = get_operation_spec
connection_mock.get_operation_specs_by_model_name.return_value = operations
connection_mock.send_request = request_handler
result = self._resource_execute_operation(params, connection=connection_mock)
assert expected_val == result
def test_module_should_fail_when_upsert_operation_is_not_supported(self, connection_mock):
connection_mock.get_operation_specs_by_model_name.return_value = {
'addObject': {'method': HTTPMethod.POST, 'modelName': 'Object', 'url': '/test'},
'editObject': {'method': HTTPMethod.PUT, 'modelName': 'Object', 'url': '/test/{objId}'},
'otherObjectOperation': {
'method': HTTPMethod.GET,
'modelName': 'Object',
'url': '/test/{objId}',
'returnMultipleItems': False}
}
operation_name = 'upsertObject'
params = {
'operation': operation_name,
'data': {'id': '123', 'name': 'testObject', 'type': 'object'},
'path_params': {'objId': '123'},
'register_as': 'test_var'
}
result = self._resource_execute_operation_with_expected_failure(
expected_exception_class=FtdInvalidOperationNameError,
params=params, connection=connection_mock)
connection_mock.send_request.assert_not_called()
assert operation_name == result.operation_name
# when create operation raised FtdConfigurationError exception without id and version
def test_module_should_fail_when_upsert_operation_and_failed_create_without_id_and_version(self, connection_mock):
url = '/test'
url_with_id_templ = '{0}/{1}'.format(url, '{objId}')
params = {
'operation': 'upsertObject',
'data': {'name': 'testObject', 'value': '3333', 'type': 'object'},
'register_as': 'test_var'
}
def request_handler(url_path=None, http_method=None, body_params=None, path_params=None, query_params=None):
if http_method == HTTPMethod.POST:
assert url_path == url
assert body_params == params['data']
assert query_params == {}
assert path_params == {}
return {
ResponseParams.SUCCESS: False,
ResponseParams.RESPONSE: DUPLICATE_NAME_ERROR_MESSAGE,
ResponseParams.STATUS_CODE: UNPROCESSABLE_ENTITY_STATUS
}
elif http_method == HTTPMethod.GET:
assert url_path == url
assert body_params == {}
assert query_params == {QueryParams.FILTER: 'name:testObject', 'limit': 10, 'offset': 0}
assert path_params == {}
return {
ResponseParams.SUCCESS: True,
ResponseParams.RESPONSE: {
'items': []
}
}
else:
assert False
operations = {
'getObjectList': {'method': HTTPMethod.GET, 'modelName': 'Object', 'url': url, 'returnMultipleItems': True},
'addObject': {'method': HTTPMethod.POST, 'modelName': 'Object', 'url': url},
'editObject': {'method': HTTPMethod.PUT, 'modelName': 'Object', 'url': url_with_id_templ},
'otherObjectOperation': {
'method': HTTPMethod.GET,
'modelName': 'Object',
'url': url_with_id_templ,
'returnMultipleItems': False}
}
def get_operation_spec(name):
return operations[name]
connection_mock.get_operation_spec = get_operation_spec
connection_mock.get_operation_specs_by_model_name.return_value = operations
connection_mock.send_request = request_handler
result = self._resource_execute_operation_with_expected_failure(
expected_exception_class=FtdServerError,
params=params, connection=connection_mock)
assert result.code == 422
assert result.response == 'Validation failed due to a duplicate name'
def test_module_should_fail_when_upsert_operation_and_failed_update_operation(self, connection_mock):
url = '/test'
obj_id = '456'
version = 'test_version'
url_with_id_templ = '{0}/{1}'.format(url, '{objId}')
error_code = 404
new_value = '0000'
old_value = '1111'
params = {
'operation': 'upsertObject',
'data': {'name': 'testObject', 'value': new_value, 'type': 'object'},
'register_as': 'test_var'
}
error_msg = 'test error'
def request_handler(url_path=None, http_method=None, body_params=None, path_params=None, query_params=None):
if http_method == HTTPMethod.POST:
assert url_path == url
assert body_params == params['data']
assert query_params == {}
assert path_params == {}
return {
ResponseParams.SUCCESS: False,
ResponseParams.RESPONSE: DUPLICATE_NAME_ERROR_MESSAGE,
ResponseParams.STATUS_CODE: UNPROCESSABLE_ENTITY_STATUS
}
elif http_method == HTTPMethod.GET:
is_get_list_req = url_path == url
is_get_req = url_path == url_with_id_templ
assert is_get_req or is_get_list_req
if is_get_list_req:
assert body_params == {}
assert query_params == {QueryParams.FILTER: 'name:testObject', 'limit': 10, 'offset': 0}
elif is_get_req:
assert body_params == {}
assert query_params == {}
assert path_params == {'objId': obj_id}
return {
ResponseParams.SUCCESS: True,
ResponseParams.RESPONSE: {
'items': [
{'name': 'testObject', 'value': old_value, 'type': 'object', 'id': obj_id,
'version': version}
]
}
}
elif http_method == HTTPMethod.PUT:
assert url_path == url_with_id_templ
raise FtdServerError(error_msg, error_code)
else:
assert False
operations = {
'getObjectList': {'method': HTTPMethod.GET, 'modelName': 'Object', 'url': url, 'returnMultipleItems': True},
'addObject': {'method': HTTPMethod.POST, 'modelName': 'Object', 'url': url},
'editObject': {'method': HTTPMethod.PUT, 'modelName': 'Object', 'url': url_with_id_templ},
'otherObjectOperation': {
'method': HTTPMethod.GET,
'modelName': 'Object',
'url': url_with_id_templ,
'returnMultipleItems': False}
}
def get_operation_spec(name):
return operations[name]
connection_mock.get_operation_spec = get_operation_spec
connection_mock.get_operation_specs_by_model_name.return_value = operations
connection_mock.send_request = request_handler
result = self._resource_execute_operation_with_expected_failure(
expected_exception_class=FtdServerError,
params=params, connection=connection_mock)
assert result.code == error_code
assert result.response == error_msg
def test_module_should_fail_when_upsert_operation_and_invalid_data_for_create_operation(self, connection_mock):
new_value = '0000'
params = {
'operation': 'upsertObject',
'data': {'name': 'testObject', 'value': new_value, 'type': 'object'},
'register_as': 'test_var'
}
connection_mock.send_request.assert_not_called()
operations = {
'getObjectList': {
'method': HTTPMethod.GET,
'modelName': 'Object',
'url': 'sd',
'returnMultipleItems': True},
'addObject': {'method': HTTPMethod.POST, 'modelName': 'Object', 'url': 'sdf'},
'editObject': {'method': HTTPMethod.PUT, 'modelName': 'Object', 'url': 'sadf'},
'otherObjectOperation': {
'method': HTTPMethod.GET,
'modelName': 'Object',
'url': 'sdfs',
'returnMultipleItems': False}
}
def get_operation_spec(name):
return operations[name]
connection_mock.get_operation_spec = get_operation_spec
connection_mock.get_operation_specs_by_model_name.return_value = operations
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))
key = 'Invalid data provided'
result = self._resource_execute_operation_with_expected_failure(
expected_exception_class=ValidationError,
params=params, connection=connection_mock)
assert len(result.args) == 1
assert key in result.args[0]
assert json.loads(result.args[0][key]) == {
'invalid_type': [{'actually_value': 1, 'expected_type': 'string', 'path': 'objects[3].id'}],
'required': ['objects[0].type']
}
def test_module_should_fail_when_upsert_operation_and_few_objects_found_by_filter(self, connection_mock):
url = '/test'
url_with_id_templ = '{0}/{1}'.format(url, '{objId}')
sample_obj = {'name': 'testObject', 'value': '3333', 'type': 'object'}
params = {
'operation': 'upsertObject',
'data': sample_obj,
'register_as': 'test_var'
}
def request_handler(url_path=None, http_method=None, body_params=None, path_params=None, query_params=None):
if http_method == HTTPMethod.POST:
assert url_path == url
assert body_params == params['data']
assert query_params == {}
assert path_params == {}
return {
ResponseParams.SUCCESS: False,
ResponseParams.RESPONSE: DUPLICATE_NAME_ERROR_MESSAGE,
ResponseParams.STATUS_CODE: UNPROCESSABLE_ENTITY_STATUS
}
elif http_method == HTTPMethod.GET:
assert url_path == url
assert body_params == {}
assert query_params == {QueryParams.FILTER: 'name:testObject', 'limit': 10, 'offset': 0}
assert path_params == {}
return {
ResponseParams.SUCCESS: True,
ResponseParams.RESPONSE: {
'items': [sample_obj, sample_obj]
}
}
else:
assert False
operations = {
'getObjectList': {'method': HTTPMethod.GET, 'modelName': 'Object', 'url': url, 'returnMultipleItems': True},
'addObject': {'method': HTTPMethod.POST, 'modelName': 'Object', 'url': url},
'editObject': {'method': HTTPMethod.PUT, 'modelName': 'Object', 'url': url_with_id_templ},
'otherObjectOperation': {
'method': HTTPMethod.GET,
'modelName': 'Object',
'url': url_with_id_templ,
'returnMultipleItems': False}
}
def get_operation_spec(name):
return operations[name]
connection_mock.get_operation_spec = get_operation_spec
connection_mock.get_operation_specs_by_model_name.return_value = operations
connection_mock.send_request = request_handler
result = self._resource_execute_operation_with_expected_failure(
expected_exception_class=FtdConfigurationError,
params=params, connection=connection_mock)
assert result.msg is MULTIPLE_DUPLICATES_FOUND_ERROR
assert result.obj is None
@staticmethod
def _resource_execute_operation(params, connection):
with mock.patch.object(BaseConfigurationResource, '_fetch_system_info') as fetch_system_info_mock:
fetch_system_info_mock.return_value = {
'databaseInfo': {
'buildVersion': '6.3.0'
}
}
resource = BaseConfigurationResource(connection)
op_name = params['operation']
resp = resource.execute_operation(op_name, params)
return resp
def _resource_execute_operation_with_expected_failure(self, expected_exception_class, params, connection):
with pytest.raises(expected_exception_class) as ex:
self._resource_execute_operation(params, connection)
# 'ex' here is the instance of '_pytest._code.code.ExceptionInfo' but not <expected_exception_class>
# actual instance of <expected_exception_class> is in the value attribute of 'ex'. That's why we should return
# 'ex.value' here, so it can be checked in a test later.
return ex.value