# -*- coding: utf-8 -*- from __future__ import absolute_import, division, print_function __metaclass__ = type import unittest from ansible_collections.vyos.rest.plugins.module_utils.vyos import dict_op from ansible_collections.vyos.rest.plugins.modules.vyos_ha import ( _device_to_argspec, _from_device_generic, _normalize_groups, _normalize_have, _to_device_generic, _want_to_device, ) from .base import load_fixture _BASE = ["high-availability"] class TestToDeviceGeneric(unittest.TestCase): """Test generic argspec->device converter — no field-name knowledge.""" def test_snake_to_kebab(self): result = _to_device_generic({"startup_delay": 30}) self.assertIn("startup-delay", result) self.assertEqual(result["startup-delay"], "30") def test_bool_true_becomes_presence_node(self): result = _to_device_generic({"disable": True}) self.assertEqual(result["disable"], {}) def test_bool_false_skipped(self): result = _to_device_generic({"no_preempt": False}) self.assertNotIn("no-preempt", result) def test_none_skipped(self): result = _to_device_generic({"address": None, "vrid": 20}) self.assertNotIn("address", result) self.assertEqual(result["vrid"], "20") def test_nested_dict_recurses(self): result = _to_device_generic({"garp": {"master_repeat": 6}}) self.assertEqual(result["garp"]["master-repeat"], "6") def test_list_in_tag_node_fields(self): result = _to_device_generic( {"address": ["192.168.1.1/24", "192.168.1.2/24"]}, tag_node_fields={"address"}, ) self.assertIn("192.168.1.1/24", result["address"]) self.assertIn("192.168.1.2/24", result["address"]) self.assertEqual(result["address"]["192.168.1.1/24"], {}) def test_list_not_in_tag_node_fields_skipped(self): result = _to_device_generic({"address": ["192.168.1.1/24"]}) self.assertNotIn("address", result) def test_empty_dict(self): self.assertEqual(_to_device_generic({}), {}) self.assertEqual(_to_device_generic(None), {}) class TestFromDeviceGeneric(unittest.TestCase): """Test generic device->argspec converter — no field-name knowledge.""" def test_kebab_to_snake(self): result = _from_device_generic({"startup-delay": "30"}) self.assertIn("startup_delay", result) self.assertEqual(result["startup_delay"], 30) def test_empty_dict_becomes_true(self): result = _from_device_generic({"disable": {}, "no-preempt": {}}) self.assertTrue(result["disable"]) self.assertTrue(result["no_preempt"]) def test_numeric_string_cast_to_int(self): result = _from_device_generic({"vrid": "20", "priority": "100"}) self.assertEqual(result["vrid"], 20) self.assertEqual(result["priority"], 100) def test_non_numeric_string_unchanged(self): result = _from_device_generic({"interface": "eth0", "peer-address": "192.168.1.1"}) self.assertEqual(result["interface"], "eth0") self.assertEqual(result["peer_address"], "192.168.1.1") def test_nested_dict_recurses(self): result = _from_device_generic({"garp": {"master-repeat": "6"}}) self.assertEqual(result["garp"]["master_repeat"], 6) def test_empty(self): self.assertEqual(_from_device_generic({}), {}) self.assertEqual(_from_device_generic(None), {}) class TestWantToDevice(unittest.TestCase): """Test argspec->device structural reshaping.""" def test_empty(self): self.assertEqual(_want_to_device({}), {}) self.assertEqual(_want_to_device(None), {}) def test_disable(self): result = _want_to_device({"disable": True}) self.assertEqual(result["disable"], {}) def test_virtual_server_keyed_by_name(self): config = {"virtual_servers": [{"name": "s1", "address": "10.0.0.1", "port": 80}]} result = _want_to_device(config) self.assertIn("s1", result["virtual-server"]) self.assertEqual(result["virtual-server"]["s1"]["address"], "10.0.0.1") self.assertEqual(result["virtual-server"]["s1"]["port"], "80") def test_real_server_keyed_by_address(self): config = { "virtual_servers": [ { "name": "s1", "port": 80, "protocol": "tcp", "real_server": [{"address": "10.0.0.2", "port": 8080}], }, ], } result = _want_to_device(config) rs = result["virtual-server"]["s1"]["real-server"] self.assertIn("10.0.0.2", rs) self.assertEqual(rs["10.0.0.2"]["port"], "8080") def test_health_check_script_nested(self): config = { "virtual_servers": [ { "name": "s1", "port": 80, "protocol": "tcp", "real_server": [ { "address": "10.0.0.2", "health_check_script": "/check.sh", }, ], }, ], } result = _want_to_device(config) rs = result["virtual-server"]["s1"]["real-server"]["10.0.0.2"] self.assertEqual(rs["health-check"]["script"], "/check.sh") def test_vrrp_global_parameters(self): config = { "vrrp": { "global_parameters": { "startup_delay": 30, "garp": {"master_repeat": 6}, }, }, } result = _want_to_device(config) gp = result["vrrp"]["global-parameters"] self.assertEqual(gp["startup-delay"], "30") self.assertEqual(gp["garp"]["master-repeat"], "6") def test_snmp_enabled_becomes_presence_node(self): config = {"vrrp": {"snmp": "enabled"}} result = _want_to_device(config) self.assertEqual(result["vrrp"]["snmp"], {}) def test_snmp_disabled_not_in_want(self): config = {"vrrp": {"snmp": "disabled"}} result = _want_to_device(config) self.assertNotIn("snmp", result.get("vrrp", {})) def test_group_keyed_by_name(self): config = { "vrrp": { "groups": [ { "name": "g1", "vrid": 20, "interface": "eth0", "priority": 100, }, ], }, } result = _want_to_device(config) g = result["vrrp"]["group"]["g1"] self.assertEqual(g["vrid"], "20") self.assertEqual(g["interface"], "eth0") def test_group_address_tag_node(self): config = { "vrrp": { "groups": [ { "name": "g1", "vrid": 1, "address": ["192.168.1.1/24", "192.168.1.2/24"], }, ], }, } result = _want_to_device(config) addr = result["vrrp"]["group"]["g1"]["address"] self.assertIn("192.168.1.1/24", addr) self.assertEqual(addr["192.168.1.1/24"], {}) def test_group_bool_fields(self): config = { "vrrp": { "groups": [ { "name": "g1", "vrid": 1, "disable": True, "no_preempt": True, "rfc3768_compatibility": False, }, ], }, } result = _want_to_device(config) g = result["vrrp"]["group"]["g1"] self.assertEqual(g["disable"], {}) self.assertEqual(g["no-preempt"], {}) self.assertNotIn("rfc3768-compatibility", g) def test_group_track_interface_tag_node(self): config = { "vrrp": { "groups": [ { "name": "g1", "vrid": 1, "track": {"interface": ["eth0", "eth1"]}, }, ], }, } result = _want_to_device(config) track = result["vrrp"]["group"]["g1"]["track"] self.assertIn("eth0", track["interface"]) self.assertIn("eth1", track["interface"]) def test_sync_group_member_tag_node(self): config = {"vrrp": {"sync_groups": [{"name": "sg1", "member": ["g1", "g2"]}]}} result = _want_to_device(config) member = result["vrrp"]["sync-group"]["sg1"]["member"] self.assertIn("g1", member) self.assertIn("g2", member) class TestDeviceToArgspec(unittest.TestCase): """Test device->argspec structural reshaping.""" def test_empty(self): self.assertEqual(_device_to_argspec({}), {}) self.assertEqual(_device_to_argspec(None), {}) def test_disable_presence_node(self): result = _device_to_argspec({"disable": {}}) self.assertTrue(result["disable"]) def test_virtual_server_list(self): raw = {"virtual-server": {"s1": {"address": "10.0.0.1", "port": "80"}}} result = _device_to_argspec(raw) self.assertEqual(len(result["virtual_servers"]), 1) self.assertEqual(result["virtual_servers"][0]["name"], "s1") self.assertEqual(result["virtual_servers"][0]["port"], 80) def test_real_server_list(self): raw = { "virtual-server": { "s1": { "port": "80", "real-server": {"10.0.0.2": {"port": "8080"}}, }, }, } result = _device_to_argspec(raw) rs = result["virtual_servers"][0]["real_server"] self.assertEqual(rs[0]["address"], "10.0.0.2") self.assertEqual(rs[0]["port"], 8080) def test_health_check_script_extracted(self): raw = { "virtual-server": { "s1": { "port": "80", "real-server": {"10.0.0.2": {"health-check": {"script": "/check.sh"}}}, }, }, } result = _device_to_argspec(raw) rs = result["virtual_servers"][0]["real_server"][0] self.assertEqual(rs["health_check_script"], "/check.sh") def test_snmp_presence_node_becomes_enabled(self): raw = {"vrrp": {"snmp": {}}} result = _device_to_argspec(raw) self.assertEqual(result["vrrp"]["snmp"], "enabled") def test_global_parameters_converted(self): raw = { "vrrp": { "global-parameters": { "startup-delay": "30", "garp": {"master-repeat": "6"}, }, }, } result = _device_to_argspec(raw) gp = result["vrrp"]["global_parameters"] self.assertEqual(gp["startup_delay"], 30) self.assertEqual(gp["garp"]["master_repeat"], 6) def test_group_list_sorted(self): raw = {"vrrp": {"group": {"z1": {"vrid": "2"}, "a1": {"vrid": "1"}}}} result = _device_to_argspec(raw) names = [g["name"] for g in result["vrrp"]["groups"]] self.assertEqual(names[0], "a1") self.assertEqual(names[1], "z1") def test_group_address_tag_node_to_list(self): raw = {"vrrp": {"group": {"g1": {"address": {"192.168.1.1/24": {}}}}}} result = _device_to_argspec(raw) self.assertEqual(result["vrrp"]["groups"][0]["address"], ["192.168.1.1/24"]) def test_group_single_address_string_to_list(self): raw = {"vrrp": {"group": {"g1": {"address": "192.168.1.1/24"}}}} result = _device_to_argspec(raw) self.assertEqual(result["vrrp"]["groups"][0]["address"], ["192.168.1.1/24"]) def test_group_bool_presence_nodes(self): raw = {"vrrp": {"group": {"g1": {"disable": {}, "no-preempt": {}}}}} result = _device_to_argspec(raw) g = result["vrrp"]["groups"][0] self.assertTrue(g["disable"]) self.assertTrue(g["no_preempt"]) def test_sync_group_member_string_to_list(self): raw = {"vrrp": {"sync-group": {"sg1": {"member": "g1"}}}} result = _device_to_argspec(raw) self.assertEqual(result["vrrp"]["sync_groups"][0]["member"], ["g1"]) def test_sync_group_member_dict_to_list(self): raw = {"vrrp": {"sync-group": {"sg1": {"member": {"g1": {}, "g2": {}}}}}} result = _device_to_argspec(raw) self.assertEqual(sorted(result["vrrp"]["sync_groups"][0]["member"]), ["g1", "g2"]) class TestFixture(unittest.TestCase): """Test against real fixture.""" def setUp(self): self.fixture = load_fixture("ha_running.json") def test_disable_parsed(self): result = _device_to_argspec(self.fixture) self.assertTrue(result["disable"]) def test_virtual_server_parsed(self): result = _device_to_argspec(self.fixture) vs = result["virtual_servers"][0] self.assertEqual(vs["name"], "s1") self.assertEqual(vs["address"], "10.10.10.5") self.assertEqual(vs["port"], 80) self.assertEqual(vs["real_server"][0]["address"], "10.10.50.2") self.assertEqual(vs["real_server"][0]["port"], 8443) def test_global_parameters_parsed(self): result = _device_to_argspec(self.fixture) gp = result["vrrp"]["global_parameters"] self.assertEqual(gp["startup_delay"], 30) self.assertEqual(gp["garp"]["master_repeat"], 6) def test_snmp_parsed(self): result = _device_to_argspec(self.fixture) self.assertEqual(result["vrrp"]["snmp"], "enabled") def test_groups_parsed(self): result = _device_to_argspec(self.fixture) groups = {g["name"]: g for g in result["vrrp"]["groups"]} self.assertIn("g1", groups) self.assertIn("g2", groups) self.assertEqual(groups["g1"]["interface"], "eth0") self.assertEqual(groups["g1"]["vrid"], 20) self.assertIn("192.168.1.100/24", groups["g1"]["address"]) self.assertTrue(groups["g1"]["no_preempt"]) # g2 single address string -> list self.assertEqual(groups["g2"]["address"], ["192.168.2.100/24"]) def test_sync_group_parsed(self): result = _device_to_argspec(self.fixture) sg = result["vrrp"]["sync_groups"][0] self.assertEqual(sg["name"], "sg1") self.assertEqual(sg["member"], ["g1"]) self.assertEqual(sg["health_check"]["failure_count"], 5) self.assertEqual(sg["health_check"]["ping"], "192.168.1.1") class TestDictOpHa(unittest.TestCase): """Test dict_op with HA device shapes.""" def test_merged_adds_vrrp_group(self): want = _want_to_device( { "vrrp": { "groups": [ {"name": "g1", "vrid": 20, "interface": "eth0", "priority": 100}, ], }, }, ) cmds = dict_op(want, {}, _BASE, op="set") paths = [c[1] for c in cmds] self.assertIn(_BASE + ["vrrp", "group", "g1", "vrid", "20"], paths) self.assertIn(_BASE + ["vrrp", "group", "g1", "interface", "eth0"], paths) def test_merged_idempotent(self): config = { "vrrp": { "groups": [ {"name": "g1", "vrid": 20, "interface": "eth0"}, ], }, } want = _want_to_device(config) norm_have = _normalize_have( { "vrrp": { "group": { "g1": {"vrid": "20", "interface": "eth0"}, }, }, }, ) cmds = dict_op(want, norm_have, _BASE, op="set") self.assertEqual(cmds, []) def test_purge_removes_extra_group(self): want = _want_to_device( { "vrrp": { "groups": [ {"name": "g1", "vrid": 20, "interface": "eth0"}, ], }, }, ) norm_have = _normalize_have( { "vrrp": { "group": { "g1": {"vrid": "20", "interface": "eth0"}, "g2": {"vrid": "21", "interface": "eth1"}, }, }, }, ) cmds = dict_op( want.get("vrrp", {}), norm_have.get("vrrp", {}), _BASE + ["vrrp"], op="purge", ) paths = [c[1] for c in cmds] self.assertIn(_BASE + ["vrrp", "group", "g2"], paths) self.assertNotIn(_BASE + ["vrrp", "group", "g1"], paths) def test_normalize_have_member_string(self): raw = {"vrrp": {"sync-group": {"sg1": {"member": "g1"}}}} norm = _normalize_have(raw) self.assertIsInstance(norm["vrrp"]["sync-group"]["sg1"]["member"], dict) self.assertIn("g1", norm["vrrp"]["sync-group"]["sg1"]["member"]) def test_normalize_groups_address_string(self): norm = {"vrrp": {"group": {"g1": {"address": "192.168.1.1/24"}}}} _normalize_groups(norm) self.assertIsInstance(norm["vrrp"]["group"]["g1"]["address"], dict) self.assertIn("192.168.1.1/24", norm["vrrp"]["group"]["g1"]["address"]) if __name__ == "__main__": unittest.main()