diff options
Diffstat (limited to 'tests/unit/modules/test_vyos_firewall_global.py')
| -rw-r--r-- | tests/unit/modules/test_vyos_firewall_global.py | 380 |
1 files changed, 272 insertions, 108 deletions
diff --git a/tests/unit/modules/test_vyos_firewall_global.py b/tests/unit/modules/test_vyos_firewall_global.py index a507c0c..e3aff0b 100644 --- a/tests/unit/modules/test_vyos_firewall_global.py +++ b/tests/unit/modules/test_vyos_firewall_global.py @@ -4,38 +4,165 @@ 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.module_utils.vyos import dict_op from ansible_collections.vyos.rest.plugins.modules.vyos_firewall_global import ( + _device_to_argspec, + _groups_from_device, + _groups_to_device, + _want_to_device, build_commands, get_running_config, ) +from .base import load_fixture + _BASE = ["firewall", "group"] -def load_fixture(filename): - fixtures_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures") - with open(os.path.join(fixtures_dir, filename)) as f: - return json.load(f) +class TestGroupHelpers(unittest.TestCase): + """Test group list <-> device dict conversion helpers.""" + def test_groups_to_device_with_members(self): + groups = [{"name": "SERVERS", "address": ["192.168.1.10", "192.168.1.11"]}] + result = _groups_to_device(groups, "address") + self.assertIn("SERVERS", result) + self.assertIn("192.168.1.10", result["SERVERS"]["address"]) + self.assertIn("192.168.1.11", result["SERVERS"]["address"]) -class VyOSModuleTestCase(unittest.TestCase): - def setUp(self): - self.mock_vyos = MagicMock() - self.fixture = load_fixture("firewall_global_running.json") - self.mock_vyos.get_config = MagicMock(return_value=self.fixture) + def test_groups_to_device_with_description(self): + groups = [{"name": "LAN", "description": "Local network", "network": ["192.168.0.0/16"]}] + result = _groups_to_device(groups, "network") + self.assertEqual(result["LAN"]["description"], "Local network") + + def test_groups_to_device_empty(self): + self.assertEqual(_groups_to_device([], "address"), {}) + self.assertEqual(_groups_to_device(None, "address"), {}) + + def test_groups_from_device_with_members(self): + raw = {"SERVERS": {"address": {"192.168.1.10": {}, "192.168.1.11": {}}}} + result = _groups_from_device(raw, "address") + self.assertEqual(len(result), 1) + self.assertEqual(result[0]["name"], "SERVERS") + self.assertIn("192.168.1.10", result[0]["address"]) + + def test_groups_from_device_single_member_string(self): + # VyOS returns single member as string + raw = {"WEB": {"port": "80"}} + result = _groups_from_device(raw, "port") + self.assertIn("80", result[0]["port"]) + + def test_groups_from_device_with_description(self): + raw = {"LAN": {"description": "Local", "network": {"192.168.0.0/16": {}}}} + result = _groups_from_device(raw, "network") + self.assertEqual(result[0]["description"], "Local") + + def test_groups_from_device_sorted(self): + raw = {"Z-GROUP": {}, "A-GROUP": {}} + result = _groups_from_device(raw, "address") + self.assertEqual(result[0]["name"], "A-GROUP") + self.assertEqual(result[1]["name"], "Z-GROUP") + + def test_groups_from_device_empty(self): + self.assertEqual(_groups_from_device({}, "address"), []) + self.assertEqual(_groups_from_device(None, "address"), []) + + +class TestWantToDevice(unittest.TestCase): + """Test argspec -> device shape conversion.""" + + def test_empty(self): + self.assertEqual(_want_to_device({}), {}) + self.assertEqual(_want_to_device(None), {}) + + def test_address_group(self): + config = { + "group": { + "address_group": [ + {"name": "SERVERS", "address": ["192.168.1.10"]}, + ], + }, + } + result = _want_to_device(config) + self.assertIn("address-group", result) + self.assertIn("SERVERS", result["address-group"]) + self.assertIn("192.168.1.10", result["address-group"]["SERVERS"]["address"]) + + def test_network_group(self): + config = { + "group": { + "network_group": [{"name": "LAN", "network": ["192.168.0.0/16"]}], + }, + } + result = _want_to_device(config) + self.assertIn("network-group", result) + self.assertIn("LAN", result["network-group"]) + + def test_port_group(self): + config = { + "group": { + "port_group": [{"name": "WEB", "port": ["80", "443"]}], + }, + } + result = _want_to_device(config) + self.assertIn("port-group", result) + self.assertIn("80", result["port-group"]["WEB"]["port"]) + + def test_interface_group(self): + config = { + "group": { + "interface_group": [{"name": "LAN-IFACES", "interface": ["eth1"]}], + }, + } + result = _want_to_device(config) + self.assertIn("interface-group", result) + + def test_ipv6_network_group(self): + config = { + "group": { + "ipv6_network_group": [{"name": "IPV6-LAN", "network": ["2001:db8::/32"]}], + }, + } + result = _want_to_device(config) + self.assertIn("ipv6-network-group", result) -class TestVyOSFirewallGlobalGetRunning(VyOSModuleTestCase): +class TestDeviceToArgspec(unittest.TestCase): + """Test device response -> argspec shape conversion.""" + + def test_empty(self): + self.assertEqual(_device_to_argspec({}), {}) + self.assertEqual(_device_to_argspec(None), {}) + + def test_address_group(self): + raw = { + "address-group": { + "SERVERS": { + "description": "Web servers", + "address": {"192.168.1.10": {}, "192.168.1.11": {}}, + }, + }, + } + result = _device_to_argspec(raw) + groups = result["group"]["address_group"] + servers = next(g for g in groups if g["name"] == "SERVERS") + self.assertEqual(servers["description"], "Web servers") + self.assertIn("192.168.1.10", servers["address"]) + + def test_no_group_returns_empty(self): + self.assertEqual(_device_to_argspec({}), {}) + + +class TestDeviceToArgspecFixture(unittest.TestCase): + """Test _device_to_argspec against fixture.""" + + def setUp(self): + self.fixture = load_fixture("firewall_global_running.json") def test_parses_address_groups(self): - result = get_running_config(self.mock_vyos) + result = _device_to_argspec(self.fixture) groups = result["group"]["address_group"] names = [g["name"] for g in groups] self.assertIn("SERVERS", names) @@ -43,148 +170,185 @@ class TestVyOSFirewallGlobalGetRunning(VyOSModuleTestCase): servers = next(g for g in groups if g["name"] == "SERVERS") self.assertEqual(servers["description"], "Web servers") self.assertIn("192.168.1.10", servers["address"]) - self.assertIn("192.168.1.11", servers["address"]) def test_parses_network_groups(self): - result = get_running_config(self.mock_vyos) + result = _device_to_argspec(self.fixture) groups = result["group"]["network_group"] dmz = next(g for g in groups if g["name"] == "DMZ") self.assertIn("10.0.0.0/8", dmz["network"]) - self.assertIn("172.16.0.0/12", dmz["network"]) def test_parses_port_groups(self): - result = get_running_config(self.mock_vyos) + result = _device_to_argspec(self.fixture) groups = result["group"]["port_group"] web = next(g for g in groups if g["name"] == "WEB-PORTS") self.assertIn("80", web["port"]) - self.assertIn("443", web["port"]) def test_parses_interface_groups(self): - result = get_running_config(self.mock_vyos) + result = _device_to_argspec(self.fixture) groups = result["group"]["interface_group"] lan = next(g for g in groups if g["name"] == "LAN-IFACES") self.assertIn("eth1", lan["interface"]) def test_parses_ipv6_network_groups(self): - result = get_running_config(self.mock_vyos) + result = _device_to_argspec(self.fixture) groups = result["group"]["ipv6_network_group"] ipv6 = next(g for g in groups if g["name"] == "IPV6-LAN") self.assertIn("2001:db8::/32", ipv6["network"]) def test_empty_config(self): - self.mock_vyos.get_config = MagicMock(return_value={}) - result = get_running_config(self.mock_vyos) + result = _device_to_argspec({}) self.assertEqual(result, {}) -class TestVyOSFirewallGlobalBuildCommands(unittest.TestCase): +class TestDictOpFirewall(unittest.TestCase): + """Test dict_op behaviour with firewall group shapes.""" - def _have(self): - return { - "group": { - "address_group": [ - {"name": "SERVERS", "address": ["192.168.1.10", "192.168.1.11"]}, - ], - "network_group": [ - {"name": "LAN", "network": ["192.168.0.0/16"]}, - ], + def test_merged_adds_address_group(self): + want = _want_to_device( + { + "group": { + "address_group": [{"name": "SERVERS", "address": ["192.168.1.10"]}], + }, }, - } - - def test_deleted_with_have(self): - cmds = build_commands({}, self._have(), "deleted") - self.assertEqual(cmds, [("delete", _BASE)]) - - def test_deleted_without_have(self): - cmds = build_commands({}, {}, "deleted") - self.assertEqual(cmds, []) - - def test_merged_address_group(self): - config = { - "group": { - "address_group": [ - {"name": "SERVERS", "address": ["192.168.1.10"]}, - ], - }, - } - cmds = build_commands(config, {}, "merged") - self.assertIn( - ("set", _BASE + ["address-group", "SERVERS", "address", "192.168.1.10"]), - cmds, - ) - - def test_merged_network_group(self): - config = { - "group": { - "network_group": [ - {"name": "LAN", "network": ["192.168.0.0/16"]}, - ], - }, - } - cmds = build_commands(config, {}, "merged") - self.assertIn( - ("set", _BASE + ["network-group", "LAN", "network", "192.168.0.0/16"]), - cmds, - ) - - def test_merged_port_group(self): - config = { - "group": { - "port_group": [ - {"name": "WEB", "port": ["80", "443"]}, - ], - }, - } - cmds = build_commands(config, {}, "merged") - self.assertIn( - ("set", _BASE + ["port-group", "WEB", "port", "80"]), - cmds, ) + cmds = dict_op(want, {}, _BASE, op="set") + paths = [c[1] for c in cmds] + self.assertIn(_BASE + ["address-group", "SERVERS", "address", "192.168.1.10"], paths) def test_merged_idempotent(self): - have = self._have() config = { "group": { - "address_group": [ - {"name": "SERVERS", "address": ["192.168.1.10", "192.168.1.11"]}, - ], - "network_group": [ - {"name": "LAN", "network": ["192.168.0.0/16"]}, - ], + "address_group": [{"name": "SERVERS", "address": ["192.168.1.10"]}], }, } - cmds = build_commands(config, have, "merged") + want = _want_to_device(config) + have = {"address-group": {"SERVERS": {"address": {"192.168.1.10": {}}}}} + cmds = dict_op(want, have, _BASE, op="set") self.assertEqual(cmds, []) - def test_replaced_removes_extra_group(self): - have = self._have() - config = { - "group": { - "network_group": [ - {"name": "DMZ", "network": ["10.0.0.0/8"]}, - ], + def test_merged_adds_network_group(self): + want = _want_to_device( + { + "group": { + "network_group": [{"name": "LAN", "network": ["192.168.0.0/16"]}], + }, }, + ) + cmds = dict_op(want, {}, _BASE, op="set") + paths = [c[1] for c in cmds] + self.assertIn(_BASE + ["network-group", "LAN", "network", "192.168.0.0/16"], paths) + + def test_purge_removes_extra_group(self): + want = _want_to_device( + { + "group": { + "network_group": [{"name": "DMZ", "network": ["10.0.0.0/8"]}], + }, + }, + ) + have = { + "address-group": {"SERVERS": {"address": {"192.168.1.10": {}}}}, + "network-group": {"LAN": {"network": {"192.168.0.0/16": {}}}}, } - cmds = build_commands(config, have, "replaced") + cmds = dict_op(want, have, _BASE, op="purge") paths = [c[1] for c in cmds] - self.assertIn(_BASE + ["address-group", "SERVERS"], paths) + # address-group entirely absent from want -> whole type deleted + self.assertIn(_BASE + ["address-group"], paths) self.assertIn(_BASE + ["network-group", "LAN"], paths) - def test_replaced_idempotent(self): - have = self._have() + def test_merged_idempotent_with_description(self): + """Regression test: the original implementation's have-side + normalization corrupted plain scalar fields (description) into + bogus presence-dicts via the same blanket conversion used for + member lists, breaking idempotency for any group with a + description set. Fixed by using dict_op's native list handling + for members and never touching scalar fields at all.""" config = { "group": { "address_group": [ - {"name": "SERVERS", "address": ["192.168.1.10", "192.168.1.11"]}, - ], - "network_group": [ - {"name": "LAN", "network": ["192.168.0.0/16"]}, + {"name": "SERVERS", "description": "Web servers", "address": ["10.0.0.1"]}, ], }, } - cmds = build_commands(config, have, "replaced") + want = _want_to_device(config) + have = { + "address-group": { + "SERVERS": {"description": "Web servers", "address": ["10.0.0.1"]}, + }, + } + cmds = dict_op(want, have, _BASE, op="set") self.assertEqual(cmds, []) + def test_replaced_removes_stale_member(self): + """Regression test: dict_op's purge mode originally had no + handling for list-valued leaves (only dicts), so a member + present on the device but absent from the desired list was + silently never removed under 'replaced'. Fixed centrally in + dict_op itself.""" + want = _want_to_device( + {"group": {"address_group": [{"name": "SERVERS", "address": ["10.0.0.1"]}]}}, + ) + have = {"address-group": {"SERVERS": {"address": ["10.0.0.1", "10.0.0.2"]}}} + cmds = dict_op(want, have, _BASE, op="purge") + self.assertIn( + ("delete", _BASE + ["address-group", "SERVERS", "address", "10.0.0.2"]), + cmds, + ) + + def test_deleted_removes_base(self): + # deleted state is handled in main() not dict_op + # just verify want_to_device produces correct shape + want = _want_to_device({}) + self.assertEqual(want, {}) + + +class TestBuildCommands(unittest.TestCase): + """build_commands/get_running_config were extracted from main() during + the dict_op refactor so they're independently testable.""" + + def test_get_running_config_fetches_at_base(self): + from unittest.mock import MagicMock + + mock_vyos = MagicMock() + mock_vyos.get_config = MagicMock(return_value={"address-group": {}}) + get_running_config(mock_vyos) + mock_vyos.get_config.assert_called_once_with(_BASE) + + def test_merged_idempotent_against_fixture(self): + fixture = load_fixture("firewall_global_running.json") + have = _device_to_argspec(fixture) + self.assertEqual(build_commands(have, fixture, "merged"), []) + + def test_replaced_idempotent_against_fixture(self): + fixture = load_fixture("firewall_global_running.json") + have = _device_to_argspec(fixture) + self.assertEqual(build_commands(have, fixture, "replaced"), []) + + def test_deleted_no_have_is_noop(self): + self.assertEqual(build_commands({}, {}, "deleted"), []) + + def test_deleted_with_have(self): + self.assertEqual( + build_commands({}, {"address-group": {"SERVERS": {}}}, "deleted"), + [("delete", _BASE)], + ) + + def test_single_value_member_collapse_no_char_iteration_bug(self): + """A group with exactly one member, collapsed by the device to a + bare string instead of a list, must not be iterated + character-by-character.""" + raw_have = {"port-group": {"WEB": {"port": "80"}}} + config = {"group": {"port_group": [{"name": "WEB", "port": ["80"]}]}} + self.assertEqual(build_commands(config, raw_have, "merged"), []) + + def test_group_tag_node_collapse_no_char_iteration_bug(self): + """A single group with zero other fields, collapsed by the + device to a bare group-name string, must not be iterated + character-by-character.""" + raw_have = {"address-group": "SERVERS"} + config = {"group": {"address_group": [{"name": "SERVERS"}]}} + self.assertEqual(build_commands(config, raw_have, "merged"), []) + if __name__ == "__main__": unittest.main() |