1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
|
"""Unit tests for vyos_rest module_utils."""
import json
from unittest.mock import MagicMock, patch
from urllib.parse import parse_qs
import pytest
# Minimal AnsibleModule mock
class FakeModule:
def __init__(self, params):
self.params = params
def fail_json(self, **kwargs):
raise AssertionError("fail_json called: {0}".format(kwargs))
def _make_client(params=None):
from ansible_collections.vyos.rest.plugins.module_utils.vyos_rest import (
VyOSRestClient,
)
p = {
"hostname": "192.0.2.1",
"port": 443,
"api_key": "test-key",
"timeout": 10,
"verify_ssl": False,
}
if params:
p.update(params)
return VyOSRestClient(FakeModule(p))
def _ok_response(data):
raw = json.dumps({"success": True, "data": data, "error": None})
mock_resp = MagicMock()
mock_resp.read.return_value = raw.encode()
return mock_resp
def _get_payload(mock_open):
"""Extract and decode the JSON payload from a mocked open_url call."""
call_args = mock_open.call_args
data_field = call_args[1]["data"] if "data" in call_args[1] else call_args[0][1]
parsed = parse_qs(data_field)
return json.loads(parsed["data"][0])
class TestVyOSRestClientInit:
def test_base_url(self):
client = _make_client()
assert client.base_url == "https://192.0.2.1:443"
def test_custom_port(self):
client = _make_client({"port": 8443})
assert client.base_url == "https://192.0.2.1:8443"
class TestConfigureSet:
def test_set_path_only(self):
client = _make_client()
with patch(
"ansible_collections.vyos.rest.plugins.module_utils.vyos_rest.open_url",
return_value=_ok_response(None),
) as mock_open:
client.configure_set(["interfaces", "ethernet", "eth0"])
payload = _get_payload(mock_open)
assert payload["op"] == "set"
assert payload["path"] == ["interfaces", "ethernet", "eth0"]
assert "value" not in payload
def test_set_path_with_value(self):
client = _make_client()
with patch(
"ansible_collections.vyos.rest.plugins.module_utils.vyos_rest.open_url",
return_value=_ok_response(None),
) as mock_open:
client.configure_set(["system", "host-name"], "vyos")
payload = _get_payload(mock_open)
assert payload["op"] == "set"
assert payload["path"] == ["system", "host-name"]
assert payload["value"] == "vyos"
class TestConfigureDelete:
def test_delete(self):
client = _make_client()
with patch(
"ansible_collections.vyos.rest.plugins.module_utils.vyos_rest.open_url",
return_value=_ok_response(None),
) as mock_open:
client.configure_delete(["protocols", "static", "route"])
payload = _get_payload(mock_open)
assert payload["op"] == "delete"
assert payload["path"] == ["protocols", "static", "route"]
class TestRetrieve:
def test_show_config(self):
client = _make_client()
expected = {"interfaces": {"ethernet": {"eth0": {"address": "dhcp"}}}}
with patch(
"ansible_collections.vyos.rest.plugins.module_utils.vyos_rest.open_url",
return_value=_ok_response(expected),
):
result = client.retrieve_show_config([])
assert result["data"] == expected
def test_exists_true(self):
client = _make_client()
with patch(
"ansible_collections.vyos.rest.plugins.module_utils.vyos_rest.open_url",
return_value=_ok_response(True),
):
assert client.retrieve_exists(["service", "ssh"]) is True
def test_exists_false(self):
client = _make_client()
with patch(
"ansible_collections.vyos.rest.plugins.module_utils.vyos_rest.open_url",
return_value=_ok_response(False),
):
assert client.retrieve_exists(["service", "nonexistent"]) is False
class TestErrorHandling:
def test_api_failure_raises(self):
from ansible_collections.vyos.rest.plugins.module_utils.vyos_rest import (
VyOSRestError,
)
client = _make_client()
err_response = MagicMock()
err_response.read.return_value = json.dumps(
{"success": False, "data": None, "error": "path not found"},
).encode()
with patch(
"ansible_collections.vyos.rest.plugins.module_utils.vyos_rest.open_url",
return_value=err_response,
):
with pytest.raises(VyOSRestError, match="path not found"):
client.configure_set(["bad", "path"])
def test_invalid_json_raises(self):
from ansible_collections.vyos.rest.plugins.module_utils.vyos_rest import (
VyOSRestError,
)
client = _make_client()
bad_response = MagicMock()
bad_response.read.return_value = b"not json"
with patch(
"ansible_collections.vyos.rest.plugins.module_utils.vyos_rest.open_url",
return_value=bad_response,
):
with pytest.raises(VyOSRestError, match="Invalid JSON"):
client.retrieve_show_config([])
|
