diff --git a/meta/runtime.yml b/meta/runtime.yml index 6a8be343a..38de511f5 100644 --- a/meta/runtime.yml +++ b/meta/runtime.yml @@ -5,3 +5,5 @@ action_groups: - api - api_facts - api_find_and_modify + - api_info + - api_modify diff --git a/plugins/module_utils/_api_data.py b/plugins/module_utils/_api_data.py index ab771d08b..aaf4923af 100644 --- a/plugins/module_utils/_api_data.py +++ b/plugins/module_utils/_api_data.py @@ -60,10 +60,7 @@ def __init__(self, _dummy=None, can_disable=False, remove_value=None, default=No def split_path(path): - parts = path.split() - if len(parts) == 1 and parts[0] == '': - parts = [] - return parts + return path.split() def join_path(path): diff --git a/plugins/modules/api_info.py b/plugins/modules/api_info.py new file mode 100644 index 000000000..95c9a25ad --- /dev/null +++ b/plugins/modules/api_info.py @@ -0,0 +1,264 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- + +# Copyright: (c) 2022, Felix Fontein +# GNU General Public License v3.0+ https://www.gnu.org/licenses/gpl-3.0.txt + +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +DOCUMENTATION = ''' +--- +module: api_info +author: + - "Felix Fontein (@felixfontein)" +short_description: Retrieve information from API +version_added: 2.1.0 +description: + - Allows to retrieve information for a path using the API. +notes: + - Supports I(check_mode). +extends_documentation_fragment: + - community.routeros.api +options: + path: + description: + - Path to query. + - An example value is C(ip address). This is equivalent to running C(/ip address print) in the RouterOS CLI. + required: true + type: str + choices: + - caps-man aaa + - certificate settings + - interface bridge port + - interface bridge port-controller + - interface bridge port-extender + - interface bridge settings + - interface detect-internet + - interface ethernet + - interface ethernet switch + - interface ethernet switch port + - interface l2tp-server server + - interface ovpn-server server + - interface pptp-server server + - interface sstp-server server + - interface wireless align + - interface wireless cap + - interface wireless sniffer + - interface wireless snooper + - ip accounting + - ip accounting web-access + - ip address + - ip cloud + - ip cloud advanced + - ip dhcp-client + - ip dhcp-client option + - ip dhcp-server + - ip dhcp-server config + - ip dhcp-server lease + - ip dhcp-server network + - ip dns + - ip dns static + - ip firewall address-list + - ip firewall connection tracking + - ip firewall filter + - ip firewall nat + - ip firewall service-port + - ip hotspot service-port + - ip ipsec settings + - ip neighbor discovery-settings + - ip pool + - ip proxy + - ip service + - ip settings + - ip smb + - ip socks + - ip ssh + - ip tftp settings + - ip traffic-flow + - ip traffic-flow ipfix + - ip upnp + - ipv6 dhcp-client + - ipv6 firewall address-list + - ipv6 firewall filter + - ipv6 nd prefix default + - ipv6 settings + - mpls + - mpls ldp + - port firmware + - ppp aaa + - queue interface + - radius incoming + - routing bgp instance + - routing mme + - routing rip + - routing ripng + - snmp + - system clock + - system clock manual + - system identity + - system leds settings + - system note + - system ntp client + - system package update + - system routerboard settings + - system upgrade mirror + - system watchdog + - tool bandwidth-server + - tool e-mail + - tool graphing + - tool mac-server + - tool mac-server mac-winbox + - tool mac-server ping + - tool romon + - tool sms + - tool sniffer + - tool traffic-generator + - user aaa + - user group + unfiltered: + description: + - Whether to output all fields, and not just the ones supported as input for M(community.routeros.api_modify). + - Unfiltered output can contain counters and other state information. + type: bool + default: false + handle_disabled: + description: + - How to handle unset values. + - C(exclamation) prepends the keys with C(!) in the output with value C(null). + - C(null-value) uses the regular key with value C(null). + - C(omit) omits these values from the result. + type: str + choices: + - exclamation + - null-value + - omit + default: exclamation + hide_defaults: + description: + - Whether to hide default values. + type: bool + default: true +seealso: + - module: community.routeros.api + - module: community.routeros.api_modify +''' + +EXAMPLES = ''' +--- +- name: Get IP addresses + community.routeros.api_info: + hostname: "{{ hostname }}" + password: "{{ password }}" + username: "{{ username }}" + path: ip address + register: ip_addresses + +- name: Print data for IP addresses + ansible.builtin.debug: + var: ip_addresses.result +''' + +RETURN = ''' +--- +result: + description: A list of all elements for the current path. + sample: + - '.id': '*1' + actual-interface: bridge + address: "192.168.88.1/24" + comment: defconf + disabled: false + dynamic: false + interface: bridge + invalid: false + network: 192.168.88.0 + type: list + elements: dict + returned: always +''' + +from ansible.module_utils.basic import AnsibleModule +from ansible.module_utils.common.text.converters import to_native + +from ansible_collections.community.routeros.plugins.module_utils.api import ( + api_argument_spec, + check_has_library, + create_api, +) + +from ansible_collections.community.routeros.plugins.module_utils._api_data import ( + PATHS, + join_path, + split_path, +) + +try: + from librouteros.exceptions import LibRouterosError +except Exception: + # Handled in api module_utils + pass + + +def compose_api_path(api, path): + api_path = api.path() + for p in path: + api_path = api_path.join(p) + return api_path + + +def main(): + module_args = dict( + path=dict(type='str', required=True, choices=sorted([join_path(path) for path in PATHS if PATHS[path].fully_understood])), + unfiltered=dict(type='bool', default=False), + handle_disabled=dict(type='str', choices=['exclamation', 'null-value', 'omit'], default='exclamation'), + hide_defaults=dict(type='bool', default=True), + ) + module_args.update(api_argument_spec()) + + module = AnsibleModule( + argument_spec=module_args, + supports_check_mode=True, + ) + + check_has_library(module) + api = create_api(module) + + path = split_path(module.params['path']) + path_info = PATHS.get(tuple(path)) + if path_info is None: + module.fail_json(msg='Path /{path} is not yet supported'.format(path='/'.join(path))) + + handle_disabled = module.params['handle_disabled'] + hide_defaults = module.params['hide_defaults'] + try: + api_path = compose_api_path(api, path) + + result = [] + unfiltered = module.params['unfiltered'] + for entry in api_path: + if not unfiltered: + for k in list(entry): + if k == '.id': + continue + if k not in path_info.fields: + entry.pop(k) + if handle_disabled != 'omit': + for k in path_info.fields: + if k not in entry: + if handle_disabled == 'exclamation': + k = '!%s' % k + entry[k] = None + if hide_defaults: + for k, field_info in path_info.fields.items(): + if field_info.default is not None and entry.get(k) == field_info.default: + entry.pop(k) + result.append(entry) + + module.exit_json(result=result) + except LibRouterosError as e: + module.fail_json(msg=to_native(e)) + + +if __name__ == '__main__': + main() diff --git a/plugins/modules/api_modify.py b/plugins/modules/api_modify.py new file mode 100644 index 000000000..4045f9547 --- /dev/null +++ b/plugins/modules/api_modify.py @@ -0,0 +1,895 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- + +# Copyright: (c) 2022, Felix Fontein +# GNU General Public License v3.0+ https://www.gnu.org/licenses/gpl-3.0.txt + +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +DOCUMENTATION = ''' +--- +module: api_modify +author: + - "Felix Fontein (@felixfontein)" +short_description: Retrieve information from API +version_added: 2.1.0 +description: + - Allows to modify information for a path using the API. +notes: + - Supports I(check_mode). +requirements: + - Needs L(ordereddict,https://pypi.org/project/ordereddict) for Python 2.6 +extends_documentation_fragment: + - community.routeros.api +options: + path: + description: + - Path to query. + - An example value is C(ip address). This is equivalent to running modification commands in C(/ip address) in the RouterOS CLI. + required: true + type: str + choices: + - caps-man aaa + - certificate settings + - interface bridge port + - interface bridge port-controller + - interface bridge port-extender + - interface bridge settings + - interface detect-internet + - interface ethernet + - interface ethernet switch + - interface ethernet switch port + - interface l2tp-server server + - interface ovpn-server server + - interface pptp-server server + - interface sstp-server server + - interface wireless align + - interface wireless cap + - interface wireless sniffer + - interface wireless snooper + - ip accounting + - ip accounting web-access + - ip address + - ip cloud + - ip cloud advanced + - ip dhcp-client + - ip dhcp-client option + - ip dhcp-server + - ip dhcp-server config + - ip dhcp-server lease + - ip dhcp-server network + - ip dns + - ip dns static + - ip firewall address-list + - ip firewall connection tracking + - ip firewall filter + - ip firewall nat + - ip firewall service-port + - ip hotspot service-port + - ip ipsec settings + - ip neighbor discovery-settings + - ip pool + - ip proxy + - ip service + - ip settings + - ip smb + - ip socks + - ip ssh + - ip tftp settings + - ip traffic-flow + - ip traffic-flow ipfix + - ip upnp + - ipv6 dhcp-client + - ipv6 firewall address-list + - ipv6 firewall filter + - ipv6 nd prefix default + - ipv6 settings + - mpls + - mpls ldp + - port firmware + - ppp aaa + - queue interface + - radius incoming + - routing bgp instance + - routing mme + - routing rip + - routing ripng + - snmp + - system clock + - system clock manual + - system identity + - system leds settings + - system note + - system ntp client + - system package update + - system routerboard settings + - system upgrade mirror + - system watchdog + - tool bandwidth-server + - tool e-mail + - tool graphing + - tool mac-server + - tool mac-server mac-winbox + - tool mac-server ping + - tool romon + - tool sms + - tool sniffer + - tool traffic-generator + - user aaa + - user group + data: + description: + - Data to ensure that is present for this path. + - Fields not provided will not be modified. + - If C(.id) appears in an entry, it will be ignored. + required: true + type: list + elements: dict + ensure_order: + description: + - Whether to ensure the same order of the config as present in I(data). + - Requires I(handle_absent_entries=remove). + type: bool + default: false + handle_absent_entries: + description: + - How to handle entries that are present in the current config, but not in I(data). + - C(ignore) ignores them. + - C(remove) removes them. + type: str + choices: + - ignore + - remove + default: ignore + handle_entries_content: + description: + - For a single entry in I(data), this describes how to handle fields that are not mentioned + in that entry, but appear in the actual config. + - If C(ignore), they are not modified. + - If C(remove), they are removed. If at least one cannot be removed, the module will fail. + - If C(remove_as_much_as_possible), all that can be removed will be removed. The ones that + cannot be removed will be kept. + type: str + choices: + - ignore + - remove + - remove_as_much_as_possible + default: ignore +seealso: + - module: community.routeros.api + - module: community.routeros.api_info +''' + +EXAMPLES = ''' +--- +- name: Setup DHCP server networks + # Ensures that we have exactly two DHCP server networks (in the specified order) + community.routeros.api_modify: + path: ip dhcp-server network + handle_absent_entries: remove + handle_entries_content: remove_as_much_as_possible + ensure_order: true + data: + - address: 192.168.88.0/24 + comment: admin network + dns-server: 192.168.88.1 + gateway: 192.168.88.1 + - address: 192.168.1.0/24 + comment: customer network 1 + dns-server: 192.168.1.1 + gateway: 192.168.1.1 + netmask: 24 + +- name: Adjust NAT + community.routeros.api_modify: + hostname: "{{ hostname }}" + password: "{{ password }}" + username: "{{ username }}" + path: ip firewall nat + data: + - action: masquerade + chain: srcnat + comment: NAT to WAN + out-interface-list: WAN + # Three ways to unset values: + # - nothing after `:` + # - "empty" value (null/~/None) + # - prepend '!' + out-interface: + to-addresses: ~ + '!to-ports': +''' + +RETURN = ''' +--- +old_data: + description: + - A list of all elements for the current path before a change was made. + sample: + - '.id': '*1' + actual-interface: bridge + address: "192.168.88.1/24" + comment: defconf + disabled: false + dynamic: false + interface: bridge + invalid: false + network: 192.168.88.0 + type: list + elements: dict + returned: always +new_data: + description: + - A list of all elements for the current path after a change was made. + sample: + - '.id': '*1' + actual-interface: bridge + address: "192.168.1.1/24" + comment: awesome + disabled: false + dynamic: false + interface: bridge + invalid: false + network: 192.168.1.0 + type: list + elements: dict + returned: always +''' + +from collections import defaultdict + +from ansible.module_utils.basic import AnsibleModule, missing_required_lib +from ansible.module_utils.common.text.converters import to_native + +from ansible_collections.community.routeros.plugins.module_utils.api import ( + api_argument_spec, + check_has_library, + create_api, +) + +from ansible_collections.community.routeros.plugins.module_utils._api_data import ( + PATHS, + join_path, + split_path, +) + +HAS_ORDEREDDICT = True +try: + from collections import OrderedDict +except ImportError: + try: + from ordereddict import OrderedDict + except ImportError: + HAS_ORDEREDDICT = False + OrderedDict = dict + +try: + from librouteros.exceptions import LibRouterosError +except Exception: + # Handled in api module_utils + pass + + +def compose_api_path(api, path): + api_path = api.path() + for p in path: + api_path = api_path.join(p) + return api_path + + +def find_modifications(old_entry, new_entry, path_info, module, for_text='', return_none_instead_of_fail=False): + modifications = OrderedDict() + updated_entry = old_entry.copy() + for k, v in new_entry.items(): + if k == '.id': + continue + disabled_k = None + if k.startswith('!'): + disabled_k = k[1:] + elif v is None or v == path_info.fields[k].remove_value: + disabled_k = k + if disabled_k is not None: + if disabled_k in old_entry: + if path_info.fields[disabled_k].remove_value is not None: + modifications[disabled_k] = path_info.fields[disabled_k].remove_value + else: + modifications['!%s' % disabled_k] = '' + del updated_entry[disabled_k] + continue + if k not in old_entry and path_info.fields[k].default == v: + continue + if k not in old_entry or old_entry[k] != v: + modifications[k] = v + updated_entry[k] = v + handle_entries_content = module.params['handle_entries_content'] + if handle_entries_content != 'ignore': + for k in old_entry: + if k == '.id' or k in new_entry or ('!%s' % k) in new_entry or k not in path_info.fields: + continue + field_info = path_info.fields[k] + if field_info.default is not None and field_info.default == old_entry[k]: + continue + if field_info.remove_value is not None and field_info.remove_value == old_entry[k]: + continue + if field_info.can_disable: + if field_info.remove_value is not None: + modifications[k] = field_info.remove_value + else: + modifications['!%s' % k] = '' + del updated_entry[k] + elif field_info.default is not None: + modifications[k] = field_info.default + updated_entry[k] = field_info.default + elif handle_entries_content == 'remove': + if return_none_instead_of_fail: + return None, None + module.fail_json(msg='Key "{key}" cannot be removed{for_text}.'.format(key=k, for_text=for_text)) + return modifications, updated_entry + + +def essentially_same_weight(old_entry, new_entry, path_info, module): + for k, v in new_entry.items(): + if k == '.id': + continue + disabled_k = None + if k.startswith('!'): + disabled_k = k[1:] + elif v is None or v == path_info.fields[k].remove_value: + disabled_k = k + if disabled_k is not None: + if disabled_k in old_entry: + return None + continue + if k not in old_entry and path_info.fields[k].default == v: + continue + if k not in old_entry or old_entry[k] != v: + return None + handle_entries_content = module.params['handle_entries_content'] + weight = 0 + for k in old_entry: + if k == '.id' or k in new_entry or ('!%s' % k) in new_entry or k not in path_info.fields: + continue + field_info = path_info.fields[k] + if field_info.default is not None and field_info.default == old_entry[k]: + continue + if handle_entries_content != 'ignore': + return None + else: + weight += 1 + return weight + + +def format_pk(primary_keys, values): + return ', '.join('{pk}="{value}"'.format(pk=pk, value=value) for pk, value in zip(primary_keys, values)) + + +def polish_entry(entry, path_info, module, for_text): + if '.id' in entry: + entry.pop('.id') + for key, value in entry.items(): + real_key = key + disabled_key = False + if key.startswith('!'): + disabled_key = True + key = key[1:] + if key in entry: + module.fail_json(msg='Not both "{key}" and "!{key}" must appear{for_text}.'.format(key=key, for_text=for_text)) + key_info = path_info.fields.get(key) + if key_info is None: + module.fail_json(msg='Unknown key "{key}"{for_text}.'.format(key=real_key, for_text=for_text)) + if disabled_key: + if not key_info.can_disable: + module.fail_json(msg='Key "!{key}" must not be disabled (leading "!"){for_text}.'.format(key=key, for_text=for_text)) + if value not in (None, '', key_info.remove_value): + module.fail_json(msg='Disabled key "!{key}" must not have a value{for_text}.'.format(key=key, for_text=for_text)) + elif value is None: + if not key_info.can_disable: + module.fail_json(msg='Key "{key}" must not be disabled (value null/~/None){for_text}.'.format(key=key, for_text=for_text)) + for key, field_info in path_info.fields.items(): + if field_info.required and key not in entry: + module.fail_json(msg='Key "{key}" must be present{for_text}.'.format(key=key, for_text=for_text)) + + +def remove_irrelevant_data(entry, path_info): + for k, v in list(entry.items()): + if k == '.id': + continue + if k not in path_info.fields or v is None: + del entry[k] + + +def match_entries(new_entries, old_entries, path_info, module): + matching_old_entries = [None for entry in new_entries] + old_entries = list(old_entries) + matches = [] + handle_absent_entries = module.params['handle_absent_entries'] + if handle_absent_entries == 'remove': + for new_index, (unused, new_entry) in enumerate(new_entries): + for old_index, (unused, old_entry) in enumerate(old_entries): + modifications, unused = find_modifications(old_entry, new_entry, path_info, module, return_none_instead_of_fail=True) + if modifications is not None: + matches.append((new_index, old_index, len(modifications))) + else: + for new_index, (unused, new_entry) in enumerate(new_entries): + for old_index, (unused, old_entry) in enumerate(old_entries): + weight = essentially_same_weight(old_entry, new_entry, path_info, module) + if weight is not None: + matches.append((new_index, old_index, weight)) + matches.sort(key=lambda entry: entry[2]) + for new_index, old_index, rating in matches: + if matching_old_entries[new_index] is not None or old_entries[old_index] is None: + continue + matching_old_entries[new_index], old_entries[old_index] = old_entries[old_index], None + unmatched_old_entries = [index_entry for index_entry in old_entries if index_entry is not None] + return matching_old_entries, unmatched_old_entries + + +def sync_list(module, api, path, path_info): + handle_absent_entries = module.params['handle_absent_entries'] + handle_entries_content = module.params['handle_entries_content'] + if handle_absent_entries == 'remove': + if handle_entries_content == 'ignore': + module.fail_json('For this path, handle_absent_entries=remove cannot be combined with handle_entries_content=ignore') + + stratify_keys = path_info.stratify_keys or () + + data = module.params['data'] + stratified_data = defaultdict(list) + for index, entry in enumerate(data): + for stratify_key in stratify_keys: + if stratify_key not in entry: + module.fail_json( + msg='Every element in data must contain "{stratify_key}". For example, the element at index #{index} does not provide it.'.format( + stratify_key=stratify_key, + index=index + 1, + ) + ) + sks = tuple(entry[stratify_key] for stratify_key in stratify_keys) + polish_entry( + entry, path_info, module, + ' at index {index}'.format(index=index + 1), + ) + stratified_data[sks].append((index, entry)) + stratified_data = dict(stratified_data) + + api_path = compose_api_path(api, path) + + old_data = list(api_path) + stratified_old_data = defaultdict(list) + for index, entry in enumerate(old_data): + sks = tuple(entry[stratify_key] for stratify_key in stratify_keys) + stratified_old_data[sks].append((index, entry)) + stratified_old_data = dict(stratified_old_data) + + create_list = [] + modify_list = [] + remove_list = [] + + new_data = [] + for key, indexed_entries in stratified_data.items(): + old_entries = stratified_old_data.pop(key, []) + + # Try to match indexed_entries with old_entries + matching_old_entries, unmatched_old_entries = match_entries(indexed_entries, old_entries, path_info, module) + + # Update existing entries + for (index, new_entry), potential_old_entry in zip(indexed_entries, matching_old_entries): + if potential_old_entry is not None: + old_index, old_entry = potential_old_entry + modifications, updated_entry = find_modifications( + old_entry, new_entry, path_info, module, + ' at index {index}'.format(index=index + 1), + ) + # Add to modification list if there are changes + if modifications: + modifications['.id'] = old_entry['.id'] + modify_list.append(modifications) + new_data.append((old_index, updated_entry)) + new_entry['.id'] = old_entry['.id'] + else: + create_list.append(new_entry) + + if handle_absent_entries == 'remove': + remove_list.extend(entry['.id'] for index, entry in unmatched_old_entries) + else: + new_data.extend(unmatched_old_entries) + + for key, entries in stratified_old_data.items(): + if handle_absent_entries == 'remove': + remove_list.extend(entry['.id'] for index, entry in entries) + else: + new_data.extend(entries) + + new_data = [entry for index, entry in sorted(new_data, key=lambda entry: entry[0])] + new_data.extend(create_list) + + reorder_list = [] + if module.params['ensure_order']: + for index, entry in enumerate(data): + if '.id' in entry: + def match(current_entry): + return current_entry['.id'] == entry['.id'] + + else: + def match(current_entry): + return current_entry is entry + + current_index = next(current_index + index for current_index, current_entry in enumerate(new_data[index:]) if match(current_entry)) + if current_index != index: + reorder_list.append((index, new_data[current_index], new_data[index])) + new_data.insert(index, new_data.pop(current_index)) + + if not module.check_mode: + if remove_list: + try: + api_path.remove(*remove_list) + except LibRouterosError as e: + module.fail_json( + msg='Error while removing {remove_list}: {error}'.format( + remove_list=', '.join(['ID {id}'.format(id=id) for id in remove_list]), + error=to_native(e), + ) + ) + for modifications in modify_list: + try: + api_path.update(**modifications) + except LibRouterosError as e: + module.fail_json( + msg='Error while modifying for ID {id}: {error}'.format( + id=modifications['.id'], + error=to_native(e), + ) + ) + for entry in create_list: + try: + entry['.id'] = api_path.add(**entry) + except LibRouterosError as e: + module.fail_json( + msg='Error while creating entry: {error}'.format( + error=to_native(e), + ) + ) + for new_index, new_entry, old_entry in reorder_list: + try: + for res in api_path('move', numbers=new_entry['.id'], destination=old_entry['.id']): + pass + except LibRouterosError as e: + module.fail_json( + msg='Error while moving entry ID {element_id} to position #{new_index} ID ({new_id}): {error}'.format( + element_id=new_entry['.id'], + new_index=new_index, + new_id=old_entry['.id'], + error=to_native(e), + ) + ) + + # For sake of completeness, retrieve the full new data: + if modify_list or create_list or reorder_list: + new_data = list(api_path) + + # Remove 'irrelevant' data + for entry in old_data: + remove_irrelevant_data(entry, path_info) + for entry in new_data: + remove_irrelevant_data(entry, path_info) + + # Produce return value + more = {} + if module._diff: + more['diff'] = { + 'before': { + 'data': old_data, + }, + 'after': { + 'data': new_data, + }, + } + module.exit_json( + changed=bool(create_list or modify_list or remove_list or reorder_list), + old_data=old_data, + new_data=new_data, + **more + ) + + +def sync_with_primary_keys(module, api, path, path_info): + primary_keys = path_info.primary_keys + + if path_info.fixed_entries: + if module.params['ensure_order']: + module.fail_json(msg='ensure_order=true cannot be used with this path') + if module.params['handle_absent_entries'] == 'remove': + module.fail_json(msg='handle_absent_entries=remove cannot be used with this path') + + data = module.params['data'] + new_data_by_key = OrderedDict() + for index, entry in enumerate(data): + for primary_key in primary_keys: + if primary_key not in entry: + module.fail_json( + msg='Every element in data must contain "{primary_key}". For example, the element at index #{index} does not provide it.'.format( + primary_key=primary_key, + index=index + 1, + ) + ) + pks = tuple(entry[primary_key] for primary_key in primary_keys) + if pks in new_data_by_key: + module.fail_json( + msg='Every element in data must contain a unique value for {primary_keys}. The value {value} appears at least twice.'.format( + primary_keys=','.join(primary_keys), + value=','.join(['"{0}"'.format(pk) for pk in pks]), + ) + ) + polish_entry( + entry, path_info, module, + ' for {values}'.format( + values=', '.join([ + '{primary_key}="{value}"'.format(primary_key=primary_key, value=value) + for primary_key, value in zip(primary_keys, pks) + ]) + ), + ) + new_data_by_key[pks] = entry + + api_path = compose_api_path(api, path) + + old_data = list(api_path) + old_data_by_key = OrderedDict() + id_by_key = {} + for entry in old_data: + pks = tuple(entry[primary_key] for primary_key in primary_keys) + old_data_by_key[pks] = entry + id_by_key[pks] = entry['.id'] + new_data = [] + + create_list = [] + modify_list = [] + remove_list = [] + remove_keys = [] + handle_absent_entries = module.params['handle_absent_entries'] + for key, old_entry in old_data_by_key.items(): + new_entry = new_data_by_key.pop(key, None) + if new_entry is None: + if handle_absent_entries == 'remove': + remove_list.append(old_entry['.id']) + remove_keys.append(key) + else: + new_data.append(old_entry) + else: + modifications, updated_entry = find_modifications( + old_entry, new_entry, path_info, module, + ' for {values}'.format( + values=', '.join([ + '{primary_key}="{value}"'.format(primary_key=primary_key, value=value) + for primary_key, value in zip(primary_keys, key) + ]) + ) + ) + new_data.append(updated_entry) + # Add to modification list if there are changes + if modifications: + modifications['.id'] = old_entry['.id'] + modify_list.append((key, modifications)) + for new_entry in new_data_by_key.values(): + if path_info.fixed_entries: + module.fail_json(msg='Cannot add new entry {values} to this path'.format( + values=', '.join([ + '{primary_key}="{value}"'.format(primary_key=primary_key, value=new_entry[primary_key]) + for primary_key in primary_keys + ]), + )) + create_list.append(new_entry) + new_entry = new_entry.copy() + for key in list(new_entry): + if key.startswith('!'): + new_entry.pop(key) + new_data.append(new_entry) + + reorder_list = [] + if module.params['ensure_order']: + index_by_key = dict() + for index, entry in enumerate(new_data): + index_by_key[tuple(entry[primary_key] for primary_key in primary_keys)] = index + for index, source_entry in enumerate(data): + source_pks = tuple(source_entry[primary_key] for primary_key in primary_keys) + source_index = index_by_key.pop(source_pks) + if index == source_index: + continue + entry = new_data[index] + pks = tuple(entry[primary_key] for primary_key in primary_keys) + reorder_list.append((source_pks, index, pks)) + for k, v in index_by_key.items(): + if v >= index and v < source_index: + index_by_key[k] = v + 1 + new_data.insert(index, new_data.pop(source_index)) + + if not module.check_mode: + if remove_list: + try: + api_path.remove(*remove_list) + except LibRouterosError as e: + module.fail_json( + msg='Error while removing {remove_list}: {error}'.format( + remove_list=', '.join([ + '{identifier} (ID {id})'.format(identifier=format_pk(primary_keys, key), id=id) + for id, key in zip(remove_list, remove_keys) + ]), + error=to_native(e), + ) + ) + for key, modifications in modify_list: + try: + api_path.update(**modifications) + except LibRouterosError as e: + module.fail_json( + msg='Error while modifying for {identifier} (ID {id}): {error}'.format( + identifier=format_pk(primary_keys, key), + id=modifications['.id'], + error=to_native(e), + ) + ) + for entry in create_list: + try: + entry['.id'] = api_path.add(**entry) + # Store ID for primary keys + pks = tuple(entry[primary_key] for primary_key in primary_keys) + id_by_key[pks] = entry['.id'] + except LibRouterosError as e: + module.fail_json( + msg='Error while creating entry for {identifier}: {error}'.format( + identifier=format_pk(primary_keys, [entry[pk] for pk in primary_keys]), + error=to_native(e), + ) + ) + for element_pks, new_index, new_pks in reorder_list: + try: + element_id = id_by_key[element_pks] + new_id = id_by_key[new_pks] + for res in api_path('move', numbers=element_id, destination=new_id): + pass + except LibRouterosError as e: + module.fail_json( + msg='Error while moving entry ID {element_id} to position of ID {new_id}: {error}'.format( + element_id=element_id, + new_id=new_id, + error=to_native(e), + ) + ) + + # For sake of completeness, retrieve the full new data: + if modify_list or create_list or reorder_list: + new_data = list(api_path) + + # Remove 'irrelevant' data + for entry in old_data: + remove_irrelevant_data(entry, path_info) + for entry in new_data: + remove_irrelevant_data(entry, path_info) + + # Produce return value + more = {} + if module._diff: + more['diff'] = { + 'before': { + 'data': old_data, + }, + 'after': { + 'data': new_data, + }, + } + module.exit_json( + changed=bool(create_list or modify_list or remove_list or reorder_list), + old_data=old_data, + new_data=new_data, + **more + ) + + +def sync_single_value(module, api, path, path_info): + data = module.params['data'] + if len(data) != 1: + module.fail_json(msg='Data must be a list with exactly one element.') + new_entry = data[0] + polish_entry(new_entry, path_info, module, '') + + api_path = compose_api_path(api, path) + + old_data = list(api_path) + if len(old_data) != 1: + module.fail_json( + msg='Internal error: retrieving /{path} resulted in {count} elements. Expected exactly 1.'.format( + path=join_path(path), + count=len(old_data) + ) + ) + old_entry = old_data[0] + + # Determine modifications + modifications, updated_entry = find_modifications(old_entry, new_entry, path_info, module, '') + # Do modifications + if modifications: + if not module.check_mode: + # Actually do modification + try: + api_path.update(**modifications) + except LibRouterosError as e: + module.fail_json(msg='Error while modifying: {error}'.format(error=to_native(e))) + # Retrieve latest version + new_data = list(api_path) + if len(new_data) == 1: + updated_entry = new_data[0] + + # Remove 'irrelevant' data + remove_irrelevant_data(old_entry, path_info) + remove_irrelevant_data(updated_entry, path_info) + + # Produce return value + more = {} + if module._diff: + more['diff'] = { + 'before': old_entry, + 'after': updated_entry, + } + module.exit_json( + changed=bool(modifications), + old_data=[old_entry], + new_data=[updated_entry], + **more + ) + + +def get_backend(path_info): + if path_info is None: + return None + if not path_info.fully_understood: + return None + + if path_info.primary_keys: + return sync_with_primary_keys + + if path_info.single_value: + return sync_single_value + + if not path_info.has_identifier: + return sync_list + + return None + + +def main(): + path_choices = sorted([join_path(path) for path, path_info in PATHS.items() if get_backend(path_info) is not None]) + module_args = dict( + path=dict(type='str', required=True, choices=path_choices), + data=dict(type='list', elements='dict', required=True), + handle_absent_entries=dict(type='str', choices=['ignore', 'remove'], default='ignore'), + handle_entries_content=dict(type='str', choices=['ignore', 'remove', 'remove_as_much_as_possible'], default='ignore'), + ensure_order=dict(type='bool', default=False), + ) + module_args.update(api_argument_spec()) + + module = AnsibleModule( + argument_spec=module_args, + supports_check_mode=True, + ) + if module.params['ensure_order'] and module.params['handle_absent_entries'] == 'ignore': + module.fail_json(msg='ensure_order=true requires handle_absent_entries=remove') + + if not HAS_ORDEREDDICT: + # This should never happen for Python 2.7+ + module.fail_json(msg=missing_required_lib('ordereddict')) + + check_has_library(module) + api = create_api(module) + + path = split_path(module.params['path']) + path_info = PATHS.get(tuple(path)) + backend = get_backend(path_info) + if path_info is None or backend is None: + module.fail_json(msg='Path /{path} is not yet supported'.format(path='/'.join(path))) + + backend(module, api, path, path_info) + + +if __name__ == '__main__': + main() diff --git a/tests/unit/plugins/modules/test_api_info.py b/tests/unit/plugins/modules/test_api_info.py new file mode 100644 index 000000000..72bb1bb26 --- /dev/null +++ b/tests/unit/plugins/modules/test_api_info.py @@ -0,0 +1,364 @@ +# 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 . + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import json +import pytest + +from ansible_collections.community.routeros.tests.unit.compat.mock import patch, MagicMock +from ansible_collections.community.routeros.tests.unit.plugins.modules.fake_api import FakeLibRouterosError, Key, fake_ros_api +from ansible_collections.community.routeros.tests.unit.plugins.modules.utils import set_module_args, AnsibleExitJson, AnsibleFailJson, ModuleTestCase +from ansible_collections.community.routeros.plugins.modules import api_info + + +class TestRouterosApiInfoModule(ModuleTestCase): + + def setUp(self): + super(TestRouterosApiInfoModule, self).setUp() + self.module = api_info + self.module.LibRouterosError = FakeLibRouterosError + self.module.connect = MagicMock(new=fake_ros_api) + self.module.check_has_library = MagicMock() + self.patch_create_api = patch('ansible_collections.community.routeros.plugins.modules.api_info.create_api', MagicMock(new=fake_ros_api)) + self.patch_create_api.start() + self.module.Key = MagicMock(new=Key) + self.config_module_args = { + 'username': 'admin', + 'password': 'pаss', + 'hostname': '127.0.0.1', + } + + def tearDown(self): + self.patch_create_api.stop() + + def test_module_fail_when_required_args_missing(self): + with self.assertRaises(AnsibleFailJson) as exc: + set_module_args({}) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['failed'], True) + + def test_invalid_path(self): + with self.assertRaises(AnsibleFailJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'something invalid' + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['failed'], True) + self.assertEqual(result['msg'].startswith('value of path must be one of: '), True) + + @patch('ansible_collections.community.routeros.plugins.modules.api_info.compose_api_path') + def test_empty_result(self, mock_compose_api_path): + mock_compose_api_path.return_value = [] + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static' + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], False) + self.assertEqual(result['result'], []) + + @patch('ansible_collections.community.routeros.plugins.modules.api_info.compose_api_path') + def test_regular_result(self, mock_compose_api_path): + mock_compose_api_path.return_value = [ + { + 'called-format': 'mac:ssid', + 'interim-update': 'enabled', + 'mac-caching': 'disabled', + 'mac-format': 'XX:XX:XX:XX:XX:XX', + 'mac-mode': 'as-username', + 'foo': 'bar', + '.id': '*1', + }, + ] + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'caps-man aaa', + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], False) + self.assertEqual(result['result'], [{ + 'interim-update': 'enabled', + '.id': '*1', + }]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_info.compose_api_path') + def test_result_with_defaults(self, mock_compose_api_path): + mock_compose_api_path.return_value = [ + { + 'called-format': 'mac:ssid', + 'interim-update': 'enabled', + 'mac-caching': 'disabled', + 'mac-format': 'XX:XX:XX:XX:XX:XX', + 'mac-mode': 'as-username', + 'foo': 'bar', + '.id': '*1', + }, + ] + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'caps-man aaa', + 'hide_defaults': False, + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], False) + self.assertEqual(result['result'], [{ + 'called-format': 'mac:ssid', + 'interim-update': 'enabled', + 'mac-caching': 'disabled', + 'mac-format': 'XX:XX:XX:XX:XX:XX', + 'mac-mode': 'as-username', + '.id': '*1', + }]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_info.compose_api_path') + def test_full_result(self, mock_compose_api_path): + mock_compose_api_path.return_value = [ + { + 'called-format': 'mac:ssid', + 'interim-update': 'enabled', + 'mac-caching': 'disabled', + 'mac-format': 'XX:XX:XX:XX:XX:XX', + 'mac-mode': 'as-username', + 'foo': 'bar', + '.id': '*1', + }, + ] + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'caps-man aaa', + 'unfiltered': True, + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], False) + self.assertEqual(result['result'], [{ + 'interim-update': 'enabled', + 'foo': 'bar', + '.id': '*1', + }]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_info.compose_api_path') + def test_disabled_exclamation(self, mock_compose_api_path): + mock_compose_api_path.return_value = [ + { + 'chain': 'input', + 'in-interface-list': 'LAN', + '.id': '*1', + }, + ] + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip firewall filter', + 'handle_disabled': 'exclamation', + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], False) + self.assertEqual(result['result'], [{ + 'chain': 'input', + 'in-interface-list': 'LAN', + '!action': None, + '!comment': None, + '!connection-bytes': None, + '!connection-limit': None, + '!connection-mark': None, + '!connection-nat-state': None, + '!connection-rate': None, + '!connection-state': None, + '!connection-type': None, + '!content': None, + '!disabled': None, + '!dscp': None, + '!dst-address': None, + '!dst-address-list': None, + '!dst-address-type': None, + '!dst-limit': None, + '!dst-port': None, + '!fragment': None, + '!hotspot': None, + '!icmp-options': None, + '!in-bridge-port': None, + '!in-bridge-port-list': None, + '!in-interface': None, + '!ingress-priority': None, + '!ipsec-policy': None, + '!ipv4-options': None, + '!layer7-protocol': None, + '!limit': None, + '!log': None, + '!log-prefix': None, + '!nth': None, + '!out-bridge-port': None, + '!out-bridge-port-list': None, + '!out-interface': None, + '!out-interface-list': None, + '!p2p': None, + '!packet-mark': None, + '!packet-size': None, + '!per-connection-classifier': None, + '!port': None, + '!priority': None, + '!protocol': None, + '!psd': None, + '!random': None, + '!routing-mark': None, + '!routing-table': None, + '!src-address': None, + '!src-address-list': None, + '!src-address-type': None, + '!src-mac-address': None, + '!src-port': None, + '!tcp-flags': None, + '!tcp-mss': None, + '!time': None, + '!tls-host': None, + '!ttl': None, + '.id': '*1', + }]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_info.compose_api_path') + def test_disabled_null_value(self, mock_compose_api_path): + mock_compose_api_path.return_value = [ + { + 'chain': 'input', + 'in-interface-list': 'LAN', + '.id': '*1', + }, + ] + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip firewall filter', + 'handle_disabled': 'null-value', + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], False) + self.assertEqual(result['result'], [{ + 'chain': 'input', + 'in-interface-list': 'LAN', + 'action': None, + 'comment': None, + 'connection-bytes': None, + 'connection-limit': None, + 'connection-mark': None, + 'connection-nat-state': None, + 'connection-rate': None, + 'connection-state': None, + 'connection-type': None, + 'content': None, + 'disabled': None, + 'dscp': None, + 'dst-address': None, + 'dst-address-list': None, + 'dst-address-type': None, + 'dst-limit': None, + 'dst-port': None, + 'fragment': None, + 'hotspot': None, + 'icmp-options': None, + 'in-bridge-port': None, + 'in-bridge-port-list': None, + 'in-interface': None, + 'ingress-priority': None, + 'ipsec-policy': None, + 'ipv4-options': None, + 'layer7-protocol': None, + 'limit': None, + 'log': None, + 'log-prefix': None, + 'nth': None, + 'out-bridge-port': None, + 'out-bridge-port-list': None, + 'out-interface': None, + 'out-interface-list': None, + 'p2p': None, + 'packet-mark': None, + 'packet-size': None, + 'per-connection-classifier': None, + 'port': None, + 'priority': None, + 'protocol': None, + 'psd': None, + 'random': None, + 'routing-mark': None, + 'routing-table': None, + 'src-address': None, + 'src-address-list': None, + 'src-address-type': None, + 'src-mac-address': None, + 'src-port': None, + 'tcp-flags': None, + 'tcp-mss': None, + 'time': None, + 'tls-host': None, + 'ttl': None, + '.id': '*1', + }]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_info.compose_api_path') + def test_disabled_omit(self, mock_compose_api_path): + mock_compose_api_path.return_value = [ + { + 'chain': 'input', + 'in-interface-list': 'LAN', + '.id': '*1', + }, + ] + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip firewall filter', + 'handle_disabled': 'omit', + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], False) + self.assertEqual(result['result'], [{ + 'chain': 'input', + 'in-interface-list': 'LAN', + '.id': '*1', + }]) diff --git a/tests/unit/plugins/modules/test_api_modify.py b/tests/unit/plugins/modules/test_api_modify.py new file mode 100644 index 000000000..a8a714156 --- /dev/null +++ b/tests/unit/plugins/modules/test_api_modify.py @@ -0,0 +1,1544 @@ +# 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 . + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import json +import pytest + +from ansible_collections.community.routeros.tests.unit.compat.mock import patch, MagicMock +from ansible_collections.community.routeros.tests.unit.plugins.modules.fake_api import ( + FakeLibRouterosError, fake_ros_api, massage_expected_result_data, create_fake_path, +) +from ansible_collections.community.routeros.tests.unit.plugins.modules.utils import set_module_args, AnsibleExitJson, AnsibleFailJson, ModuleTestCase +from ansible_collections.community.routeros.plugins.module_utils._api_data import PATHS +from ansible_collections.community.routeros.plugins.modules import api_modify + + +START_IP_DNS_STATIC = [ + { + '.id': '*1', + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + 'dynamic': False, + }, + { + '.id': '*A', + 'name': 'router', + 'text': 'Router Text Entry', + 'dynamic': False, + }, + { + '.id': '*7', + 'comment': '', + 'name': 'foo', + 'address': '192.168.88.2', + 'dynamic': False, + }, +] + +START_IP_DNS_STATIC_OLD_DATA = massage_expected_result_data(START_IP_DNS_STATIC, ('ip', 'dns', 'static')) + +START_IP_SETTINGS = [ + { + 'accept-redirects': True, + 'accept-source-route': False, + 'allow-fast-path': True, + 'arp-timeout': '30s', + 'icmp-rate-limit': 20, + 'icmp-rate-mask': '0x1818', + 'ip-forward': True, + 'max-neighbor-entries': 8192, + 'route-cache': True, + 'rp-filter': False, + 'secure-redirects': True, + 'send-redirects': True, + 'tcp-syncookies': False, + }, +] + +START_IP_SETTINGS_OLD_DATA = massage_expected_result_data(START_IP_SETTINGS, ('ip', 'settings')) + +START_IP_ADDRESS = [ + { + '.id': '*1', + 'address': '192.168.88.0/24', + 'interface': 'bridge', + 'disabled': False, + }, + { + '.id': '*3', + 'address': '192.168.1.0/24', + 'interface': 'LAN', + 'disabled': False, + }, + { + '.id': '*F', + 'address': '10.0.0.0/16', + 'interface': 'WAN', + 'disabled': True, + }, +] + +START_IP_ADDRESS_OLD_DATA = massage_expected_result_data(START_IP_ADDRESS, ('ip', 'address')) + + +class TestRouterosApiModifyModule(ModuleTestCase): + + def setUp(self): + super(TestRouterosApiModifyModule, self).setUp() + self.module = api_modify + self.module.LibRouterosError = FakeLibRouterosError + self.module.connect = MagicMock(new=fake_ros_api) + self.module.check_has_library = MagicMock() + self.patch_create_api = patch( + 'ansible_collections.community.routeros.plugins.modules.api_modify.create_api', + MagicMock(new=fake_ros_api)) + self.patch_create_api.start() + self.config_module_args = { + 'username': 'admin', + 'password': 'pаss', + 'hostname': '127.0.0.1', + } + + def tearDown(self): + self.patch_create_api.stop() + + def test_module_fail_when_required_args_missing(self): + with self.assertRaises(AnsibleFailJson) as exc: + set_module_args({}) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['failed'], True) + + def test_invalid_path(self): + with self.assertRaises(AnsibleFailJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'something invalid', + 'data': [], + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['failed'], True) + self.assertEqual(result['msg'].startswith('value of path must be one of: '), True) + + def test_invalid_option(self): + with self.assertRaises(AnsibleFailJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [{ + 'name': 'baz', + 'foo': 'bar', + }], + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['failed'], True) + self.assertEqual(result['msg'], 'Unknown key "foo" at index 1.') + + def test_invalid_disabled_and_enabled_option(self): + with self.assertRaises(AnsibleFailJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [{ + 'name': 'baz', + 'comment': 'foo', + '!comment': None, + }], + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['failed'], True) + self.assertEqual(result['msg'], 'Not both "comment" and "!comment" must appear at index 1.') + + def test_invalid_disabled_option(self): + with self.assertRaises(AnsibleFailJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [{ + 'name': 'foo', + '!disabled': None, + }], + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['failed'], True) + self.assertEqual(result['msg'], 'Key "!disabled" must not be disabled (leading "!") at index 1.') + + def test_invalid_disabled_option_value(self): + with self.assertRaises(AnsibleFailJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [{ + 'name': 'baz', + '!comment': 'foo', + }], + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['failed'], True) + self.assertEqual(result['msg'], 'Disabled key "!comment" must not have a value at index 1.') + + def test_invalid_non_disabled_option_value(self): + with self.assertRaises(AnsibleFailJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [{ + 'name': None, + }], + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['failed'], True) + self.assertEqual(result['msg'], 'Key "name" must not be disabled (value null/~/None) at index 1.') + + def test_invalid_required_missing(self): + with self.assertRaises(AnsibleFailJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [{ + 'address': '1.2.3.4', + }], + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['failed'], True) + self.assertEqual(result['msg'], 'Every element in data must contain "name". For example, the element at index #1 does not provide it.') + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC, read_only=True)) + def test_sync_list_idempotent(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + '.id': 'bam', # this should be ignored + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + }, + { + 'name': 'foo', + 'address': '192.168.88.2', + }, + { + 'comment': None, + 'name': 'router', + 'text': 'Router Text Entry', + }, + ], + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], False) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], START_IP_DNS_STATIC_OLD_DATA) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC, read_only=True)) + def test_sync_list_idempotent_2(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + }, + { + 'name': 'foo', + 'comment': '', + 'address': '192.168.88.2', + }, + { + 'name': 'router', + '!comment': None, + 'text': 'Router Text Entry', + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], False) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], START_IP_DNS_STATIC_OLD_DATA) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC, read_only=True)) + def test_sync_list_idempotent_3(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + }, + ], + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], False) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], START_IP_DNS_STATIC_OLD_DATA) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC)) + def test_sync_list_add(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + }, + { + 'name': 'router', + 'text': 'Router Text Entry', + }, + { + 'name': 'router', + 'text': 'Router Text Entry 2', + }, + { + 'name': 'foo', + 'address': '192.168.88.2', + }, + ], + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*1', + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*A', + 'name': 'router', + 'text': 'Router Text Entry', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*7', + 'name': 'foo', + 'address': '192.168.88.2', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*NEW1', + 'name': 'router', + 'text': 'Router Text Entry 2', + 'ttl': '1d', + 'disabled': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC)) + def test_sync_list_modify_1(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + }, + { + 'name': 'router', + 'text': 'Router Text Entry 2', + }, + { + 'name': 'foo', + 'address': '192.168.88.2', + }, + ], + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*1', + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*A', + 'name': 'router', + 'text': 'Router Text Entry', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*7', + 'name': 'foo', + 'address': '192.168.88.2', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*NEW1', + 'name': 'router', + 'text': 'Router Text Entry 2', + 'ttl': '1d', + 'disabled': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC, read_only=True)) + def test_sync_list_modify_1_check(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + }, + { + 'name': 'router', + 'text': 'Router Text Entry 2', + }, + { + 'name': 'foo', + 'address': '192.168.88.2', + }, + ], + '_ansible_check_mode': True, + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*1', + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*A', + 'name': 'router', + 'text': 'Router Text Entry', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*7', + 'name': 'foo', + 'address': '192.168.88.2', + 'ttl': '1d', + 'disabled': False, + }, + { + 'name': 'router', + 'text': 'Router Text Entry 2', + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC)) + def test_sync_list_modify_2(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + 'comment': '', + 'name': 'router', + 'address': '192.168.88.1', + }, + { + 'name': 'router', + 'text': 'Router Text Entry 2', + }, + { + 'name': 'foo', + 'address': '192.168.88.2', + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*1', + 'name': 'router', + 'address': '192.168.88.1', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*A', + 'name': 'router', + 'text': 'Router Text Entry 2', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*7', + 'name': 'foo', + 'address': '192.168.88.2', + 'ttl': '1d', + 'disabled': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC, read_only=True)) + def test_sync_list_modify_2_check(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + 'comment': '', + 'name': 'router', + 'address': '192.168.88.1', + }, + { + 'name': 'router', + 'text': 'Router Text Entry 2', + }, + { + 'name': 'foo', + 'address': '192.168.88.2', + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + '_ansible_check_mode': True, + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*1', + 'name': 'router', + 'address': '192.168.88.1', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*A', + 'name': 'router', + 'text': 'Router Text Entry 2', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*7', + 'name': 'foo', + 'address': '192.168.88.2', + 'ttl': '1d', + 'disabled': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC)) + def test_sync_list_modify_3(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + '!comment': None, + 'name': 'router', + 'address': '192.168.88.1', + }, + { + 'name': 'router', + 'cname': 'router.com.', + }, + { + 'name': 'foo', + 'address': '192.168.88.2', + 'ttl': '1d', + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*1', + 'name': 'router', + 'address': '192.168.88.1', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*7', + 'name': 'foo', + 'address': '192.168.88.2', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*NEW1', + 'name': 'router', + 'cname': 'router.com.', + 'ttl': '1d', + 'disabled': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC, read_only=True)) + def test_sync_list_modify_3_check(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + '!comment': None, + 'name': 'router', + 'address': '192.168.88.1', + }, + { + 'name': 'router', + 'cname': 'router.com.', + }, + { + 'name': 'foo', + 'address': '192.168.88.2', + 'ttl': '1d', + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + '_ansible_check_mode': True, + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*1', + 'name': 'router', + 'address': '192.168.88.1', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*7', + 'name': 'foo', + 'address': '192.168.88.2', + 'ttl': '1d', + 'disabled': False, + }, + { + 'name': 'router', + 'cname': 'router.com.', + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC)) + def test_sync_list_modify_4(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + 'name': 'router', + 'address': '192.168.88.1', + }, + { + 'name': 'router', + 'comment': 'defconf', + 'text': 'Router Text Entry 2', + }, + { + 'name': 'foo', + 'address': '192.168.88.2', + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*1', + 'name': 'router', + 'address': '192.168.88.1', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*A', + 'comment': 'defconf', + 'name': 'router', + 'text': 'Router Text Entry 2', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*7', + 'name': 'foo', + 'address': '192.168.88.2', + 'ttl': '1d', + 'disabled': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC, read_only=True)) + def test_sync_list_modify_4_check(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + 'name': 'router', + 'address': '192.168.88.1', + }, + { + 'name': 'router', + 'comment': 'defconf', + 'text': 'Router Text Entry 2', + }, + { + 'name': 'foo', + 'address': '192.168.88.2', + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + '_ansible_check_mode': True, + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*1', + 'name': 'router', + 'address': '192.168.88.1', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*A', + 'comment': 'defconf', + 'name': 'router', + 'text': 'Router Text Entry 2', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*7', + 'name': 'foo', + 'address': '192.168.88.2', + 'ttl': '1d', + 'disabled': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC)) + def test_sync_list_delete(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*1', + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + 'ttl': '1d', + 'disabled': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC, read_only=True)) + def test_sync_list_delete_check(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + '_ansible_check_mode': True, + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*1', + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + 'ttl': '1d', + 'disabled': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC)) + def test_sync_list_reorder(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + 'name': 'foo', + 'address': '192.168.88.2', + }, + { + 'name': 'foo', + 'text': 'bar', + }, + { + 'name': 'router', + 'text': 'Router Text Entry', + }, + { + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + 'ensure_order': True, + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*7', + 'name': 'foo', + 'address': '192.168.88.2', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*NEW1', + 'name': 'foo', + 'text': 'bar', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*A', + 'name': 'router', + 'text': 'Router Text Entry', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*1', + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + 'ttl': '1d', + 'disabled': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'dns', 'static'), START_IP_DNS_STATIC, read_only=True)) + def test_sync_list_reorder_check(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip dns static', + 'data': [ + { + 'name': 'foo', + 'address': '192.168.88.2', + }, + { + 'name': 'foo', + 'text': 'bar', + }, + { + 'name': 'router', + 'text': 'Router Text Entry', + }, + { + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + 'ensure_order': True, + '_ansible_check_mode': True, + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_DNS_STATIC_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*7', + 'name': 'foo', + 'address': '192.168.88.2', + 'ttl': '1d', + 'disabled': False, + }, + { + 'name': 'foo', + 'text': 'bar', + }, + { + '.id': '*A', + 'name': 'router', + 'text': 'Router Text Entry', + 'ttl': '1d', + 'disabled': False, + }, + { + '.id': '*1', + 'comment': 'defconf', + 'name': 'router', + 'address': '192.168.88.1', + 'ttl': '1d', + 'disabled': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'settings'), START_IP_SETTINGS, read_only=True)) + def test_sync_value_idempotent(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip settings', + 'data': [ + { + 'arp-timeout': '30s', + 'icmp-rate-limit': 20, + 'icmp-rate-mask': '0x1818', + 'ip-forward': True, + 'max-neighbor-entries': 8192, + }, + ], + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], False) + self.assertEqual(result['old_data'], START_IP_SETTINGS_OLD_DATA) + self.assertEqual(result['new_data'], START_IP_SETTINGS_OLD_DATA) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'settings'), START_IP_SETTINGS, read_only=True)) + def test_sync_value_idempotent_2(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip settings', + 'data': [ + { + 'accept-redirects': True, + 'icmp-rate-limit': 20, + }, + ], + 'handle_entries_content': 'remove', + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], False) + self.assertEqual(result['old_data'], START_IP_SETTINGS_OLD_DATA) + self.assertEqual(result['new_data'], START_IP_SETTINGS_OLD_DATA) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'settings'), START_IP_SETTINGS)) + def test_sync_value_modify(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip settings', + 'data': [ + { + 'accept-redirects': True, + 'accept-source-route': True, + 'max-neighbor-entries': 4096, + }, + ], + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_SETTINGS_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + 'accept-redirects': True, + 'accept-source-route': True, + 'allow-fast-path': True, + 'arp-timeout': '30s', + 'icmp-rate-limit': 20, + 'icmp-rate-mask': '0x1818', + 'ip-forward': True, + 'max-neighbor-entries': 4096, + 'route-cache': True, + 'rp-filter': False, + 'secure-redirects': True, + 'send-redirects': True, + 'tcp-syncookies': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'settings'), START_IP_SETTINGS, read_only=True)) + def test_sync_value_modify_check(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip settings', + 'data': [ + { + 'accept-redirects': True, + 'accept-source-route': True, + 'max-neighbor-entries': 4096, + }, + ], + '_ansible_check_mode': True, + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_SETTINGS_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + 'accept-redirects': True, + 'accept-source-route': True, + 'allow-fast-path': True, + 'arp-timeout': '30s', + 'icmp-rate-limit': 20, + 'icmp-rate-mask': '0x1818', + 'ip-forward': True, + 'max-neighbor-entries': 4096, + 'route-cache': True, + 'rp-filter': False, + 'secure-redirects': True, + 'send-redirects': True, + 'tcp-syncookies': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'settings'), START_IP_SETTINGS)) + def test_sync_value_modify_2(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip settings', + 'data': [ + { + 'accept-redirects': True, + 'accept-source-route': True, + 'max-neighbor-entries': 4096, + }, + ], + 'handle_entries_content': 'remove', + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_SETTINGS_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + 'accept-redirects': True, + 'accept-source-route': True, + 'allow-fast-path': True, + 'arp-timeout': '30s', + 'icmp-rate-limit': 10, + 'icmp-rate-mask': '0x1818', + 'ip-forward': True, + 'max-neighbor-entries': 4096, + 'route-cache': True, + 'rp-filter': False, + 'secure-redirects': True, + 'send-redirects': True, + 'tcp-syncookies': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'settings'), START_IP_SETTINGS, read_only=True)) + def test_sync_value_modify_2_check(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip settings', + 'data': [ + { + 'accept-redirects': True, + 'accept-source-route': True, + 'max-neighbor-entries': 4096, + }, + ], + 'handle_entries_content': 'remove', + '_ansible_check_mode': True, + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_SETTINGS_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + 'accept-redirects': True, + 'accept-source-route': True, + 'allow-fast-path': True, + 'arp-timeout': '30s', + 'icmp-rate-limit': 10, + 'icmp-rate-mask': '0x1818', + 'ip-forward': True, + 'max-neighbor-entries': 4096, + 'route-cache': True, + 'rp-filter': False, + 'secure-redirects': True, + 'send-redirects': True, + 'tcp-syncookies': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'address'), START_IP_ADDRESS, read_only=True)) + def test_sync_primary_key_idempotent(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip address', + 'data': [ + { + 'address': '192.168.1.0/24', + 'interface': 'LAN', + 'comment': '', + }, + { + 'address': '192.168.88.0/24', + 'interface': 'bridge', + '!comment': None, + }, + ], + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], False) + self.assertEqual(result['old_data'], START_IP_ADDRESS_OLD_DATA) + self.assertEqual(result['new_data'], START_IP_ADDRESS_OLD_DATA) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'address'), START_IP_ADDRESS, read_only=True)) + def test_sync_primary_key_idempotent_2(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip address', + 'data': [ + { + 'address': '192.168.88.0/24', + 'interface': 'bridge', + }, + { + 'address': '10.0.0.0/16', + 'interface': 'WAN', + 'disabled': True, + '!comment': '', + }, + { + 'address': '192.168.1.0/24', + 'interface': 'LAN', + 'disabled': False, + 'comment': None, + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], False) + self.assertEqual(result['old_data'], START_IP_ADDRESS_OLD_DATA) + self.assertEqual(result['new_data'], START_IP_ADDRESS_OLD_DATA) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'address'), START_IP_ADDRESS)) + def test_sync_primary_key_cru(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip address', + 'data': [ + { + 'address': '10.10.0.0/16', + 'interface': 'WIFI', + }, + { + 'address': '192.168.1.0/24', + 'interface': 'LAN', + 'disabled': True, + }, + { + 'address': '192.168.88.0/24', + 'interface': 'bridge', + 'comment': 'foo', + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_ADDRESS_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*1', + 'comment': 'foo', + 'address': '192.168.88.0/24', + 'interface': 'bridge', + 'disabled': False, + }, + { + '.id': '*3', + 'address': '192.168.1.0/24', + 'interface': 'LAN', + 'disabled': True, + }, + { + '.id': '*NEW1', + 'address': '10.10.0.0/16', + 'interface': 'WIFI', + 'disabled': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'address'), START_IP_ADDRESS, read_only=True)) + def test_sync_primary_key_cru_check(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip address', + 'data': [ + { + 'address': '10.10.0.0/16', + 'interface': 'WIFI', + }, + { + 'address': '192.168.1.0/24', + 'interface': 'LAN', + 'disabled': True, + }, + { + 'address': '192.168.88.0/24', + 'interface': 'bridge', + 'comment': 'foo', + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + '_ansible_check_mode': True, + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_ADDRESS_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*1', + 'comment': 'foo', + 'address': '192.168.88.0/24', + 'interface': 'bridge', + 'disabled': False, + }, + { + '.id': '*3', + 'address': '192.168.1.0/24', + 'interface': 'LAN', + 'disabled': True, + }, + { + 'address': '10.10.0.0/16', + 'interface': 'WIFI', + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'address'), START_IP_ADDRESS)) + def test_sync_primary_key_cru_reorder(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip address', + 'data': [ + { + 'address': '10.10.0.0/16', + 'interface': 'WIFI', + }, + { + 'address': '192.168.1.0/24', + 'interface': 'LAN', + 'disabled': True, + }, + { + 'address': '192.168.88.0/24', + 'interface': 'bridge', + 'comment': 'foo', + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + 'ensure_order': True, + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_ADDRESS_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + '.id': '*NEW1', + 'address': '10.10.0.0/16', + 'interface': 'WIFI', + 'disabled': False, + }, + { + '.id': '*3', + 'address': '192.168.1.0/24', + 'interface': 'LAN', + 'disabled': True, + }, + { + '.id': '*1', + 'comment': 'foo', + 'address': '192.168.88.0/24', + 'interface': 'bridge', + 'disabled': False, + }, + ]) + + @patch('ansible_collections.community.routeros.plugins.modules.api_modify.compose_api_path', + new=create_fake_path(('ip', 'address'), START_IP_ADDRESS, read_only=True)) + def test_sync_primary_key_cru_reorder_check(self): + with self.assertRaises(AnsibleExitJson) as exc: + args = self.config_module_args.copy() + args.update({ + 'path': 'ip address', + 'data': [ + { + 'address': '10.10.0.0/16', + 'interface': 'WIFI', + }, + { + 'address': '192.168.1.0/24', + 'interface': 'LAN', + 'disabled': True, + }, + { + 'address': '192.168.88.0/24', + 'interface': 'bridge', + 'comment': 'foo', + }, + ], + 'handle_absent_entries': 'remove', + 'handle_entries_content': 'remove', + 'ensure_order': True, + '_ansible_check_mode': True, + }) + set_module_args(args) + self.module.main() + + result = exc.exception.args[0] + self.assertEqual(result['changed'], True) + self.assertEqual(result['old_data'], START_IP_ADDRESS_OLD_DATA) + self.assertEqual(result['new_data'], [ + { + 'address': '10.10.0.0/16', + 'interface': 'WIFI', + }, + { + '.id': '*3', + 'address': '192.168.1.0/24', + 'interface': 'LAN', + 'disabled': True, + }, + { + '.id': '*1', + 'comment': 'foo', + 'address': '192.168.88.0/24', + 'interface': 'bridge', + 'disabled': False, + }, + ])