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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
|
# -*- coding: utf-8 -*-
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import unittest
from unittest.mock import MagicMock
from ansible_collections.vyos.rest.plugins.modules.vyos_ospf_interfaces import (
_AF_OPTIONS,
ARGUMENT_SPEC,
_af4_entry_from_device,
_af4_entry_to_device,
_af6_entry_from_device,
_af6_entry_to_device,
_auth_from_device,
_auth_to_device,
_derive_key_field,
_device_to_argspec,
_kebab_fields,
build_commands,
cast_by_spec,
get_running_config,
)
from .base import load_fixture
_BASE4 = ["protocols", "ospf", "interface"]
_BASE6 = ["protocols", "ospfv3", "interface"]
class VyOSModuleTestCase(unittest.TestCase):
def setUp(self):
self.mock_vyos = MagicMock()
self.ipv4_fixture = load_fixture("ospf_interfaces_ipv4.json")
self.ipv6_fixture = load_fixture("ospf_interfaces_ipv6.json")
self.mock_vyos.get_config = MagicMock(
side_effect=lambda path: (self.ipv4_fixture if path == _BASE4 else self.ipv6_fixture),
)
def gather(self):
raw4, raw6 = get_running_config(self.mock_vyos)
have = _device_to_argspec(raw4, raw6)
for iface in have:
for af in iface.get("address_family") or []:
cast_by_spec(af, _AF_OPTIONS)
return have
class TestGetRunningConfig(VyOSModuleTestCase):
def test_returns_both_afi_trees_separately(self):
raw4, raw6 = get_running_config(self.mock_vyos)
self.assertIn("eth1", raw4)
self.assertIn("eth1", raw6)
def test_empty_config(self):
self.mock_vyos.get_config = MagicMock(return_value=None)
raw4, raw6 = get_running_config(self.mock_vyos)
self.assertEqual(raw4, {})
self.assertEqual(raw6, {})
class TestDeriveKeyField(unittest.TestCase):
def test_derives_name(self):
opts = ARGUMENT_SPEC["config"]["options"]
self.assertEqual(_derive_key_field(opts), "name")
def test_derives_afi(self):
af_opts = ARGUMENT_SPEC["config"]["options"]["address_family"]["options"]
self.assertEqual(_derive_key_field(af_opts), "afi")
class TestKebabFields(unittest.TestCase):
"""Regression coverage for the confirmed bug from vyos_ospfv2's
build, applied here from the start: dict_op requires have's keys
to already be genuine device kebab-case, but autoclean deliberately
leaves keys as given."""
def test_converts_multiword_keys(self):
result = _kebab_fields({"dead_interval": 40, "mtu_ignore": True})
self.assertEqual(result, {"dead-interval": 40, "mtu-ignore": {}})
class TestAuthentication(unittest.TestCase):
"""IPv4-only. Confirmed against vyos-1x: md5 is a fixed "key-id"
node containing a single tagNode entry, not a list."""
def test_plaintext_to_device(self):
result = _auth_to_device({"plaintext_password": "pw"})
self.assertEqual(result, {"plaintext-password": "pw"})
def test_md5_to_device(self):
result = _auth_to_device({"md5_key": {"key_id": 10, "key": "secret"}})
self.assertEqual(result, {"md5": {"key-id": {"10": {"md5-key": "secret"}}}})
def test_md5_from_device(self):
entry = _auth_from_device({"md5": {"key-id": {"10": {"md5-key": "secret"}}}})
self.assertEqual(entry["md5_key"], {"key_id": 10, "key": "secret"})
def test_empty(self):
self.assertEqual(_auth_to_device({}), {})
self.assertIsNone(_auth_from_device({}))
class TestAf6InstanceRename(unittest.TestCase):
"""Confirmed genuine rename: the argspec's "instance" (matching
the CLI collection's naming) maps to the device's actual leaf
name "instance-id"."""
def test_to_device(self):
result = _af6_entry_to_device({"instance": "5"})
self.assertEqual(result, {"instance-id": "5"})
def test_from_device(self):
entry = _af6_entry_from_device({"instance-id": "5"})
self.assertEqual(entry["instance"], "5")
class TestAf4EntryToDeviceFromDevice(unittest.TestCase):
def test_generic_fields(self):
result = _af4_entry_to_device({"cost": 100, "priority": 26})
self.assertEqual(result, {"cost": 100, "priority": 26})
def test_mtu_ignore_presence(self):
result = _af4_entry_to_device({"mtu_ignore": True})
self.assertEqual(result, {"mtu-ignore": {}})
def test_from_device_with_authentication(self):
entry = _af4_entry_from_device(
{"cost": "100", "authentication": {"plaintext-password": "pw"}},
)
self.assertEqual(entry["authentication"]["plaintext_password"], "pw")
class TestDeviceToArgspecFixture(VyOSModuleTestCase):
def test_both_interfaces_present(self):
have = self.gather()
names = [i["name"] for i in have]
self.assertIn("eth1", names)
self.assertIn("eth2", names)
def test_eth1_has_both_afis(self):
have = self.gather()
eth1 = next(i for i in have if i["name"] == "eth1")
afis = {af["afi"] for af in eth1["address_family"]}
self.assertEqual(afis, {"ipv4", "ipv6"})
def test_eth2_ipv4_only_minimal(self):
"""eth2's ipv6 fixture entry is empty ({}) -- confirms an
interface with config on only one AFI doesn't spuriously
appear on the other."""
have = self.gather()
eth2 = next(i for i in have if i["name"] == "eth2")
afis = {af["afi"] for af in eth2["address_family"]}
self.assertEqual(afis, {"ipv4"})
def test_ipv4_fields_parsed(self):
have = self.gather()
eth1 = next(i for i in have if i["name"] == "eth1")
af4 = next(af for af in eth1["address_family"] if af["afi"] == "ipv4")
self.assertEqual(af4["cost"], 100)
self.assertTrue(af4["mtu_ignore"])
self.assertEqual(af4["authentication"]["md5_key"]["key_id"], 10)
def test_ipv6_fields_parsed(self):
have = self.gather()
eth1 = next(i for i in have if i["name"] == "eth1")
af6 = next(af for af in eth1["address_family"] if af["afi"] == "ipv6")
self.assertTrue(af6["passive"])
self.assertEqual(af6["instance"], "5")
self.assertEqual(af6["ifmtu"], 1500)
def test_empty(self):
self.assertEqual(_device_to_argspec({}, {}), [])
class TestBuildCommands(VyOSModuleTestCase):
def test_merged_idempotent_against_own_fixture(self):
have = self.gather()
raw4, raw6 = get_running_config(self.mock_vyos)
self.assertEqual(build_commands(have, (raw4, raw6), "merged"), [])
def test_replaced_idempotent_against_own_fixture(self):
have = self.gather()
raw4, raw6 = get_running_config(self.mock_vyos)
self.assertEqual(build_commands(have, (raw4, raw6), "replaced"), [])
def test_clear_omitted_attribute_on_replaced(self):
"""Primary confirmed bug: the original's clear-on-omit logic
only fired for op="replace", which was never actually passed
anywhere in build_commands -- genuinely dead code, since
"replaced" already deleted and recreated the whole AFI block
before calling it with an empty have."""
raw4 = {"eth1": {"cost": "100", "priority": "26"}}
config = [{"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 100}]}]
cmds = build_commands(config, (raw4, {}), "replaced")
self.assertIn(("delete", _BASE4 + ["eth1", "priority"]), cmds)
def test_replaced_does_not_touch_unrelated_interfaces(self):
"""Confirmed fix for the original's disruptive "delete
everything and recreate" replaced heuristic."""
raw4 = {"eth1": {"cost": "100"}, "eth2": {"cost": "200"}}
config = [{"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 100}]}]
cmds = build_commands(config, (raw4, {}), "replaced")
self.assertEqual(cmds, [])
def test_ipv4_and_ipv6_target_separate_device_trees(self):
config = [
{
"name": "eth1",
"address_family": [
{"afi": "ipv4", "cost": 100},
{"afi": "ipv6", "passive": True},
],
},
]
cmds = build_commands(config, ({}, {}), "merged")
self.assertIn(("set", _BASE4 + ["eth1", "cost", "100"]), cmds)
self.assertIn(("set", _BASE6 + ["eth1", "passive"]), cmds)
def test_overridden_removes_omitted_interface(self):
raw4 = {"eth1": {"cost": "100"}, "eth3": {"cost": "50"}}
config = [{"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 100}]}]
cmds = build_commands(config, (raw4, {}), "overridden")
self.assertIn(("delete", _BASE4 + ["eth3"]), cmds)
def test_deleted_all_covers_both_trees(self):
cmds = build_commands([], ({"eth1": {}}, {"eth1": {}}), "deleted")
self.assertIn(("delete", _BASE4 + ["eth1"]), cmds)
self.assertIn(("delete", _BASE6 + ["eth1"]), cmds)
self.assertEqual(len(cmds), 2)
def test_deleted_named_no_afi_removes_both(self):
cmds = build_commands([{"name": "eth1"}], ({"eth1": {}}, {"eth1": {}}), "deleted")
self.assertIn(("delete", _BASE4 + ["eth1"]), cmds)
self.assertIn(("delete", _BASE6 + ["eth1"]), cmds)
self.assertEqual(len(cmds), 2)
def test_deleted_named_specific_afi_only(self):
config = [{"name": "eth1", "address_family": [{"afi": "ipv4"}]}]
cmds = build_commands(config, ({"eth1": {}}, {"eth1": {}}), "deleted")
self.assertEqual(cmds, [("delete", _BASE4 + ["eth1"])])
def test_deleted_named_nonexistent_is_noop(self):
cmds = build_commands([{"name": "eth99"}], ({"eth1": {}}, {}), "deleted")
self.assertEqual(cmds, [])
def test_merged_adds_cost(self):
config = [{"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 100}]}]
cmds = build_commands(config, ({}, {}), "merged")
self.assertIn(("set", _BASE4 + ["eth1", "cost", "100"]), cmds)
if __name__ == "__main__":
unittest.main()
|
