diff options
Diffstat (limited to 'tests/unit/modules/test_vyos_ntp_global.py')
| -rw-r--r-- | tests/unit/modules/test_vyos_ntp_global.py | 265 |
1 files changed, 132 insertions, 133 deletions
diff --git a/tests/unit/modules/test_vyos_ntp_global.py b/tests/unit/modules/test_vyos_ntp_global.py index c536141..4bbeed8 100644 --- a/tests/unit/modules/test_vyos_ntp_global.py +++ b/tests/unit/modules/test_vyos_ntp_global.py @@ -4,179 +4,178 @@ from __future__ import absolute_import, division, print_function __metaclass__ = type -import json -import os import unittest from unittest.mock import MagicMock from ansible_collections.vyos.rest.plugins.modules.vyos_ntp_global import ( + _device_to_argspec, + _servers_from_device, + _servers_to_device, + _want_to_device, build_commands, get_running_config, - normalize_config, - normalize_servers, ) +from .base import load_fixture -def load_fixture(filename): - fixtures_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures") - path = os.path.join(fixtures_dir, filename) - with open(path) as f: - return json.load(f) + +_BASE = ["service", "ntp"] class VyOSModuleTestCase(unittest.TestCase): def setUp(self): self.mock_vyos = MagicMock() - self.mock_vyos.get_config = MagicMock(return_value={}) - - def set_running_config(self, data): - self.mock_vyos.get_config.return_value = data - - -class TestVyOSNtpGlobalNormalize(unittest.TestCase): - """Test normalize_config and normalize_servers — no device needed.""" - - def test_normalize_config_empty(self): - result = normalize_config({}) - self.assertEqual(result["allow_clients"], []) - self.assertEqual(result["listen_addresses"], []) - self.assertEqual(result["servers"], {}) - - def test_normalize_config_servers_sorted(self): - config = { - "servers": [ - {"server": "b.example.com", "options": ["prefer", "noselect"]}, - {"server": "a.example.com"}, - ], - } - result = normalize_config(config) - self.assertIn("a.example.com", result["servers"]) - self.assertIn("b.example.com", result["servers"]) - self.assertEqual(result["servers"]["b.example.com"], ["noselect", "prefer"]) - - def test_normalize_servers_dict_with_options(self): - raw = { - "time1.vyos.net": {}, - "203.0.113.0": {"prefer": {}}, - } - result = normalize_servers(raw) - self.assertEqual(result["time1.vyos.net"], []) - self.assertEqual(result["203.0.113.0"], ["prefer"]) - - def test_normalize_servers_list(self): - raw = ["time1.vyos.net", "time2.vyos.net"] - result = normalize_servers(raw) - self.assertEqual(result["time1.vyos.net"], []) - - def test_normalize_servers_string(self): - result = normalize_servers("time1.vyos.net") - self.assertEqual(result["time1.vyos.net"], []) - - -class TestVyOSNtpGlobalGetRunning(VyOSModuleTestCase): - """Test get_running_config parsing against fixture API responses.""" - - def setUp(self): - super().setUp() self.fixture = load_fixture("ntp_global_running.json") + self.mock_vyos.get_config = MagicMock(return_value=self.fixture) - def test_parses_allow_clients(self): - self.set_running_config(self.fixture) - result = get_running_config(self.mock_vyos) - self.assertIn("10.6.6.0/24", result["allow_clients"]) - def test_parses_listen_addresses(self): - self.set_running_config(self.fixture) - result = get_running_config(self.mock_vyos) - self.assertIn("10.1.3.1", result["listen_addresses"]) +class TestGetRunningConfig(VyOSModuleTestCase): + def test_returns_raw_device_dict(self): + self.assertEqual(get_running_config(self.mock_vyos), self.fixture) - def test_parses_servers(self): - self.set_running_config(self.fixture) - result = get_running_config(self.mock_vyos) - self.assertIn("time1.vyos.net", result["servers"]) - self.assertIn("203.0.113.0", result["servers"]) - self.assertIn("prefer", result["servers"]["203.0.113.0"]) + def test_empty_config(self): + self.mock_vyos.get_config = MagicMock(return_value=None) + self.assertEqual(get_running_config(self.mock_vyos), {}) - def test_empty_config_returns_empty(self): - self.set_running_config({}) - result = get_running_config(self.mock_vyos) - self.assertEqual(result["allow_clients"], []) - self.assertEqual(result["servers"], {}) +class TestServersToDeviceFromDevice(unittest.TestCase): + """options is the one genuine structural exception: the argspec + wraps them in a named field, but the device puts each option as a + direct presence-leaf sibling under the server tag node itself.""" -class TestVyOSNtpGlobalBuildCommands(unittest.TestCase): - """Test build_commands diff logic — no device needed.""" + def test_to_device_bare_server_is_presence(self): + self.assertEqual(_servers_to_device([{"server": "time1.vyos.net"}]), {"time1.vyos.net": {}}) - def _have(self, **kwargs): - base = {"allow_clients": [], "listen_addresses": [], "servers": {}} - base.update(kwargs) - return base + def test_to_device_options_become_sibling_presence_leaves(self): + result = _servers_to_device([{"server": "203.0.113.0", "options": ["prefer", "nts"]}]) + self.assertEqual(result, {"203.0.113.0": {"prefer": {}, "nts": {}}}) - def _want(self, **kwargs): - return self._have(**kwargs) + def test_from_device_bare_server(self): + result = _servers_from_device({"time1.vyos.net": {}}) + self.assertEqual(result, [{"server": "time1.vyos.net"}]) - def test_merged_adds_new_server(self): - want = self._want(servers={"new.server.com": []}) - have = self._have(servers={}) - cmds = build_commands(want, have, "merged") - self.assertIn(("set", ["service", "ntp", "server", "new.server.com"]), cmds) + def test_from_device_options_extracted_as_sorted_list(self): + result = _servers_from_device({"203.0.113.0": {"prefer": {}, "nts": {}}}) + self.assertEqual(result, [{"server": "203.0.113.0", "options": ["nts", "prefer"]}]) + + +class TestWantToDevice(unittest.TestCase): + def test_empty(self): + self.assertEqual(_want_to_device({}), {}) + + def test_allow_clients_nested_under_address(self): + """allow_clients is a flat argspec list, but the device nests + the multi-value leaf one level deeper under a literal "address" + child -- confirmed against vyos-1x (allow-client.xml.i).""" + result = _want_to_device({"allow_clients": ["10.6.6.0/24"]}) + self.assertEqual(result, {"allow-client": {"address": ["10.6.6.0/24"]}}) - def test_merged_idempotent_existing_server(self): - want = self._want(servers={"time1.vyos.net": []}) - have = self._have(servers={"time1.vyos.net": []}) - cmds = build_commands(want, have, "merged") - self.assertEqual(cmds, []) + def test_listen_addresses_direct_no_nesting(self): + result = _want_to_device({"listen_addresses": ["10.1.3.1"]}) + self.assertEqual(result, {"listen-address": ["10.1.3.1"]}) + + def test_servers_keyed_by_address(self): + result = _want_to_device({"servers": [{"server": "203.0.113.0", "options": ["prefer"]}]}) + self.assertEqual(result, {"server": {"203.0.113.0": {"prefer": {}}}}) + + +class TestDeviceToArgspecFixture(VyOSModuleTestCase): + def test_allow_clients_parsed(self): + result = _device_to_argspec(self.fixture) + self.assertIn("10.6.6.0/24", result["allow_clients"]) + + def test_listen_addresses_parsed(self): + result = _device_to_argspec(self.fixture) + self.assertIn("10.1.3.1", result["listen_addresses"]) + + def test_servers_parsed_with_options(self): + result = _device_to_argspec(self.fixture) + servers = {s["server"]: s.get("options", []) for s in result["servers"]} + self.assertIn("time1.vyos.net", servers) + self.assertIn("prefer", servers["203.0.113.0"]) + + def test_empty_config(self): + result = _device_to_argspec({}) + self.assertEqual(result, {"allow_clients": [], "listen_addresses": [], "servers": []}) + + def test_1_5_plus_shape_no_address_wrapper(self): + """Confirmed against vyos-1x, but kept defensive: some REST + responses omit the "address" subnode under allow-client.""" + raw = {"allow-client": {"10.6.6.0/24": {}}} + result = _device_to_argspec(raw) + self.assertEqual(result["allow_clients"], ["10.6.6.0/24"]) + + +class TestBuildCommands(VyOSModuleTestCase): + def test_merged_idempotent_against_own_fixture(self): + have = _device_to_argspec(self.fixture) + self.assertEqual(build_commands(have, self.fixture, "merged"), []) + + def test_replaced_idempotent_against_own_fixture(self): + have = _device_to_argspec(self.fixture) + self.assertEqual(build_commands(have, self.fixture, "replaced"), []) + + def test_1_5_plus_shape_idempotent(self): + """Regression test for the real bug caught this session: want + always emits the "address"-wrapped shape, but dict_op compares + directly against the raw device tree -- without normalizing + have's shape first, a device reporting the unwrapped 1.5+ + variant would never be idempotent.""" + raw_have = {"allow-client": {"10.6.6.0/24": {}}} + have = _device_to_argspec(raw_have) + self.assertEqual(build_commands(have, raw_have, "merged"), []) + + def test_merged_adds_new_server(self): + cmds = build_commands({"servers": [{"server": "new.server.com"}]}, {}, "merged") + self.assertIn(("set", _BASE + ["server", "new.server.com"]), cmds) def test_merged_adds_server_option(self): - want = self._want(servers={"time1.vyos.net": ["prefer"]}) - have = self._have(servers={"time1.vyos.net": []}) - cmds = build_commands(want, have, "merged") - self.assertIn(("set", ["service", "ntp", "server", "time1.vyos.net", "prefer"]), cmds) + raw_have = {"server": {"time1.vyos.net": {}}} + config = {"servers": [{"server": "time1.vyos.net", "options": ["prefer"]}]} + cmds = build_commands(config, raw_have, "merged") + self.assertIn(("set", _BASE + ["server", "time1.vyos.net", "prefer"]), cmds) def test_replaced_removes_extra_server(self): - want = self._want(servers={"time1.vyos.net": []}) - have = self._have(servers={"time1.vyos.net": [], "time2.vyos.net": []}) - cmds = build_commands(want, have, "replaced") - self.assertIn(("delete", ["service", "ntp", "server", "time2.vyos.net"]), cmds) + raw_have = {"server": {"time1.vyos.net": {}, "time2.vyos.net": {}}} + config = {"servers": [{"server": "time1.vyos.net"}]} + cmds = build_commands(config, raw_have, "replaced") + self.assertIn(("delete", _BASE + ["server", "time2.vyos.net"]), cmds) def test_replaced_removes_extra_allow_client(self): - want = self._want(allow_clients=["10.1.0.0/24"]) - have = self._have(allow_clients=["10.1.0.0/24", "10.2.0.0/24"]) - cmds = build_commands(want, have, "replaced") - self.assertIn( - ("delete", ["service", "ntp", "allow-client", "address", "10.2.0.0/24"]), - cmds, - ) + """This exercises the real dict_op purge gap fixed this session: + have's allow-client returned as dict-of-presence (not a plain + list) while want is a plain list -- purge must still correctly + remove the stale entry.""" + raw_have = {"allow-client": {"address": {"10.1.0.0/24": {}, "10.2.0.0/24": {}}}} + config = {"allow_clients": ["10.1.0.0/24"]} + cmds = build_commands(config, raw_have, "replaced") + self.assertIn(("delete", _BASE + ["allow-client", "address", "10.2.0.0/24"]), cmds) def test_deleted_removes_all(self): - have = self._have( - servers={"time1.vyos.net": []}, - allow_clients=["10.0.0.0/24"], - listen_addresses=["192.168.1.1"], - ) - cmds = build_commands({}, have, "deleted") - self.assertEqual(len(cmds), 1) - self.assertEqual(cmds[0], ("delete", ["service", "ntp"])) + raw_have = {"server": {"time1.vyos.net": {}}} + cmds = build_commands({}, raw_have, "deleted") + self.assertEqual(cmds, [("delete", _BASE)]) def test_deleted_idempotent_when_empty(self): - have = self._have(servers={}, allow_clients=[], listen_addresses=[]) - cmds = build_commands({}, have, "deleted") - self.assertEqual(cmds, []) + self.assertEqual(build_commands({}, {}, "deleted"), []) def test_overridden_deletes_then_merges(self): - want = self._want(servers={"new.server.com": []}) - have = self._have(servers={"old.server.com": []}) - cmds = build_commands(want, have, "overridden") - ops_paths = [(c[0], c[1]) for c in cmds] - self.assertIn(("delete", ["service", "ntp", "server", "old.server.com"]), ops_paths) - self.assertIn(("set", ["service", "ntp", "server", "new.server.com"]), ops_paths) - self.assertNotIn(("delete", ["service", "ntp", "server"]), ops_paths) + raw_have = {"server": {"old.server.com": {}}} + config = {"servers": [{"server": "new.server.com"}]} + cmds = build_commands(config, raw_have, "overridden") + self.assertIn(("delete", _BASE + ["server", "old.server.com"]), cmds) + self.assertIn(("set", _BASE + ["server", "new.server.com"]), cmds) def test_no_commands_when_already_correct(self): - state = {"allow_clients": ["10.0.0.0/24"], "listen_addresses": [], "servers": {}} - cmds = build_commands(state, state, "merged") - self.assertEqual(cmds, []) + raw_have = {"allow-client": {"address": {"10.0.0.0/24": {}}}} + config = {"allow_clients": ["10.0.0.0/24"]} + self.assertEqual(build_commands(config, raw_have, "merged"), []) + + def test_collapsed_single_server_no_char_iteration_bug(self): + raw_have = {"server": "203.0.113.0"} + config = {"servers": [{"server": "203.0.113.0"}]} + self.assertEqual(build_commands(config, raw_have, "merged"), []) if __name__ == "__main__": |