ansible-collections/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2025-07-27 15:11:23 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

FTD modules: upsert functionality and bug fixes (#47747)

Browse source 
* FTD modules: bug fixes and upsert functionality

* Fix sanity checks

* Fix unit tests for Python 2.6

* Log status code for login/logout

* Use string formatting in logging
This commit is contained in:
Anton Nikulin 2018-11-16 00:25:36 -06:00 • committed by Deepak Agrawal
commit 9770ac70f9
15 changed files with 2232 additions and 547 deletions
Download patch file Download diff file Expand all files Collapse all files

7
test/units/module_utils/network/ftd/test_common.py
View file

@ -70,6 +70,13 @@ def test_equal_objects_return_true_with_equal_objects():
)
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}},

485
test/units/module_utils/network/ftd/test_configuration.py
View file

@ -16,34 +16,81 @@
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
#
import json
import unittest
import pytest
from units.compat import mock
from units.compat.mock import call, patch
from ansible.module_utils.network.ftd.configuration import iterate_over_pageable_resource, BaseConfigurationResource
from ansible.module_utils.network.ftd.common import HTTPMethod, FtdUnexpectedResponse
from ansible.module_utils.network.ftd.configuration import iterate_over_pageable_resource, BaseConfigurationResource, \
OperationChecker, OperationNamePrefix, ParamName, QueryParams
from ansible.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.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
@patch.object(BaseConfigurationResource, 'send_request')
def test_get_objects_by_filter_with_multiple_filters(self, send_request_mock):
return connection_instance
@patch.object(BaseConfigurationResource, '_send_request')
def test_get_objects_by_filter_with_multiple_filters(self, send_request_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'}}
]
resource = BaseConfigurationResource(None)
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': []}]
assert objects == resource.get_objects_by_filter('/objects', {})
# 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': []}]
assert [objects[0]] == resource.get_objects_by_filter('/objects', {'name': 'obj1'})
# 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': []}]
assert [objects[1]] == resource.get_objects_by_filter('/objects',
{'type': 1, 'foo': {'bar': 'buz'}})
# 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: {'type': 1, 'foo': {'bar': 'buz'}}}))
@patch.object(BaseConfigurationResource, 'send_request')
def test_get_objects_by_filter_with_multiple_responses(self, send_request_mock):
send_request_mock.assert_has_calls(
[
mock.call('/object/', 'get', {}, {},
{QueryParams.FILTER: "foo:{'bar': 'buz'};type:1", 'limit': 10, 'offset': 0})
]
)
@patch.object(BaseConfigurationResource, '_send_request')
def test_get_objects_by_filter_with_multiple_responses(self, send_request_mock, connection_mock):
send_request_mock.side_effect = [
{'items': [
{'name': 'obj1', 'type': 'foo'},
@ -54,11 +101,204 @@ class TestBaseConfigurationResource(object):
]},
{'items': []}
]
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', {}, {},
{QueryParams.FILTER: "type:foo", 'limit': 10, 'offset': 0})
]
)
resource = BaseConfigurationResource(None)
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', {}, {},
{QueryParams.FILTER: "type:foo", 'limit': 2, 'offset': 0}),
mock.call('/object/', 'get', {}, {},
{QueryParams.FILTER: "type:foo", 'limit': 2, 'offset': 2})
]
)
assert [{'name': 'obj1', 'type': 'foo'}, {'name': 'obj3', 'type': 'foo'}] == resource.get_objects_by_filter(
'/objects', {'type': 'foo'})
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']}}
class TestIterateOverPageableResource(object):
@ -66,7 +306,7 @@ 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)
items = iterate_over_pageable_resource(resource_func, {'query_params': {}})
assert [] == list(items)
@ -76,33 +316,37 @@ class TestIterateOverPageableResource(object):
{'items': []},
])
items = iterate_over_pageable_resource(resource_func)
items = iterate_over_pageable_resource(resource_func, {'query_params': {}})
assert ['foo', 'bar'] == list(items)
resource_func.assert_has_calls([
call(query_params={'offset': 0, 'limit': 10}),
call(query_params={'offset': 10, 'limit': 10})
call(params={'query_params': {'offset': 0, 'limit': 10}})
])
def test_iterate_over_pageable_resource_with_multiple_pages(self):
resource_func = mock.Mock(side_effect=[
objects = [
{'items': ['foo']},
{'items': ['bar']},
{'items': ['buzz']},
{'items': []},
])
]
resource_func = mock.Mock(side_effect=objects)
items = iterate_over_pageable_resource(resource_func)
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, {'filter': 'name:123'})
items = iterate_over_pageable_resource(resource_func, {'query_params': {'filter': 'name:123'}})
assert [] == list(items)
resource_func.assert_called_once_with(query_params={'filter': 'name:123', 'offset': 0, 'limit': 10})
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=[
@ -110,12 +354,11 @@ class TestIterateOverPageableResource(object):
{'items': []},
])
items = iterate_over_pageable_resource(resource_func, {'limit': 1})
items = iterate_over_pageable_resource(resource_func, {'query_params': {'limit': 1}})
assert ['foo'] == list(items)
resource_func.assert_has_calls([
call(query_params={'offset': 0, 'limit': 1}),
call(query_params={'offset': 1, 'limit': 1})
call(params={'query_params': {'offset': 0, 'limit': 1}})
])
def test_iterate_over_pageable_resource_should_preserve_offset(self):
@ -124,12 +367,11 @@ class TestIterateOverPageableResource(object):
{'items': []},
])
items = iterate_over_pageable_resource(resource_func, {'offset': 3})
items = iterate_over_pageable_resource(resource_func, {'query_params': {'offset': 3}})
assert ['foo'] == list(items)
resource_func.assert_has_calls([
call(query_params={'offset': 3, 'limit': 10}),
call(query_params={'offset': 13, 'limit': 10})
call(params={'query_params': {'offset': 3, 'limit': 10}}),
])
def test_iterate_over_pageable_resource_should_pass_with_string_offset_and_limit(self):
@ -138,10 +380,191 @@ class TestIterateOverPageableResource(object):
{'items': []},
])
items = iterate_over_pageable_resource(resource_func, {'offset': '1', 'limit': '1'})
items = iterate_over_pageable_resource(resource_func, {'query_params': {'offset': '1', 'limit': '1'}})
assert ['foo'] == list(items)
resource_func.assert_has_calls([
call(query_params={'offset': '1', 'limit': '1'}),
call(query_params={'offset': 2, 'limit': '1'})
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
)
@patch.object(OperationChecker, "is_add_operation")
@patch.object(OperationChecker, "is_edit_operation")
@patch.object(OperationChecker, "is_get_list_operation")
def test_is_upsert_operation_supported_operation(self, is_add_mock, is_edit_mock, is_get_list_mock):
operations_spec = {
'add': 1,
'edit': 1,
'getList': 1
}
is_add_mock.side_effect = [1, 0, 0]
is_edit_mock.side_effect = [1, 0, 0]
is_get_list_mock.side_effect = [1, 0, 0]
assert self._checker.is_upsert_operation_supported(operations_spec)
is_add_mock.side_effect = [1, 0, 0]
is_edit_mock.side_effect = [0, 1, 0]
is_get_list_mock.side_effect = [0, 0, 0]
assert not self._checker.is_upsert_operation_supported(operations_spec)
is_add_mock.side_effect = [1, 0, 0]
is_edit_mock.side_effect = [0, 0, 0]
is_get_list_mock.side_effect = [1, 0, 0]
assert not self._checker.is_upsert_operation_supported(operations_spec)

183
test/units/module_utils/network/ftd/test_fdm_swagger_parser.py
View file

@ -39,7 +39,7 @@ base = {
"$ref": "#/definitions/FQDNDNSResolution"},
"id": {"type": "string"},
"type": {"type": "string", "default": "networkobject"}},
"required": ["subType", "type", "value"]},
"required": ["subType", "type", "value", "name"]},
"NetworkObjectWrapper": {
"allOf": [{"$ref": "#/definitions/NetworkObject"}, {"$ref": "#/definitions/LinksWrapper"}]}
},
@ -140,14 +140,16 @@ class TestFdmSwaggerParser(unittest.TestCase):
'type': 'string'
}
}
}
},
'returnMultipleItems': True
},
'addNetworkObject': {
'method': HTTPMethod.POST,
'url': '/api/fdm/v2/object/networks',
'modelName': 'NetworkObject',
'parameters': {'path': {},
'query': {}}
'query': {}},
'returnMultipleItems': False
},
'getNetworkObject': {
'method': HTTPMethod.GET,
@ -161,7 +163,8 @@ class TestFdmSwaggerParser(unittest.TestCase):
}
},
'query': {}
}
},
'returnMultipleItems': False
},
'editNetworkObject': {
'method': HTTPMethod.PUT,
@ -175,12 +178,13 @@ class TestFdmSwaggerParser(unittest.TestCase):
}
},
'query': {}
}
},
'returnMultipleItems': False
},
'deleteNetworkObject': {
'method': HTTPMethod.DELETE,
'url': '/api/fdm/v2/object/networks/{objId}',
'modelName': None,
'modelName': 'NetworkObject',
'parameters': {
'path': {
'objId': {
@ -189,8 +193,173 @@ class TestFdmSwaggerParser(unittest.TestCase):
}
},
'query': {}
}
},
'returnMultipleItems': False
}
}
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
},
'addSomeModel': {
'method': HTTPMethod.POST,
'url': '/v2/path1',
'modelName': 'Model2',
'parameters': {
'path': {},
'query': {}
},
'returnMultipleItems': False
},
'getSomeModel': {
'method': HTTPMethod.GET,
'url': '/v2/path2/{id}',
'modelName': 'Model3',
'returnMultipleItems': False
},
'editSomeModel': {
'method': HTTPMethod.PUT,
'url': '/v2/path2/{id}',
'modelName': 'Model1',
'parameters': {
'path': {},
'query': {}
},
'returnMultipleItems': False
},
'deleteModel3': {
'method': HTTPMethod.DELETE,
'url': '/v2/path2/{id}',
'modelName': 'Model3',
'returnMultipleItems': False
},
'deleteNoneModel': {
'method': HTTPMethod.DELETE,
'url': '/v2/path3',
'modelName': None,
'returnMultipleItems': False
}
}
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']

