mirror of https://github.com/ansible/ansible.git
You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
266 lines
7.1 KiB
Python
266 lines
7.1 KiB
Python
![]()
5 years ago
|
# Copyright: (c) 2017 Ansible Project
|
||
|
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
|
||
|
|
||
![]()
5 years ago
|
import copy
|
||
![]()
5 years ago
|
import json
|
||
|
import pytest
|
||
|
|
||
|
from mock import MagicMock
|
||
![]()
5 years ago
|
from ansible.module_utils import hetzner
|
||
![]()
5 years ago
|
|
||
|
|
||
|
class ModuleFailException(Exception):
|
||
|
def __init__(self, msg, **kwargs):
|
||
|
super(ModuleFailException, self).__init__(msg)
|
||
|
self.fail_msg = msg
|
||
|
self.fail_kwargs = kwargs
|
||
|
|
||
|
|
||
|
def get_module_mock():
|
||
|
def f(msg, **kwargs):
|
||
|
raise ModuleFailException(msg, **kwargs)
|
||
|
|
||
|
module = MagicMock()
|
||
|
module.fail_json = f
|
||
|
module.from_json = json.loads
|
||
|
return module
|
||
|
|
||
|
|
||
|
# ########################################################################################
|
||
|
|
||
|
FETCH_URL_JSON_SUCCESS = [
|
||
|
(
|
||
|
(None, dict(
|
||
|
body=json.dumps(dict(
|
||
|
a='b'
|
||
|
)).encode('utf-8'),
|
||
|
)),
|
||
|
None,
|
||
|
(dict(
|
||
|
a='b'
|
||
|
), None)
|
||
|
),
|
||
|
(
|
||
|
(None, dict(
|
||
|
body=json.dumps(dict(
|
||
|
error=dict(
|
||
|
code="foo",
|
||
|
status=400,
|
||
|
message="bar",
|
||
|
),
|
||
|
a='b'
|
||
|
)).encode('utf-8'),
|
||
|
)),
|
||
|
['foo'],
|
||
|
(dict(
|
||
|
error=dict(
|
||
|
code="foo",
|
||
|
status=400,
|
||
|
message="bar",
|
||
|
),
|
||
|
a='b'
|
||
|
), 'foo')
|
||
|
),
|
||
|
]
|
||
|
|
||
|
|
||
|
FETCH_URL_JSON_FAIL = [
|
||
|
(
|
||
|
(None, dict(
|
||
|
body=json.dumps(dict(
|
||
|
error=dict(
|
||
|
code="foo",
|
||
|
status=400,
|
||
|
message="bar",
|
||
|
),
|
||
|
)).encode('utf-8'),
|
||
|
)),
|
||
|
None,
|
||
|
'Request failed: 400 foo (bar)'
|
||
|
),
|
||
|
(
|
||
|
(None, dict(
|
||
|
body=json.dumps(dict(
|
||
|
error=dict(
|
||
|
code="foo",
|
||
|
status=400,
|
||
|
message="bar",
|
||
|
),
|
||
|
)).encode('utf-8'),
|
||
|
)),
|
||
|
['bar'],
|
||
|
'Request failed: 400 foo (bar)'
|
||
|
),
|
||
|
]
|
||
|
|
||
|
|
||
|
@pytest.mark.parametrize("return_value, accept_errors, result", FETCH_URL_JSON_SUCCESS)
|
||
|
def test_fetch_url_json(monkeypatch, return_value, accept_errors, result):
|
||
|
module = get_module_mock()
|
||
![]()
5 years ago
|
hetzner.fetch_url = MagicMock(return_value=return_value)
|
||
![]()
5 years ago
|
|
||
![]()
5 years ago
|
assert hetzner.fetch_url_json(module, 'https://foo/bar', accept_errors=accept_errors) == result
|
||
![]()
5 years ago
|
|
||
|
|
||
|
@pytest.mark.parametrize("return_value, accept_errors, result", FETCH_URL_JSON_FAIL)
|
||
|
def test_fetch_url_json_fail(monkeypatch, return_value, accept_errors, result):
|
||
|
module = get_module_mock()
|
||
![]()
5 years ago
|
hetzner.fetch_url = MagicMock(return_value=return_value)
|
||
![]()
5 years ago
|
|
||
|
with pytest.raises(ModuleFailException) as exc:
|
||
![]()
5 years ago
|
hetzner.fetch_url_json(module, 'https://foo/bar', accept_errors=accept_errors)
|
||
![]()
5 years ago
|
|
||
|
assert exc.value.fail_msg == result
|
||
|
assert exc.value.fail_kwargs == dict()
|
||
|
|
||
|
|
||
|
# ########################################################################################
|
||
|
|
||
|
GET_FAILOVER_SUCCESS = [
|
||
|
(
|
||
|
'1.2.3.4',
|
||
|
(None, dict(
|
||
|
body=json.dumps(dict(
|
||
|
failover=dict(
|
||
|
active_server_ip='1.1.1.1',
|
||
![]()
5 years ago
|
ip='1.2.3.4',
|
||
|
netmask='255.255.255.255',
|
||
![]()
5 years ago
|
)
|
||
|
)).encode('utf-8'),
|
||
|
)),
|
||
![]()
5 years ago
|
'1.1.1.1',
|
||
|
dict(
|
||
|
active_server_ip='1.1.1.1',
|
||
|
ip='1.2.3.4',
|
||
|
netmask='255.255.255.255',
|
||
|
)
|
||
![]()
5 years ago
|
),
|
||
|
]
|
||
|
|
||
|
|
||
|
GET_FAILOVER_FAIL = [
|
||
|
(
|
||
|
'1.2.3.4',
|
||
|
(None, dict(
|
||
|
body=json.dumps(dict(
|
||
|
error=dict(
|
||
|
code="foo",
|
||
|
status=400,
|
||
|
message="bar",
|
||
|
),
|
||
|
)).encode('utf-8'),
|
||
|
)),
|
||
|
'Request failed: 400 foo (bar)'
|
||
|
),
|
||
|
]
|
||
|
|
||
|
|
||
![]()
5 years ago
|
@pytest.mark.parametrize("ip, return_value, result, record", GET_FAILOVER_SUCCESS)
|
||
|
def test_get_failover_record(monkeypatch, ip, return_value, result, record):
|
||
|
module = get_module_mock()
|
||
|
hetzner.fetch_url = MagicMock(return_value=copy.deepcopy(return_value))
|
||
|
|
||
|
assert hetzner.get_failover_record(module, ip) == record
|
||
|
|
||
|
|
||
|
@pytest.mark.parametrize("ip, return_value, result", GET_FAILOVER_FAIL)
|
||
|
def test_get_failover_record_fail(monkeypatch, ip, return_value, result):
|
||
|
module = get_module_mock()
|
||
|
hetzner.fetch_url = MagicMock(return_value=copy.deepcopy(return_value))
|
||
|
|
||
|
with pytest.raises(ModuleFailException) as exc:
|
||
|
hetzner.get_failover_record(module, ip)
|
||
|
|
||
|
assert exc.value.fail_msg == result
|
||
|
assert exc.value.fail_kwargs == dict()
|
||
|
|
||
|
|
||
|
@pytest.mark.parametrize("ip, return_value, result, record", GET_FAILOVER_SUCCESS)
|
||
|
def test_get_failover(monkeypatch, ip, return_value, result, record):
|
||
![]()
5 years ago
|
module = get_module_mock()
|
||
![]()
5 years ago
|
hetzner.fetch_url = MagicMock(return_value=copy.deepcopy(return_value))
|
||
![]()
5 years ago
|
|
||
![]()
5 years ago
|
assert hetzner.get_failover(module, ip) == result
|
||
![]()
5 years ago
|
|
||
|
|
||
|
@pytest.mark.parametrize("ip, return_value, result", GET_FAILOVER_FAIL)
|
||
|
def test_get_failover_fail(monkeypatch, ip, return_value, result):
|
||
|
module = get_module_mock()
|
||
![]()
5 years ago
|
hetzner.fetch_url = MagicMock(return_value=copy.deepcopy(return_value))
|
||
![]()
5 years ago
|
|
||
|
with pytest.raises(ModuleFailException) as exc:
|
||
![]()
5 years ago
|
hetzner.get_failover(module, ip)
|
||
![]()
5 years ago
|
|
||
|
assert exc.value.fail_msg == result
|
||
|
assert exc.value.fail_kwargs == dict()
|
||
|
|
||
|
|
||
|
# ########################################################################################
|
||
|
|
||
|
SET_FAILOVER_SUCCESS = [
|
||
|
(
|
||
|
'1.2.3.4',
|
||
|
'1.1.1.1',
|
||
|
(None, dict(
|
||
|
body=json.dumps(dict(
|
||
|
failover=dict(
|
||
|
active_server_ip='1.1.1.2',
|
||
|
)
|
||
|
)).encode('utf-8'),
|
||
|
)),
|
||
|
('1.1.1.2', True)
|
||
|
),
|
||
|
(
|
||
|
'1.2.3.4',
|
||
|
'1.1.1.1',
|
||
|
(None, dict(
|
||
|
body=json.dumps(dict(
|
||
|
error=dict(
|
||
|
code="FAILOVER_ALREADY_ROUTED",
|
||
|
status=400,
|
||
|
message="Failover already routed",
|
||
|
),
|
||
|
)).encode('utf-8'),
|
||
|
)),
|
||
|
('1.1.1.1', False)
|
||
|
),
|
||
|
]
|
||
|
|
||
|
|
||
|
SET_FAILOVER_FAIL = [
|
||
|
(
|
||
|
'1.2.3.4',
|
||
|
'1.1.1.1',
|
||
|
(None, dict(
|
||
|
body=json.dumps(dict(
|
||
|
error=dict(
|
||
|
code="foo",
|
||
|
status=400,
|
||
|
message="bar",
|
||
|
),
|
||
|
)).encode('utf-8'),
|
||
|
)),
|
||
|
'Request failed: 400 foo (bar)'
|
||
|
),
|
||
|
]
|
||
|
|
||
|
|
||
|
@pytest.mark.parametrize("ip, value, return_value, result", SET_FAILOVER_SUCCESS)
|
||
|
def test_set_failover(monkeypatch, ip, value, return_value, result):
|
||
|
module = get_module_mock()
|
||
![]()
5 years ago
|
hetzner.fetch_url = MagicMock(return_value=copy.deepcopy(return_value))
|
||
![]()
5 years ago
|
|
||
![]()
5 years ago
|
assert hetzner.set_failover(module, ip, value) == result
|
||
![]()
5 years ago
|
|
||
|
|
||
|
@pytest.mark.parametrize("ip, value, return_value, result", SET_FAILOVER_FAIL)
|
||
|
def test_set_failover_fail(monkeypatch, ip, value, return_value, result):
|
||
|
module = get_module_mock()
|
||
![]()
5 years ago
|
hetzner.fetch_url = MagicMock(return_value=copy.deepcopy(return_value))
|
||
![]()
5 years ago
|
|
||
|
with pytest.raises(ModuleFailException) as exc:
|
||
![]()
5 years ago
|
hetzner.set_failover(module, ip, value)
|
||
![]()
5 years ago
|
|
||
|
assert exc.value.fail_msg == result
|
||
|
assert exc.value.fail_kwargs == dict()
|