Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Ansible FTD Module improvements and tests update. #60640

Merged
merged 3 commits into from Sep 3, 2019
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
2 changes: 1 addition & 1 deletion lib/ansible/module_utils/network/ftd/common.py
Expand Up @@ -24,7 +24,7 @@
INVALID_IDENTIFIER_SYMBOLS = r'[^a-zA-Z0-9_]'

IDENTITY_PROPERTIES = ['id', 'version', 'ruleId']
NON_COMPARABLE_PROPERTIES = IDENTITY_PROPERTIES + ['isSystemDefined', 'links']
NON_COMPARABLE_PROPERTIES = IDENTITY_PROPERTIES + ['isSystemDefined', 'links', 'token', 'rulePosition']


class HTTPMethod:
Expand Down
20 changes: 19 additions & 1 deletion lib/ansible/module_utils/network/ftd/configuration.py
Expand Up @@ -208,6 +208,7 @@ def __init__(self, conn, check_mode=False):
self._models_operations_specs_cache = {}
self._check_mode = check_mode
self._operation_checker = OperationChecker
self._system_info = None

def execute_operation(self, op_name, params):
"""
Expand Down Expand Up @@ -281,13 +282,30 @@ def match_filters(filter_params, obj):
filters = params.get(ParamName.FILTERS) or {}
if QueryParams.FILTER not in url_params[ParamName.QUERY_PARAMS] and 'name' in filters:
# most endpoints only support filtering by name, so remaining `filters` are applied on returned objects
url_params[ParamName.QUERY_PARAMS][QueryParams.FILTER] = 'name:%s' % filters['name']
url_params[ParamName.QUERY_PARAMS][QueryParams.FILTER] = self._stringify_name_filter(filters)

item_generator = iterate_over_pageable_resource(
partial(self.send_general_request, operation_name=operation_name), url_params
)
return (i for i in item_generator if match_filters(filters, i))

def _stringify_name_filter(self, filters):
build_version = self.get_build_version()
if build_version >= '6.4.0':
return "fts~%s" % filters['name']
return "name:%s" % filters['name']

def _fetch_system_info(self):
if not self._system_info:
params = {ParamName.PATH_PARAMS: PATH_PARAMS_FOR_DEFAULT_OBJ}
self._system_info = self.send_general_request('getSystemInformation', params)

return self._system_info

def get_build_version(self):
system_info = self._fetch_system_info()
return system_info['databaseInfo']['buildVersion']

def add_object(self, operation_name, params):
def is_duplicate_name_error(err):
return err.code == UNPROCESSABLE_ENTITY_STATUS and DUPLICATE_NAME_ERROR_MESSAGE in str(err)
Expand Down
10 changes: 7 additions & 3 deletions lib/ansible/module_utils/network/ftd/fdm_swagger_client.py
Expand Up @@ -78,6 +78,10 @@ class QueryParams:
FILTER = 'filter'


class PathParams:
OBJ_ID = 'objId'


def _get_model_name_from_url(schema_ref):
path = schema_ref.split('/')
return path[len(path) - 1]
Expand Down Expand Up @@ -515,9 +519,9 @@ def _validate_object(self, status, model, data, path):
def _is_enum(self, model):
return self._is_string_type(model) and PropName.ENUM in model

def _check_enum(self, status, model, data, path):
if data is not None and data not in model[PropName.ENUM]:
self._add_invalid_type_report(status, path, '', PropName.ENUM, data)
def _check_enum(self, status, model, value, path):
if value is not None and value not in model[PropName.ENUM]:
self._add_invalid_type_report(status, path, '', PropName.ENUM, value)

def _add_invalid_type_report(self, status, path, prop_name, expected_type, actually_value):
status[PropName.INVALID_TYPE].append({
Expand Down
41 changes: 39 additions & 2 deletions test/units/module_utils/network/ftd/test_configuration.py
Expand Up @@ -40,13 +40,22 @@ def connection_mock(self, mocker):

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, connection_mock):
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/'
Expand Down Expand Up @@ -88,8 +97,10 @@ def test_get_objects_by_filter_with_multiple_filters(self, send_request_mock, co
]
)

@patch.object(BaseConfigurationResource, '_fetch_system_info')
@patch.object(BaseConfigurationResource, '_send_request')
def test_get_objects_by_filter_with_multiple_responses(self, send_request_mock, connection_mock):
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'},
Expand All @@ -100,6 +111,11 @@ def test_get_objects_by_filter_with_multiple_responses(self, send_request_mock,
]},
{'items': []}
]
fetch_system_info_mock.return_value = {
'databaseInfo': {
'buildVersion': '6.3.0'
}
}
connection_mock.get_operation_spec.return_value = {
'method': HTTPMethod.GET,
'url': '/object/'
Expand Down Expand Up @@ -296,6 +312,27 @@ def test_module_should_fail_if_validation_error_in_all_params(self, connection_m
'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):

Expand Down
2 changes: 1 addition & 1 deletion test/units/module_utils/network/ftd/test_device.py
Expand Up @@ -78,7 +78,7 @@ def test_parse_rommon_file_location(self):
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)
assert 'The ROMMON image must be downloaded from TFTP server' in str(ex.value)


class TestFtd2100Platform(object):
Expand Down
22 changes: 18 additions & 4 deletions test/units/module_utils/network/ftd/test_upsert_functionality.py
Expand Up @@ -39,9 +39,16 @@


class TestUpsertOperationUnitTests(unittest.TestCase):
def setUp(self):

@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()
Expand Down Expand Up @@ -856,10 +863,17 @@ def get_operation_spec(name):

@staticmethod
def _resource_execute_operation(params, connection):
resource = BaseConfigurationResource(connection)
op_name = params['operation']

resp = resource.execute_operation(op_name, params)
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

Expand Down
1 change: 0 additions & 1 deletion test/units/plugins/httpapi/test_ftd.py
@@ -1,5 +1,4 @@
# Copyright (c) 2018 Cisco and/or its affiliates.
# Copyright (c) 2018 Cisco and/or its affiliates.
#
# This file is part of Ansible
#
Expand Down