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
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
|
# -*- 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_ospfv2 import (
_CONFIG_OPTIONS,
ARGUMENT_SPEC,
_area_type_from_device,
_area_type_to_device,
_derive_key_field,
_device_to_argspec,
_distance_from_device,
_distance_to_device,
_kebab_fields,
_passive_from_device,
_passive_to_device,
_timers_from_device,
_timers_to_device_refresh,
_timers_to_device_throttle,
_vlink_auth_from_device,
_vlink_auth_to_device,
build_commands,
cast_by_spec,
get_running_config,
)
from .base import load_fixture
_BASE = ["protocols", "ospf"]
class VyOSModuleTestCase(unittest.TestCase):
def setUp(self):
self.mock_vyos = MagicMock()
self.fixture = load_fixture("ospfv2_running.json")
self.mock_vyos.get_config = MagicMock(return_value=self.fixture)
def gather(self):
have = _device_to_argspec(self.fixture)
cast_by_spec(have, _CONFIG_OPTIONS)
return have
class TestGetRunningConfig(VyOSModuleTestCase):
def test_returns_config_directly(self):
result = get_running_config(self.mock_vyos)
self.assertIn("area", result)
def test_empty_config(self):
self.mock_vyos.get_config = MagicMock(return_value=None)
self.assertEqual(get_running_config(self.mock_vyos), {})
class TestDeriveKeyField(unittest.TestCase):
def test_derives_area_id(self):
area_opts = ARGUMENT_SPEC["config"]["options"]["areas"]["options"]
self.assertEqual(_derive_key_field(area_opts), "area_id")
def test_derives_neighbor_id(self):
nb_opts = ARGUMENT_SPEC["config"]["options"]["neighbor"]["options"]
self.assertEqual(_derive_key_field(nb_opts), "neighbor_id")
def test_raises_if_none_required(self):
with self.assertRaises(ValueError):
_derive_key_field({"a": {"type": "str"}})
class TestKebabFields(unittest.TestCase):
"""Regression tests for the primary confirmed bug found during
development: dict_op requires have's keys to already be genuine
device kebab-case (it only normalizes underscores for its own
lookup index, not for the output path). autoclean deliberately
leaves keys as given, so any multi-word field reconstructed via
this module's own entry-transforms (rather than coming straight
from the device) needs explicit conversion, or a generated delete
command uses the wrong (snake_case) path segment."""
def test_converts_multiword_keys(self):
result = _kebab_fields({"default_cost": 20, "no_summary": True})
self.assertEqual(result, {"default-cost": 20, "no-summary": {}})
def test_drops_none_and_false(self):
result = _kebab_fields({"default_cost": None, "no_summary": False})
self.assertEqual(result, {})
def test_single_word_keys_unaffected(self):
result = _kebab_fields({"cost": 10})
self.assertEqual(result, {"cost": 10})
class TestAreaType(unittest.TestCase):
def test_nssa_set_flag_is_node_presence_not_a_device_leaf(self):
"""Confirmed genuine structural exception: the argspec's
nssa.set/stub.set boolean doesn't exist as a device leaf --
the node's own presence IS the set flag."""
result = _area_type_to_device({"nssa": {"set": True, "default_cost": 5}})
self.assertEqual(result, {"nssa": {"default-cost": 5}})
self.assertNotIn("set", result["nssa"])
def test_stub_no_summary(self):
"""Regression test for the confirmed original bug: stub's
no_summary was declared in ARGUMENT_SPEC but never checked
anywhere in the original hand-rolled command-building logic."""
result = _area_type_to_device({"stub": {"no_summary": True}})
self.assertEqual(result, {"stub": {"no-summary": {}}})
def test_from_device_restores_set_flag(self):
entry = _area_type_from_device({"nssa": {"default-cost": "5"}})
self.assertTrue(entry["nssa"]["set"])
def test_normal_presence(self):
result = _area_type_to_device({"normal": True})
self.assertEqual(result, {"normal": {}})
def test_empty(self):
self.assertEqual(_area_type_to_device({}), {})
self.assertIsNone(_area_type_from_device({}))
class TestDistance(unittest.TestCase):
def test_global_and_ospf(self):
result = _distance_to_device({"global": 110, "ospf": {"inter_area": 120}})
self.assertEqual(result, {"global": 110, "ospf": {"inter-area": 120}})
def test_from_device(self):
entry = _distance_from_device({"global": "110", "ospf": {"external": "150"}})
self.assertEqual(entry["global"], 110)
self.assertEqual(entry["ospf"]["external"], "150") # cast_by_spec's job downstream
class TestTimers(unittest.TestCase):
"""Confirmed genuine structural exception: the argspec groups
"refresh" and "throttle" both under one "timers" parent, but the
device has them as two separate top-level nodes."""
def test_refresh_maps_to_separate_top_level_node(self):
result = _timers_to_device_refresh({"refresh": {"timers": 300}})
self.assertEqual(result, {"timers": 300})
def test_throttle_maps_to_device_timers_node(self):
result = _timers_to_device_throttle({"throttle": {"spf": {"delay": 200}}})
self.assertEqual(result, {"throttle": {"spf": {"delay": 200}}})
def test_from_device_recombines_both(self):
entry = _timers_from_device({"timers": "300"}, {"throttle": {"spf": {"delay": "200"}}})
self.assertEqual(entry["refresh"]["timers"], 300)
self.assertEqual(entry["throttle"]["spf"]["delay"], "200")
class TestPassiveInterface(unittest.TestCase):
"""Confirmed genuine structural exception: passive_interface and
passive_interface_exclude both map onto the same per-interface
"interface <name> passive" device subtree -- presence alone means
enabled, "passive.disable" means explicitly excluded."""
def test_to_device_both(self):
result = _passive_to_device(["eth1"], ["eth2"])
self.assertEqual(result, {"eth1": {"passive": {}}, "eth2": {"passive": {"disable": {}}}})
def test_from_device_both(self):
passive, excluded = _passive_from_device(
{"eth1": {"passive": {}}, "eth2": {"passive": {"disable": {}}}},
)
self.assertEqual(passive, ["eth1"])
self.assertEqual(excluded, ["eth2"])
def test_from_device_ignores_non_passive_interfaces(self):
passive, excluded = _passive_from_device({"eth3": {"some-other-key": {}}})
self.assertEqual(passive, [])
self.assertEqual(excluded, [])
class TestVirtualLinkAuth(unittest.TestCase):
def test_md5_to_device(self):
result = _vlink_auth_to_device({"md5": [{"key_id": 10, "md5_key": "secret"}]})
self.assertEqual(result, {"md5": {"10": {"md5-key": "secret"}}})
def test_plaintext_to_device(self):
result = _vlink_auth_to_device({"plaintext_password": "pw"})
self.assertEqual(result, {"plaintext-password": "pw"})
def test_md5_from_device(self):
entry = _vlink_auth_from_device({"md5": {"10": {"md5-key": "secret"}}})
self.assertEqual(entry["md5"], [{"key_id": 10, "md5_key": "secret"}])
class TestDeviceToArgspecFixture(VyOSModuleTestCase):
def test_areas_parsed(self):
have = self.gather()
area_ids = [a["area_id"] for a in have["areas"]]
self.assertEqual(set(area_ids), {"2", "4", "5"})
def test_stub_area_no_summary_parsed(self):
have = self.gather()
area4 = next(a for a in have["areas"] if a["area_id"] == "4")
self.assertTrue(area4["area_type"]["stub"]["no_summary"])
def test_virtual_link_parsed(self):
have = self.gather()
area4 = next(a for a in have["areas"] if a["area_id"] == "4")
vlink = area4["virtual_link"][0]
self.assertEqual(vlink["address"], "10.0.0.1")
self.assertEqual(vlink["authentication"]["plaintext_password"], "secret")
self.assertEqual(vlink["dead_interval"], 40)
def test_passive_interface_and_exclude_parsed(self):
have = self.gather()
self.assertEqual(have["passive_interface"], ["eth1"])
self.assertEqual(have["passive_interface_exclude"], ["eth2"])
def test_timers_parsed(self):
have = self.gather()
self.assertEqual(have["timers"]["refresh"]["timers"], 300)
self.assertEqual(have["timers"]["throttle"]["spf"]["delay"], 200)
def test_max_metric_parsed(self):
have = self.gather()
self.assertTrue(have["max_metric"]["router_lsa"]["administrative"])
self.assertEqual(have["max_metric"]["router_lsa"]["on_shutdown"], 10)
def test_mpls_te_parsed(self):
have = self.gather()
self.assertTrue(have["mpls_te"]["enabled"])
self.assertEqual(have["mpls_te"]["router_address"], "192.0.11.11")
def test_empty_config(self):
self.assertEqual(_device_to_argspec({}), {})
self.assertEqual(_device_to_argspec(None), {})
class TestBuildCommands(VyOSModuleTestCase):
def test_merged_idempotent_against_own_fixture(self):
have = self.gather()
self.assertEqual(build_commands(have, self.fixture, "merged"), [])
def test_replaced_idempotent_against_own_fixture(self):
have = self.gather()
self.assertEqual(build_commands(have, self.fixture, "replaced"), [])
def test_clear_omitted_attribute_on_replaced(self):
"""Primary confirmed bug from the original hand-rolled
implementation: clearing an omitted attribute never generated
a delete command. Also the exact scenario that caught the
kebab-key regression during development."""
raw_have = {"area": {"4": {"area-type": {"stub": {"default-cost": "20"}}}}}
config = {"areas": [{"area_id": "4", "area_type": {"stub": {"set": True}}}]}
cmds = build_commands(config, raw_have, "replaced")
expected = ("delete", _BASE + ["area", "4", "area-type", "stub", "default-cost"])
self.assertIn(expected, cmds)
def test_stub_no_summary_now_works(self):
config = {"areas": [{"area_id": "4", "area_type": {"stub": {"no_summary": True}}}]}
cmds = build_commands(config, {}, "merged")
expected = ("set", _BASE + ["area", "4", "area-type", "stub", "no-summary"])
self.assertIn(expected, cmds)
def test_replaced_only_touches_what_changed(self):
"""Confirmed fix for the original's disruptive "delete
everything and recreate" replaced heuristic -- a targeted
dict_op purge only touches the sections that actually
differ."""
raw_have = {
"area": {"2": {"area-type": {"normal": {}}}},
"parameters": {"router-id": "1.1.1.1"},
}
config = {
"areas": [{"area_id": "2", "area_type": {"normal": True}}],
"parameters": {"router_id": "2.2.2.2"},
}
cmds = build_commands(config, raw_have, "replaced")
self.assertFalse(any("area" in str(c) for c in cmds))
expected = ("set", _BASE + ["parameters", "router-id", "2.2.2.2"])
self.assertIn(expected, cmds)
def test_passive_interface_and_exclude_together(self):
config = {"passive_interface": ["eth1"], "passive_interface_exclude": ["eth2"]}
cmds = build_commands(config, {}, "merged")
self.assertIn(("set", _BASE + ["interface", "eth1", "passive"]), cmds)
self.assertIn(("set", _BASE + ["interface", "eth2", "passive", "disable"]), cmds)
def test_deleted_with_have(self):
cmds = build_commands({}, {"parameters": {"router-id": "1.1.1.1"}}, "deleted")
self.assertEqual(cmds, [("delete", _BASE)])
def test_deleted_no_have_is_noop(self):
self.assertEqual(build_commands({}, {}, "deleted"), [])
def test_merged_new_virtual_link(self):
config = {
"areas": [
{
"area_id": "4",
"virtual_link": [
{"address": "10.0.0.1", "authentication": {"plaintext_password": "pw"}},
],
},
],
}
cmds = build_commands(config, {}, "merged")
vlink_path = _BASE + [
"area",
"4",
"virtual-link",
"10.0.0.1",
"authentication",
"plaintext-password",
"pw",
]
self.assertIn(("set", vlink_path), cmds)
if __name__ == "__main__":
unittest.main()
|
