From 703daa9500c26ad0a3e8e524fadc2343863ac1ca Mon Sep 17 00:00:00 2001 From: Felix Fontein Date: Thu, 23 Apr 2020 14:50:44 +0200 Subject: [PATCH] Add small unit test framework for testing fetch_url() based modules (#171) * Convert integrated simple fetch_url() checking framework from hetzner_firewall tests to proper framework which can also be used by other modules. * Linting. * One more. * Use community.internal_test_tools collection. --- tests/requirements.yml | 1 + .../net_tools/test_hetzner_firewall.py | 2482 ++++++++--------- .../net_tools/test_hetzner_firewall_info.py | 423 +-- tests/utils/shippable/shippable.sh | 1 + 4 files changed, 1349 insertions(+), 1558 deletions(-) diff --git a/tests/requirements.yml b/tests/requirements.yml index 79e0f11761..21e2a9ab1a 100644 --- a/tests/requirements.yml +++ b/tests/requirements.yml @@ -7,6 +7,7 @@ unit_tests_dependencies: - ansible.netcommon - ansible.posix - cisco.intersight +- community.internal_test_tools - community.kubernetes - google.cloud - ovirt.ovirt diff --git a/tests/unit/plugins/modules/net_tools/test_hetzner_firewall.py b/tests/unit/plugins/modules/net_tools/test_hetzner_firewall.py index d1f1984a89..bc87a51ce9 100644 --- a/tests/unit/plugins/modules/net_tools/test_hetzner_firewall.py +++ b/tests/unit/plugins/modules/net_tools/test_hetzner_firewall.py @@ -7,1172 +7,15 @@ __metaclass__ = type import pytest +from ansible_collections.community.internal_test_tools.tests.unit.utils.fetch_url_module_framework import ( + FetchUrlCall, + BaseTestModule, +) + from ansible_collections.community.general.plugins.module_utils.hetzner import BASE_URL from ansible_collections.community.general.plugins.modules.net_tools import hetzner_firewall -# ########################################################## -# ## General test framework - -import json - -from mock import MagicMock -from ansible_collections.community.general.tests.unit.plugins.modules.utils import set_module_args -from ansible.module_utils.six.moves.urllib.parse import parse_qs - - -class FetchUrlCall: - def __init__(self, method, status): - assert method == method.upper(), \ - 'HTTP method names are case-sensitive and should be upper-case (RFCs 7230 and 7231)' - self.method = method - self.status = status - self.body = None - self.headers = {} - self.error_data = {} - self.expected_url = None - self.expected_headers = {} - self.form_parse = False - self.form_present = set() - self.form_values = {} - self.form_values_one = {} - - def result(self, body): - self.body = body - assert self.error_data.get('body') is None, 'Error body must not be given' - return self - - def result_str(self, str_body): - return self.result(str_body.encode('utf-8')) - - def result_json(self, json_body): - return self.result(json.dumps(json_body).encode('utf-8')) - - def result_error(self, msg, body=None): - self.error_data['msg'] = msg - if body is not None: - self.error_data['body'] = body - assert self.body is None, 'Result must not be given if error body is provided' - return self - - def expect_url(self, url): - self.expected_url = url - return self - - def return_header(self, name, value): - assert value is not None - self.headers[name] = value - return self - - def expect_header(self, name, value): - self.expected_headers[name] = value - return self - - def expect_header_unset(self, name): - self.expected_headers[name] = None - return self - - def expect_form_present(self, key): - self.form_parse = True - self.form_present.append(key) - return self - - def expect_form_value(self, key, value): - self.form_parse = True - self.form_values[key] = [value] - return self - - def expect_form_value_absent(self, key): - self.form_parse = True - self.form_values[key] = [] - return self - - def expect_form_value_one_of(self, key, value): - self.form_parse = True - if key not in self.form_values_subset: - self.form_values_subset[key] = set() - self.form_values_subset[key].add(value) - return self - - -class FetchUrlProxy: - def __init__(self, calls): - self.calls = calls - self.index = 0 - - def _validate_form(self, call, data): - form = {} - if data is not None: - form = parse_qs(data, keep_blank_values=True) - for k in call.form_present: - assert k in form - for k, v in call.form_values.items(): - if len(v) == 0: - assert k not in form - else: - assert form[k] == v - for k, v in call.form_values_one.items(): - assert v <= set(form[k]) - - def _validate_headers(self, call, headers): - given_headers = {} - if headers is not None: - for k, v in headers.items(): - given_headers[k.lower()] = v - for k, v in call.expected_headers: - if v is None: - assert k.lower() not in given_headers, \ - 'Header "{0}" specified for fetch_url call, but should not be'.format(k) - else: - assert given_headers.get(k.lower()) == v, \ - 'Header "{0}" specified for fetch_url call, but with wrong value'.format(k) - - def __call__(self, module, url, data=None, headers=None, method=None, - use_proxy=True, force=False, last_mod_time=None, timeout=10, - use_gssapi=False, unix_socket=None, ca_path=None, cookies=None): - assert self.index < len(self.calls), 'Got more fetch_url calls than expected' - call = self.calls[self.index] - self.index += 1 - - # Validate call - assert method == call.method - if call.expected_url is not None: - assert url == call.expected_url, \ - 'Exepected URL does not match for fetch_url call' - if call.expected_headers: - self._validate_headers(call, headers) - if call.form_parse: - self._validate_form(call, data) - - # Compose result - info = dict(status=call.status) - for k, v in call.headers.items(): - info[k.lower()] = v - info.update(call.error_data) - res = object() - if call.body is not None: - res = MagicMock() - res.read = MagicMock(return_value=call.body) - return (res, info) - - def assert_is_done(self): - assert self.index == len(self.calls), 'Got less fetch_url calls than expected' - - -class ModuleExitException(Exception): - def __init__(self, kwargs): - self.kwargs = kwargs - - -class ModuleFailException(Exception): - def __init__(self, kwargs): - self.kwargs = kwargs - - -def run_module(mocker, module, arguments, fetch_url): - def exit_json(module, **kwargs): - module._return_formatted(kwargs) - raise ModuleExitException(kwargs) - - def fail_json(module, **kwargs): - module._return_formatted(kwargs) - raise ModuleFailException(kwargs) - - mocker.patch('ansible_collections.community.general.plugins.module_utils.hetzner.fetch_url', fetch_url) - mocker.patch('ansible_collections.community.general.plugins.module_utils.hetzner.time.sleep', lambda duration: None) - mocker.patch('ansible_collections.community.general.plugins.modules.net_tools.hetzner_firewall.AnsibleModule.exit_json', exit_json) - mocker.patch('ansible_collections.community.general.plugins.modules.net_tools.hetzner_firewall.AnsibleModule.fail_json', fail_json) - set_module_args(arguments) - module.main() - - -def run_module_success(mocker, module, arguments, fetch_url_calls): - fetch_url = FetchUrlProxy(fetch_url_calls or []) - with pytest.raises(ModuleExitException) as e: - run_module(mocker, module, arguments, fetch_url) - fetch_url.assert_is_done() - return e.value.kwargs - - -def run_module_failed(mocker, module, arguments, fetch_url_calls): - fetch_url = FetchUrlProxy(fetch_url_calls or []) - with pytest.raises(ModuleFailException) as e: - run_module(mocker, module, arguments, fetch_url) - fetch_url.assert_is_done() - return e.value.kwargs - - -# ########################################################## -# ## Hetzner firewall tests - - -# Tests for state (absent and present) - - -def test_absent_idempotency(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'absent', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['diff']['before']['status'] == 'disabled' - assert result['diff']['after']['status'] == 'disabled' - assert result['firewall']['status'] == 'disabled' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_absent_changed(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'absent', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('status', 'disabled'), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'disabled' - assert result['firewall']['status'] == 'disabled' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_present_idempotency(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_present_changed(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('status', 'active'), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'disabled' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -# Tests for state (absent and present) with check mode - - -def test_absent_idempotency_check(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'absent', - '_ansible_check_mode': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['diff']['before']['status'] == 'disabled' - assert result['diff']['after']['status'] == 'disabled' - assert result['firewall']['status'] == 'disabled' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_absent_changed_check(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'absent', - '_ansible_check_mode': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'disabled' - assert result['firewall']['status'] == 'disabled' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_present_idempotency_check(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - '_ansible_check_mode': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_present_changed_check(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - '_ansible_check_mode': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'disabled' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -# Tests for port - - -def test_port_idempotency(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'port': 'main', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['diff']['before']['port'] == 'main' - assert result['diff']['after']['port'] == 'main' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - assert result['firewall']['port'] == 'main' - - -def test_port_changed(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'port': 'main', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': True, - 'port': 'kvm', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('port', 'main'), - ]) - assert result['changed'] is True - assert result['diff']['before']['port'] == 'kvm' - assert result['diff']['after']['port'] == 'main' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - assert result['firewall']['port'] == 'main' - - -# Tests for whitelist_hos - - -def test_whitelist_hos_idempotency(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'whitelist_hos': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['diff']['before']['whitelist_hos'] is True - assert result['diff']['after']['whitelist_hos'] is True - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - assert result['firewall']['whitelist_hos'] is True - - -def test_whitelist_hos_changed(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'whitelist_hos': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('whitelist_hos', 'true'), - ]) - assert result['changed'] is True - assert result['diff']['before']['whitelist_hos'] is False - assert result['diff']['after']['whitelist_hos'] is True - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - assert result['firewall']['whitelist_hos'] is True - - -# Tests for wait_for_configured in getting status - - -def test_wait_get(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'wait_for_configured': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_wait_get_timeout(mocker): - result = run_module_failed(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'wait_for_configured': True, - 'timeout': 0, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['msg'] == 'Timeout while waiting for firewall to be configured.' - - -def test_nowait_get(mocker): - result = run_module_failed(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'wait_for_configured': False, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['msg'] == 'Firewall configuration cannot be read as it is not configured.' - - -# Tests for wait_for_configured in setting status - - -def test_wait_update(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'wait_for_configured': True, - 'state': 'present', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'disabled' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_wait_update_timeout(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'wait_for_configured': True, - 'timeout': 0, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'disabled' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'in process' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - assert 'Timeout while waiting for firewall to be configured.' in result['warnings'] - - -def test_nowait_update(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'wait_for_configured': False, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'disabled' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'in process' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -# Idempotency checks: different amount of input rules - -def test_input_rule_len_change_0_1(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'rules': { - 'input': [ - { - 'ip_version': 'ipv4', - 'action': 'discard', - }, - ], - }, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [ - { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - }, - ], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('status', 'active') - .expect_form_value_absent('rules[input][0][name]') - .expect_form_value('rules[input][0][ip_version]', 'ipv4') - .expect_form_value_absent('rules[input][0][dst_ip]') - .expect_form_value_absent('rules[input][0][dst_port]') - .expect_form_value_absent('rules[input][0][src_ip]') - .expect_form_value_absent('rules[input][0][src_port]') - .expect_form_value_absent('rules[input][0][protocol]') - .expect_form_value_absent('rules[input][0][tcp_flags]') - .expect_form_value('rules[input][0][action]', 'discard') - .expect_form_value_absent('rules[input][1][action]'), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert len(result['diff']['before']['rules']['input']) == 0 - assert len(result['diff']['after']['rules']['input']) == 1 - assert result['firewall']['status'] == 'active' - assert len(result['firewall']['rules']['input']) == 1 - - -def test_input_rule_len_change_1_0(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'rules': { - }, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [ - { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - }, - ], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('status', 'active') - .expect_form_value_absent('rules[input][0][action]'), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert len(result['diff']['before']['rules']['input']) == 1 - assert len(result['diff']['after']['rules']['input']) == 0 - assert result['firewall']['status'] == 'active' - assert len(result['firewall']['rules']['input']) == 0 - - -def test_input_rule_len_change_1_2(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'rules': { - 'input': [ - { - 'ip_version': 'ipv4', - 'dst_port': 80, - 'protocol': 'tcp', - 'action': 'accept', - }, - { - 'ip_version': 'ipv4', - 'action': 'discard', - }, - ], - }, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [ - { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - }, - ], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [ - { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': '80', - 'src_ip': None, - 'src_port': None, - 'protocol': 'tcp', - 'tcp_flags': None, - 'action': 'accept', - }, - { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - }, - ], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('status', 'active') - .expect_form_value('rules[input][0][action]', 'accept') - .expect_form_value('rules[input][1][action]', 'discard') - .expect_form_value_absent('rules[input][2][action]'), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert len(result['diff']['before']['rules']['input']) == 1 - assert len(result['diff']['after']['rules']['input']) == 2 - assert result['firewall']['status'] == 'active' - assert len(result['firewall']['rules']['input']) == 2 - - -# Idempotency checks: change one value - - def create_params(parameter, *values): assert len(values) > 1 result = [] @@ -1188,69 +31,135 @@ def flatten(list_of_lists): return result -@pytest.mark.parametrize("parameter, before, after", flatten([ - create_params('name', None, '', 'Test', 'Test', 'foo', '', None), - create_params('ip_version', 'ipv4', 'ipv4', 'ipv6', 'ipv6'), - create_params('dst_ip', None, '1.2.3.4/24', '1.2.3.4/32', '1.2.3.4/32', None), - create_params('dst_port', None, '80', '80-443', '80-443', None), - create_params('src_ip', None, '1.2.3.4/24', '1.2.3.4/32', '1.2.3.4/32', None), - create_params('src_port', None, '80', '80-443', '80-443', None), - create_params('protocol', None, 'tcp', 'tcp', 'udp', 'udp', None), - create_params('tcp_flags', None, 'syn', 'syn|fin', 'syn|fin', 'syn&fin', '', None), - create_params('action', 'accept', 'accept', 'discard', 'discard'), -])) -def test_input_rule_value_change(mocker, parameter, before, after): - input_call = { - 'ip_version': 'ipv4', - 'action': 'discard', - } - input_before = { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - } - input_after = { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - } - if after is not None: - input_call[parameter] = after - input_before[parameter] = before - input_after[parameter] = after +class TestHetznerFirewall(BaseTestModule): + MOCK_ANSIBLE_MODULEUTILS_BASIC_ANSIBLEMODULE = 'ansible_collections.community.general.plugins.modules.net_tools.hetzner_firewall.AnsibleModule' + MOCK_ANSIBLE_MODULEUTILS_URLS_FETCH_URL = 'ansible_collections.community.general.plugins.module_utils.hetzner.fetch_url' - calls = [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [input_before], + # Tests for state (absent and present) + + def test_absent_idempotency(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'absent', + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'disabled', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ] + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is False + assert result['diff']['before']['status'] == 'disabled' + assert result['diff']['after']['status'] == 'disabled' + assert result['firewall']['status'] == 'disabled' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 - changed = (before != after) - if changed: - after_call = ( + def test_absent_changed(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'absent', + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': True, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + FetchUrlCall('POST', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'disabled', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) + .expect_form_value('status', 'disabled'), + ]) + assert result['changed'] is True + assert result['diff']['before']['status'] == 'active' + assert result['diff']['after']['status'] == 'disabled' + assert result['firewall']['status'] == 'disabled' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + + def test_present_idempotency(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is False + assert result['diff']['before']['status'] == 'active' + assert result['diff']['after']['status'] == 'active' + assert result['firewall']['status'] == 'active' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + + def test_present_changed(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'disabled', + 'whitelist_hos': True, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), FetchUrlCall('POST', 200) .result_json({ 'firewall': { @@ -1260,146 +169,1025 @@ def test_input_rule_value_change(mocker, parameter, before, after): 'whitelist_hos': False, 'port': 'main', 'rules': { - 'input': [input_after], + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) + .expect_form_value('status', 'active'), + ]) + assert result['changed'] is True + assert result['diff']['before']['status'] == 'disabled' + assert result['diff']['after']['status'] == 'active' + assert result['firewall']['status'] == 'active' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + + # Tests for state (absent and present) with check mode + + def test_absent_idempotency_check(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'absent', + '_ansible_check_mode': True, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'disabled', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is False + assert result['diff']['before']['status'] == 'disabled' + assert result['diff']['after']['status'] == 'disabled' + assert result['firewall']['status'] == 'disabled' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + + def test_absent_changed_check(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'absent', + '_ansible_check_mode': True, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': True, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is True + assert result['diff']['before']['status'] == 'active' + assert result['diff']['after']['status'] == 'disabled' + assert result['firewall']['status'] == 'disabled' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + + def test_present_idempotency_check(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + '_ansible_check_mode': True, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is False + assert result['diff']['before']['status'] == 'active' + assert result['diff']['after']['status'] == 'active' + assert result['firewall']['status'] == 'active' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + + def test_present_changed_check(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + '_ansible_check_mode': True, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'disabled', + 'whitelist_hos': True, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is True + assert result['diff']['before']['status'] == 'disabled' + assert result['diff']['after']['status'] == 'active' + assert result['firewall']['status'] == 'active' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + + # Tests for port + + def test_port_idempotency(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + 'port': 'main', + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is False + assert result['diff']['before']['port'] == 'main' + assert result['diff']['after']['port'] == 'main' + assert result['firewall']['status'] == 'active' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + assert result['firewall']['port'] == 'main' + + def test_port_changed(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + 'port': 'main', + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'disabled', + 'whitelist_hos': True, + 'port': 'kvm', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + FetchUrlCall('POST', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) + .expect_form_value('port', 'main'), + ]) + assert result['changed'] is True + assert result['diff']['before']['port'] == 'kvm' + assert result['diff']['after']['port'] == 'main' + assert result['firewall']['status'] == 'active' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + assert result['firewall']['port'] == 'main' + + # Tests for whitelist_hos + + def test_whitelist_hos_idempotency(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + 'whitelist_hos': True, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': True, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is False + assert result['diff']['before']['whitelist_hos'] is True + assert result['diff']['after']['whitelist_hos'] is True + assert result['firewall']['status'] == 'active' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + assert result['firewall']['whitelist_hos'] is True + + def test_whitelist_hos_changed(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + 'whitelist_hos': True, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'disabled', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + FetchUrlCall('POST', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': True, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) + .expect_form_value('whitelist_hos', 'true'), + ]) + assert result['changed'] is True + assert result['diff']['before']['whitelist_hos'] is False + assert result['diff']['after']['whitelist_hos'] is True + assert result['firewall']['status'] == 'active' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + assert result['firewall']['whitelist_hos'] is True + + # Tests for wait_for_configured in getting status + + def test_wait_get(self, mocker): + mocker.patch('time.sleep', lambda duration: None) + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + 'wait_for_configured': True, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'in process', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is False + assert result['diff']['before']['status'] == 'active' + assert result['diff']['after']['status'] == 'active' + assert result['firewall']['status'] == 'active' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + + def test_wait_get_timeout(self, mocker): + mocker.patch('time.sleep', lambda duration: None) + result = self.run_module_failed(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + 'wait_for_configured': True, + 'timeout': 0, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'in process', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'in process', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['msg'] == 'Timeout while waiting for firewall to be configured.' + + def test_nowait_get(self, mocker): + result = self.run_module_failed(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + 'wait_for_configured': False, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'in process', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['msg'] == 'Firewall configuration cannot be read as it is not configured.' + + # Tests for wait_for_configured in setting status + + def test_wait_update(self, mocker): + mocker.patch('time.sleep', lambda duration: None) + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'wait_for_configured': True, + 'state': 'present', + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'disabled', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + FetchUrlCall('POST', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'in process', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is True + assert result['diff']['before']['status'] == 'disabled' + assert result['diff']['after']['status'] == 'active' + assert result['firewall']['status'] == 'active' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + + def test_wait_update_timeout(self, mocker): + mocker.patch('time.sleep', lambda duration: None) + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + 'wait_for_configured': True, + 'timeout': 0, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'disabled', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + FetchUrlCall('POST', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'in process', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'in process', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is True + assert result['diff']['before']['status'] == 'disabled' + assert result['diff']['after']['status'] == 'active' + assert result['firewall']['status'] == 'in process' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + assert 'Timeout while waiting for firewall to be configured.' in result['warnings'] + + def test_nowait_update(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + 'wait_for_configured': False, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'disabled', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + FetchUrlCall('POST', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'in process', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is True + assert result['diff']['before']['status'] == 'disabled' + assert result['diff']['after']['status'] == 'active' + assert result['firewall']['status'] == 'in process' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + + # Idempotency checks: different amount of input rules + + def test_input_rule_len_change_0_1(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + 'rules': { + 'input': [ + { + 'ip_version': 'ipv4', + 'action': 'discard', + }, + ], + }, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': True, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + FetchUrlCall('POST', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [ + { + 'name': None, + 'ip_version': 'ipv4', + 'dst_ip': None, + 'dst_port': None, + 'src_ip': None, + 'src_port': None, + 'protocol': None, + 'tcp_flags': None, + 'action': 'discard', + }, + ], }, }, }) .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) .expect_form_value('status', 'active') - .expect_form_value_absent('rules[input][1][action]') - ) - if parameter != 'ip_version': - after_call.expect_form_value('rules[input][0][ip_version]', 'ipv4') - if parameter != 'action': - after_call.expect_form_value('rules[input][0][action]', 'discard') + .expect_form_value_absent('rules[input][0][name]') + .expect_form_value('rules[input][0][ip_version]', 'ipv4') + .expect_form_value_absent('rules[input][0][dst_ip]') + .expect_form_value_absent('rules[input][0][dst_port]') + .expect_form_value_absent('rules[input][0][src_ip]') + .expect_form_value_absent('rules[input][0][src_port]') + .expect_form_value_absent('rules[input][0][protocol]') + .expect_form_value_absent('rules[input][0][tcp_flags]') + .expect_form_value('rules[input][0][action]', 'discard') + .expect_form_value_absent('rules[input][1][action]'), + ]) + assert result['changed'] is True + assert result['diff']['before']['status'] == 'active' + assert result['diff']['after']['status'] == 'active' + assert len(result['diff']['before']['rules']['input']) == 0 + assert len(result['diff']['after']['rules']['input']) == 1 + assert result['firewall']['status'] == 'active' + assert len(result['firewall']['rules']['input']) == 1 + + def test_input_rule_len_change_1_0(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + 'rules': { + }, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': True, + 'port': 'main', + 'rules': { + 'input': [ + { + 'name': None, + 'ip_version': 'ipv4', + 'dst_ip': None, + 'dst_port': None, + 'src_ip': None, + 'src_port': None, + 'protocol': None, + 'tcp_flags': None, + 'action': 'discard', + }, + ], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + FetchUrlCall('POST', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) + .expect_form_value('status', 'active') + .expect_form_value_absent('rules[input][0][action]'), + ]) + assert result['changed'] is True + assert result['diff']['before']['status'] == 'active' + assert result['diff']['after']['status'] == 'active' + assert len(result['diff']['before']['rules']['input']) == 1 + assert len(result['diff']['after']['rules']['input']) == 0 + assert result['firewall']['status'] == 'active' + assert len(result['firewall']['rules']['input']) == 0 + + def test_input_rule_len_change_1_2(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + 'rules': { + 'input': [ + { + 'ip_version': 'ipv4', + 'dst_port': 80, + 'protocol': 'tcp', + 'action': 'accept', + }, + { + 'ip_version': 'ipv4', + 'action': 'discard', + }, + ], + }, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': True, + 'port': 'main', + 'rules': { + 'input': [ + { + 'name': None, + 'ip_version': 'ipv4', + 'dst_ip': None, + 'dst_port': None, + 'src_ip': None, + 'src_port': None, + 'protocol': None, + 'tcp_flags': None, + 'action': 'discard', + }, + ], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + FetchUrlCall('POST', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [ + { + 'name': None, + 'ip_version': 'ipv4', + 'dst_ip': None, + 'dst_port': '80', + 'src_ip': None, + 'src_port': None, + 'protocol': 'tcp', + 'tcp_flags': None, + 'action': 'accept', + }, + { + 'name': None, + 'ip_version': 'ipv4', + 'dst_ip': None, + 'dst_port': None, + 'src_ip': None, + 'src_port': None, + 'protocol': None, + 'tcp_flags': None, + 'action': 'discard', + }, + ], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) + .expect_form_value('status', 'active') + .expect_form_value('rules[input][0][action]', 'accept') + .expect_form_value('rules[input][1][action]', 'discard') + .expect_form_value_absent('rules[input][2][action]'), + ]) + assert result['changed'] is True + assert result['diff']['before']['status'] == 'active' + assert result['diff']['after']['status'] == 'active' + assert len(result['diff']['before']['rules']['input']) == 1 + assert len(result['diff']['after']['rules']['input']) == 2 + assert result['firewall']['status'] == 'active' + assert len(result['firewall']['rules']['input']) == 2 + + # Idempotency checks: change one value + + @pytest.mark.parametrize("parameter, before, after", flatten([ + create_params('name', None, '', 'Test', 'Test', 'foo', '', None), + create_params('ip_version', 'ipv4', 'ipv4', 'ipv6', 'ipv6'), + create_params('dst_ip', None, '1.2.3.4/24', '1.2.3.4/32', '1.2.3.4/32', None), + create_params('dst_port', None, '80', '80-443', '80-443', None), + create_params('src_ip', None, '1.2.3.4/24', '1.2.3.4/32', '1.2.3.4/32', None), + create_params('src_port', None, '80', '80-443', '80-443', None), + create_params('protocol', None, 'tcp', 'tcp', 'udp', 'udp', None), + create_params('tcp_flags', None, 'syn', 'syn|fin', 'syn|fin', 'syn&fin', '', None), + create_params('action', 'accept', 'accept', 'discard', 'discard'), + ])) + def test_input_rule_value_change(self, mocker, parameter, before, after): + input_call = { + 'ip_version': 'ipv4', + 'action': 'discard', + } + input_before = { + 'name': None, + 'ip_version': 'ipv4', + 'dst_ip': None, + 'dst_port': None, + 'src_ip': None, + 'src_port': None, + 'protocol': None, + 'tcp_flags': None, + 'action': 'discard', + } + input_after = { + 'name': None, + 'ip_version': 'ipv4', + 'dst_ip': None, + 'dst_port': None, + 'src_ip': None, + 'src_port': None, + 'protocol': None, + 'tcp_flags': None, + 'action': 'discard', + } if after is not None: - after_call.expect_form_value('rules[input][0][{0}]'.format(parameter), after) - else: - after_call.expect_form_value_absent('rules[input][0][{0}]'.format(parameter)) - calls.append(after_call) + input_call[parameter] = after + input_before[parameter] = before + input_after[parameter] = after - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'rules': { - 'input': [input_call], - }, - }, calls) - assert result['changed'] == changed - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert len(result['diff']['before']['rules']['input']) == 1 - assert len(result['diff']['after']['rules']['input']) == 1 - assert result['diff']['before']['rules']['input'][0][parameter] == before - assert result['diff']['after']['rules']['input'][0][parameter] == after - assert result['firewall']['status'] == 'active' - assert len(result['firewall']['rules']['input']) == 1 - assert result['firewall']['rules']['input'][0][parameter] == after - - -# Idempotency checks: IP address normalization - - -@pytest.mark.parametrize("ip_version, parameter, before_normalized, after_normalized, after", [ - ('ipv4', 'src_ip', '1.2.3.4/32', '1.2.3.4/32', '1.2.3.4'), - ('ipv6', 'src_ip', '1:2:3::4/128', '1:2:3::4/128', '1:2:3::4'), - ('ipv6', 'dst_ip', '1:2:3::4/128', '1:2:3::4/128', '1:2:3:0::4'), - ('ipv6', 'dst_ip', '::/0', '::/0', '0:0::0/0'), -]) -def test_input_rule_ip_normalization(mocker, ip_version, parameter, before_normalized, after_normalized, after): - assert ip_version in ('ipv4', 'ipv6') - assert parameter in ('src_ip', 'dst_ip') - input_call = { - 'ip_version': ip_version, - 'action': 'discard', - } - input_before = { - 'name': None, - 'ip_version': ip_version, - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - } - input_after = { - 'name': None, - 'ip_version': ip_version, - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - } - if after is not None: - input_call[parameter] = after - input_before[parameter] = before_normalized - input_after[parameter] = after_normalized - - calls = [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [input_before], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ] - - changed = (before_normalized != after_normalized) - if changed: - after_call = ( - FetchUrlCall('POST', 200) + calls = [ + FetchUrlCall('GET', 200) .result_json({ 'firewall': { 'server_ip': '1.2.3.4', 'server_number': 1, 'status': 'active', - 'whitelist_hos': False, + 'whitelist_hos': True, 'port': 'main', 'rules': { - 'input': [input_after], + 'input': [input_before], }, }, }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('status', 'active') - .expect_form_value_absent('rules[input][1][action]') - ) - after_call.expect_form_value('rules[input][0][ip_version]', ip_version) - after_call.expect_form_value('rules[input][0][action]', 'discard') - after_call.expect_form_value('rules[input][0][{0}]'.format(parameter), after_normalized) - calls.append(after_call) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ] - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'rules': { - 'input': [input_call], - }, - }, calls) - assert result['changed'] == changed - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert len(result['diff']['before']['rules']['input']) == 1 - assert len(result['diff']['after']['rules']['input']) == 1 - assert result['diff']['before']['rules']['input'][0][parameter] == before_normalized - assert result['diff']['after']['rules']['input'][0][parameter] == after_normalized - assert result['firewall']['status'] == 'active' - assert len(result['firewall']['rules']['input']) == 1 - assert result['firewall']['rules']['input'][0][parameter] == after_normalized + changed = (before != after) + if changed: + after_call = ( + FetchUrlCall('POST', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [input_after], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) + .expect_form_value('status', 'active') + .expect_form_value_absent('rules[input][1][action]') + ) + if parameter != 'ip_version': + after_call.expect_form_value('rules[input][0][ip_version]', 'ipv4') + if parameter != 'action': + after_call.expect_form_value('rules[input][0][action]', 'discard') + if after is not None: + after_call.expect_form_value('rules[input][0][{0}]'.format(parameter), after) + else: + after_call.expect_form_value_absent('rules[input][0][{0}]'.format(parameter)) + calls.append(after_call) + + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + 'rules': { + 'input': [input_call], + }, + }, calls) + assert result['changed'] == changed + assert result['diff']['before']['status'] == 'active' + assert result['diff']['after']['status'] == 'active' + assert len(result['diff']['before']['rules']['input']) == 1 + assert len(result['diff']['after']['rules']['input']) == 1 + assert result['diff']['before']['rules']['input'][0][parameter] == before + assert result['diff']['after']['rules']['input'][0][parameter] == after + assert result['firewall']['status'] == 'active' + assert len(result['firewall']['rules']['input']) == 1 + assert result['firewall']['rules']['input'][0][parameter] == after + + # Idempotency checks: IP address normalization + + @pytest.mark.parametrize("ip_version, parameter, before_normalized, after_normalized, after", [ + ('ipv4', 'src_ip', '1.2.3.4/32', '1.2.3.4/32', '1.2.3.4'), + ('ipv6', 'src_ip', '1:2:3::4/128', '1:2:3::4/128', '1:2:3::4'), + ('ipv6', 'dst_ip', '1:2:3::4/128', '1:2:3::4/128', '1:2:3:0::4'), + ('ipv6', 'dst_ip', '::/0', '::/0', '0:0::0/0'), + ]) + def test_input_rule_ip_normalization(self, mocker, ip_version, parameter, before_normalized, after_normalized, after): + assert ip_version in ('ipv4', 'ipv6') + assert parameter in ('src_ip', 'dst_ip') + input_call = { + 'ip_version': ip_version, + 'action': 'discard', + } + input_before = { + 'name': None, + 'ip_version': ip_version, + 'dst_ip': None, + 'dst_port': None, + 'src_ip': None, + 'src_port': None, + 'protocol': None, + 'tcp_flags': None, + 'action': 'discard', + } + input_after = { + 'name': None, + 'ip_version': ip_version, + 'dst_ip': None, + 'dst_port': None, + 'src_ip': None, + 'src_port': None, + 'protocol': None, + 'tcp_flags': None, + 'action': 'discard', + } + if after is not None: + input_call[parameter] = after + input_before[parameter] = before_normalized + input_after[parameter] = after_normalized + + calls = [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': True, + 'port': 'main', + 'rules': { + 'input': [input_before], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ] + + changed = (before_normalized != after_normalized) + if changed: + after_call = ( + FetchUrlCall('POST', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [input_after], + }, + }, + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) + .expect_form_value('status', 'active') + .expect_form_value_absent('rules[input][1][action]') + ) + after_call.expect_form_value('rules[input][0][ip_version]', ip_version) + after_call.expect_form_value('rules[input][0][action]', 'discard') + after_call.expect_form_value('rules[input][0][{0}]'.format(parameter), after_normalized) + calls.append(after_call) + + result = self.run_module_success(mocker, hetzner_firewall, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'state': 'present', + 'rules': { + 'input': [input_call], + }, + }, calls) + assert result['changed'] == changed + assert result['diff']['before']['status'] == 'active' + assert result['diff']['after']['status'] == 'active' + assert len(result['diff']['before']['rules']['input']) == 1 + assert len(result['diff']['after']['rules']['input']) == 1 + assert result['diff']['before']['rules']['input'][0][parameter] == before_normalized + assert result['diff']['after']['rules']['input'][0][parameter] == after_normalized + assert result['firewall']['status'] == 'active' + assert len(result['firewall']['rules']['input']) == 1 + assert result['firewall']['rules']['input'][0][parameter] == after_normalized diff --git a/tests/unit/plugins/modules/net_tools/test_hetzner_firewall_info.py b/tests/unit/plugins/modules/net_tools/test_hetzner_firewall_info.py index 6b43663c9f..3d70df4788 100644 --- a/tests/unit/plugins/modules/net_tools/test_hetzner_firewall_info.py +++ b/tests/unit/plugins/modules/net_tools/test_hetzner_firewall_info.py @@ -5,235 +5,236 @@ from __future__ import (absolute_import, division, print_function) __metaclass__ = type -import pytest +from ansible_collections.community.internal_test_tools.tests.unit.utils.fetch_url_module_framework import ( + FetchUrlCall, + BaseTestModule, +) from ansible_collections.community.general.plugins.module_utils.hetzner import BASE_URL from ansible_collections.community.general.plugins.modules.net_tools import hetzner_firewall_info -from .test_hetzner_firewall import FetchUrlCall, run_module_success, run_module_failed -# Tests for state (absent and present) +class TestHetznerFirewallInfo(BaseTestModule): + MOCK_ANSIBLE_MODULEUTILS_BASIC_ANSIBLEMODULE = 'ansible_collections.community.general.plugins.modules.net_tools.hetzner_firewall_info.AnsibleModule' + MOCK_ANSIBLE_MODULEUTILS_URLS_FETCH_URL = 'ansible_collections.community.general.plugins.module_utils.hetzner.fetch_url' + # Tests for state (absent and present) -def test_absent(mocker): - result = run_module_success(mocker, hetzner_firewall_info, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], + def test_absent(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall_info, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'disabled', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['firewall']['status'] == 'disabled' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is False + assert result['firewall']['status'] == 'disabled' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 - -def test_present(mocker): - result = run_module_success(mocker, hetzner_firewall_info, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], + def test_present(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall_info, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - assert len(result['firewall']['rules']['input']) == 0 + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is False + assert result['firewall']['status'] == 'active' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + assert len(result['firewall']['rules']['input']) == 0 - -def test_present_w_rules(mocker): - result = run_module_success(mocker, hetzner_firewall_info, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [ - { - 'name': 'Accept HTTPS traffic', - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': '443', - 'src_ip': None, - 'src_port': None, - 'protocol': 'tcp', - 'tcp_flags': None, - 'action': 'accept', - }, - { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - } - ], + def test_present_w_rules(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall_info, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [ + { + 'name': 'Accept HTTPS traffic', + 'ip_version': 'ipv4', + 'dst_ip': None, + 'dst_port': '443', + 'src_ip': None, + 'src_port': None, + 'protocol': 'tcp', + 'tcp_flags': None, + 'action': 'accept', + }, + { + 'name': None, + 'ip_version': 'ipv4', + 'dst_ip': None, + 'dst_port': None, + 'src_ip': None, + 'src_port': None, + 'protocol': None, + 'tcp_flags': None, + 'action': 'discard', + } + ], + }, }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - assert len(result['firewall']['rules']['input']) == 2 - assert result['firewall']['rules']['input'][0]['name'] == 'Accept HTTPS traffic' - assert result['firewall']['rules']['input'][0]['dst_port'] == '443' - assert result['firewall']['rules']['input'][0]['action'] == 'accept' - assert result['firewall']['rules']['input'][1]['dst_port'] is None - assert result['firewall']['rules']['input'][1]['action'] == 'discard' + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is False + assert result['firewall']['status'] == 'active' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 + assert len(result['firewall']['rules']['input']) == 2 + assert result['firewall']['rules']['input'][0]['name'] == 'Accept HTTPS traffic' + assert result['firewall']['rules']['input'][0]['dst_port'] == '443' + assert result['firewall']['rules']['input'][0]['action'] == 'accept' + assert result['firewall']['rules']['input'][1]['dst_port'] is None + assert result['firewall']['rules']['input'][1]['action'] == 'discard' + # Tests for wait_for_configured in getting status -# Tests for wait_for_configured in getting status - - -def test_wait_get(mocker): - result = run_module_success(mocker, hetzner_firewall_info, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'wait_for_configured': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], + def test_wait_get(self, mocker): + mocker.patch('time.sleep', lambda duration: None) + result = self.run_module_success(mocker, hetzner_firewall_info, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'wait_for_configured': True, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'in process', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'active', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is False + assert result['firewall']['status'] == 'active' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 - -def test_wait_get_timeout(mocker): - result = run_module_failed(mocker, hetzner_firewall_info, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'wait_for_configured': True, - 'timeout': 0, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], + def test_wait_get_timeout(self, mocker): + mocker.patch('time.sleep', lambda duration: None) + result = self.run_module_failed(mocker, hetzner_firewall_info, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'wait_for_configured': True, + 'timeout': 0, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'in process', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'in process', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['msg'] == 'Timeout while waiting for firewall to be configured.' + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['msg'] == 'Timeout while waiting for firewall to be configured.' - -def test_nowait_get(mocker): - result = run_module_success(mocker, hetzner_firewall_info, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'wait_for_configured': False, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], + def test_nowait_get(self, mocker): + result = self.run_module_success(mocker, hetzner_firewall_info, { + 'hetzner_user': '', + 'hetzner_password': '', + 'server_ip': '1.2.3.4', + 'wait_for_configured': False, + }, [ + FetchUrlCall('GET', 200) + .result_json({ + 'firewall': { + 'server_ip': '1.2.3.4', + 'server_number': 1, + 'status': 'in process', + 'whitelist_hos': False, + 'port': 'main', + 'rules': { + 'input': [], + }, }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['firewall']['status'] == 'in process' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 + }) + .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), + ]) + assert result['changed'] is False + assert result['firewall']['status'] == 'in process' + assert result['firewall']['server_ip'] == '1.2.3.4' + assert result['firewall']['server_number'] == 1 diff --git a/tests/utils/shippable/shippable.sh b/tests/utils/shippable/shippable.sh index c5b3143512..018b0a5c62 100755 --- a/tests/utils/shippable/shippable.sh +++ b/tests/utils/shippable/shippable.sh @@ -60,6 +60,7 @@ retry ansible-galaxy -vvv collection install ansible.posix # https://github.com/CiscoDevNet/ansible-intersight/issues/9 retry ansible-galaxy -vvv collection install cisco.intersight:1.0.4 retry ansible-galaxy -vvv collection install community.crypto +retry ansible-galaxy -vvv collection install community.internal_test_tools retry ansible-galaxy -vvv collection install community.kubernetes retry ansible-galaxy -vvv collection install google.cloud retry ansible-galaxy -vvv collection install ovirt.ovirt