34
test/units/module_utils/network/ftd/test_fdm_swagger_validator.py
View file

@ -15,7 +15,7 @@
# 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
@ -248,7 +248,7 @@ class TestFdmSwaggerValidator(unittest.TestCase):
def test_path_params_invalid_params(self):
self.url_params_invalid_params(method='validate_path_params', parameters_type='path')
def test_path_params_invalid_params(self):
def test_query_params_invalid_params(self):
self.url_params_invalid_params(method='validate_query_params', parameters_type='query')
@staticmethod
@ -384,7 +384,7 @@ class TestFdmSwaggerValidator(unittest.TestCase):
'someParam': 1.2,
'p_integer': "1",
'p_boolean': "",
'p_number': "2"
'p_number': "2.1"
}
valid, rez = getattr(validator, method)('getNetwork', data)
assert not valid
@ -405,20 +405,10 @@ class TestFdmSwaggerValidator(unittest.TestCase):
'expected_type': 'string',
'actually_value': 1.2
},
{
'path': 'p_integer',
'expected_type': 'integer',
'actually_value': "1"
},
{
'path': 'p_boolean',
'expected_type': 'boolean',
'actually_value': ""
},
{
'path': 'p_number',
'expected_type': 'number',
'actually_value': "2"
}
]
}) == sort_validator_rez(rez)
@ -603,6 +593,15 @@ class TestFdmSwaggerValidator(unittest.TestCase):
assert valid
assert rez is None
def test_pass_no_data_with_no_required_fields(self):
spec = copy.deepcopy(mock_data)
del spec['models']['NetworkObject']['required']
valid, rez = FdmSwaggerValidator(spec).validate_data('getNetworkObjectList', {})
assert valid
assert rez is None
def test_pass_all_fields_with_correct_data(self):
data = {
'id': 'id-di',
@ -818,7 +817,7 @@ class TestFdmSwaggerValidator(unittest.TestCase):
"f_string": False,
"f_number": "1",
"f_boolean": "",
"f_integer": 1.2
"f_integer": "1.2"
}
valid, rez = FdmSwaggerValidator(local_mock_data).validate_data('getdata', invalid_data)
@ -830,11 +829,6 @@ class TestFdmSwaggerValidator(unittest.TestCase):
'expected_type': 'string',
'actually_value': False
},
{
'path': 'f_number',
'expected_type': 'number',
'actually_value': "1"
},
{
'path': 'f_boolean',
'expected_type': 'boolean',
@ -843,7 +837,7 @@ class TestFdmSwaggerValidator(unittest.TestCase):
{
'path': 'f_integer',
'expected_type': 'integer',
'actually_value': 1.2
'actually_value': '1.2'
}
]
}) == sort_validator_rez(rez)

762
test/units/module_utils/network/ftd/test_upsert_functionality.py Normal file
View file

@ -0,0 +1,762 @@
# 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 units.compat import mock
from ansible.module_utils.network.ftd.common import FtdServerError, HTTPMethod, ResponseParams, FtdConfigurationError
from ansible.module_utils.network.ftd.configuration import DUPLICATE_NAME_ERROR_MESSAGE, UNPROCESSABLE_ENTITY_STATUS, \
MULTIPLE_DUPLICATES_FOUND_ERROR, BaseConfigurationResource, FtdInvalidOperationNameError, QueryParams
from ansible.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):
def setUp(self):
conn = mock.MagicMock()
self._resource = BaseConfigurationResource(conn)
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)
self.assertRaises(
FtdConfigurationError,
self._resource._get_operation_name, checker_wrapper(None), operations
)
@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, "edit_object")
@mock.patch("ansible.module_utils.network.ftd.configuration.copy_identity_properties")
@mock.patch("ansible.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.object(BaseConfigurationResource, "get_operation_specs_by_model_name")
@mock.patch("ansible.module_utils.network.ftd.configuration.OperationChecker.is_upsert_operation_supported")
@mock.patch("ansible.module_utils.network.ftd.configuration._extract_model_from_upsert_operation")
def test_is_upsert_operation_supported(self, extract_model_mock, is_upsert_supported_mock, get_operation_spec_mock):
op_name = mock.MagicMock()
result = self._resource.is_upsert_operation_supported(op_name)
assert result == is_upsert_supported_mock.return_value
extract_model_mock.assert_called_once_with(op_name)
get_operation_spec_mock.assert_called_once_with(extract_model_mock.return_value)
is_upsert_supported_mock.assert_called_once_with(get_operation_spec_mock.return_value)
@mock.patch.object(BaseConfigurationResource, "is_upsert_operation_supported")
@mock.patch.object(BaseConfigurationResource, "get_operation_specs_by_model_name")
@mock.patch.object(BaseConfigurationResource, "_add_upserted_object")
@mock.patch.object(BaseConfigurationResource, "_edit_upserted_object")
@mock.patch("ansible.module_utils.network.ftd.configuration._extract_model_from_upsert_operation")
def test_upsert_object_succesfully_added(self, extract_model_mock, edit_mock, add_mock, get_operation_mock,
is_upsert_supported_mock):
op_name = mock.MagicMock()
params = mock.MagicMock()
is_upsert_supported_mock.return_value = True
result = self._resource.upsert_object(op_name, params)
assert result == add_mock.return_value
is_upsert_supported_mock.assert_called_once_with(op_name)
extract_model_mock.assert_called_once_with(op_name)
get_operation_mock.assert_called_once_with(extract_model_mock.return_value)
add_mock.assert_called_once_with(get_operation_mock.return_value, params)
edit_mock.assert_not_called()
@mock.patch.object(BaseConfigurationResource, "is_upsert_operation_supported")
@mock.patch.object(BaseConfigurationResource, "get_operation_specs_by_model_name")
@mock.patch.object(BaseConfigurationResource, "_add_upserted_object")
@mock.patch.object(BaseConfigurationResource, "_edit_upserted_object")
@mock.patch("ansible.module_utils.network.ftd.configuration._extract_model_from_upsert_operation")
def test_upsert_object_succesfully_edited(self, extract_model_mock, edit_mock, add_mock, get_operation_mock,
is_upsert_supported_mock):
op_name = mock.MagicMock()
params = mock.MagicMock()
is_upsert_supported_mock.return_value = True
error = FtdConfigurationError("Obj duplication error")
error.obj = mock.MagicMock()
add_mock.side_effect = error
result = self._resource.upsert_object(op_name, params)
assert result == edit_mock.return_value
is_upsert_supported_mock.assert_called_once_with(op_name)
extract_model_mock.assert_called_once_with(op_name)
get_operation_mock.assert_called_once_with(extract_model_mock.return_value)
add_mock.assert_called_once_with(get_operation_mock.return_value, params)
edit_mock.assert_called_once_with(get_operation_mock.return_value, error.obj, params)
@mock.patch.object(BaseConfigurationResource, "is_upsert_operation_supported")
@mock.patch.object(BaseConfigurationResource, "get_operation_specs_by_model_name")
@mock.patch.object(BaseConfigurationResource, "_add_upserted_object")
@mock.patch.object(BaseConfigurationResource, "_edit_upserted_object")
@mock.patch("ansible.module_utils.network.ftd.configuration._extract_model_from_upsert_operation")
def test_upsert_object_not_supported(self, extract_model_mock, edit_mock, add_mock, get_operation_mock,
is_upsert_supported_mock):
op_name = mock.MagicMock()
params = mock.MagicMock()
is_upsert_supported_mock.return_value = False
self.assertRaises(
FtdInvalidOperationNameError,
self._resource.upsert_object, op_name, params
)
is_upsert_supported_mock.assert_called_once_with(op_name)
extract_model_mock.assert_not_called()
get_operation_mock.assert_not_called()
add_mock.assert_not_called()
edit_mock.assert_not_called()
@mock.patch.object(BaseConfigurationResource, "is_upsert_operation_supported")
@mock.patch.object(BaseConfigurationResource, "get_operation_specs_by_model_name")
@mock.patch.object(BaseConfigurationResource, "_add_upserted_object")
@mock.patch.object(BaseConfigurationResource, "_edit_upserted_object")
@mock.patch("ansible.module_utils.network.ftd.configuration._extract_model_from_upsert_operation")
def test_upsert_object_neither_added_nor_edited(self, extract_model_mock, edit_mock, add_mock, get_operation_mock,
is_upsert_supported_mock):
op_name = mock.MagicMock()
params = mock.MagicMock()
is_upsert_supported_mock.return_value = True
error = FtdConfigurationError("Obj duplication error")
error.obj = mock.MagicMock()
add_mock.side_effect = error
edit_mock.side_effect = FtdConfigurationError("Some object edit error")
self.assertRaises(
FtdConfigurationError,
self._resource.upsert_object, op_name, params
)
is_upsert_supported_mock.assert_called_once_with(op_name)
extract_model_mock.assert_called_once_with(op_name)
get_operation_mock.assert_called_once_with(extract_model_mock.return_value)
add_mock.assert_called_once_with(get_operation_mock.return_value, params)
edit_mock.assert_called_once_with(get_operation_mock.return_value, error.obj, params)
@mock.patch.object(BaseConfigurationResource, "is_upsert_operation_supported")
@mock.patch.object(BaseConfigurationResource, "get_operation_specs_by_model_name")
@mock.patch.object(BaseConfigurationResource, "_add_upserted_object")
@mock.patch.object(BaseConfigurationResource, "_edit_upserted_object")
@mock.patch("ansible.module_utils.network.ftd.configuration._extract_model_from_upsert_operation")
def test_upsert_object_with_fatal_error_during_add(self, extract_model_mock, edit_mock, add_mock,
get_operation_mock, is_upsert_supported_mock):
op_name = mock.MagicMock()
params = mock.MagicMock()
is_upsert_supported_mock.return_value = True
error = FtdConfigurationError("Obj duplication error")
add_mock.side_effect = error
self.assertRaises(
FtdConfigurationError,
self._resource.upsert_object, op_name, params
)
is_upsert_supported_mock.assert_called_once_with(op_name)
extract_model_mock.assert_called_once_with(op_name)
get_operation_mock.assert_called_once_with(extract_model_mock.return_value)
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.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]
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: ADD_RESPONSE
}
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)
connection_mock.send_request.assert_called_once_with(url_path=url,
http_method=HTTPMethod.POST,
path_params=params['path_params'],
query_params={},
body_params=params['data'])
assert ADD_RESPONSE == result
# 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):
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