diff options
Diffstat (limited to 'tests/unit/modules/test_vyos_firewall_interfaces.py')
| -rw-r--r-- | tests/unit/modules/test_vyos_firewall_interfaces.py | 306 |
1 files changed, 210 insertions, 96 deletions
diff --git a/tests/unit/modules/test_vyos_firewall_interfaces.py b/tests/unit/modules/test_vyos_firewall_interfaces.py index 66b3883..26d2c4e 100644 --- a/tests/unit/modules/test_vyos_firewall_interfaces.py +++ b/tests/unit/modules/test_vyos_firewall_interfaces.py @@ -4,107 +4,188 @@ 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_firewall_interfaces import ( + _device_to_argspec, + _hook_filter_from_device, + _hook_filter_to_device, + _rules_from_device, + _rules_to_device, + _want_to_device, build_commands, get_running_config, ) - -_BASE = ["firewall"] +from .base import load_fixture -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) +_BASE = ["firewall"] class VyOSModuleTestCase(unittest.TestCase): def setUp(self): self.mock_vyos = MagicMock() self.fixture = load_fixture("firewall_interfaces_running.json") + self.mock_vyos.get_config = MagicMock(return_value=self.fixture) - def _set_afi(self, afi): - data = self.fixture.get(afi, {}) - self.mock_vyos.get_config = MagicMock(return_value=data) +class TestGetRunningConfig(VyOSModuleTestCase): + def test_single_combined_fetch(self): + """Confirm get_running_config fetches once at _BASE, not per-AFI.""" + get_running_config(self.mock_vyos) + self.mock_vyos.get_config.assert_called_once_with(_BASE) -class TestVyOSFirewallInterfacesGetRunning(VyOSModuleTestCase): + def test_returns_raw_device_dict(self): + self.assertEqual(get_running_config(self.mock_vyos), self.fixture) - def test_parses_ipv4_hooks(self): - self._set_afi("ipv4") - result = get_running_config(self.mock_vyos) - ipv4 = next((e for e in result if e["afi"] == "ipv4"), None) - self.assertIsNotNone(ipv4) - hook_names = [h["hook"] for h in ipv4["hooks"]] - self.assertIn("input", hook_names) - self.assertIn("forward", hook_names) - self.assertIn("output", hook_names) + def test_empty_config(self): + self.mock_vyos.get_config = MagicMock(return_value=None) + self.assertEqual(get_running_config(self.mock_vyos), {}) - def test_parses_input_rules(self): - self._set_afi("ipv4") - result = get_running_config(self.mock_vyos) - ipv4 = next(e for e in result if e["afi"] == "ipv4") - input_hook = next(h for h in ipv4["hooks"] if h["hook"] == "input") - self.assertEqual(input_hook["default_action"], "accept") - self.assertEqual(len(input_hook["rules"]), 2) - r10 = next(r for r in input_hook["rules"] if r["number"] == 10) - self.assertEqual(r10["action"], "accept") - self.assertEqual(r10["state"], "established") - - def test_parses_ipv6_hooks(self): - self._set_afi("ipv6") - result = get_running_config(self.mock_vyos) - ipv6 = next((e for e in result if e["afi"] == "ipv6"), None) - self.assertIsNotNone(ipv6) - self.assertEqual(ipv6["hooks"][0]["hook"], "input") - def test_empty_config(self): - self.mock_vyos.get_config = MagicMock(return_value={}) - result = get_running_config(self.mock_vyos) - self.assertEqual(result, []) +class TestRulesToDeviceFromDevice(unittest.TestCase): + def test_bare_rule_is_presence(self): + self.assertEqual(_rules_to_device([{"number": 10}]), {"10": {}}) + + def test_full_rule(self): + result = _rules_to_device( + [ + { + "number": 20, + "action": "drop", + "state": "invalid", + "source": {"address": "10.0.0.0/8"}, + "disable": True, + }, + ], + ) + self.assertEqual( + result, + { + "20": { + "action": "drop", + "state": "invalid", + "source": {"address": "10.0.0.0/8"}, + "disable": {}, + }, + }, + ) + + def test_from_device_number_cast_to_int(self): + result = _rules_from_device({"10": {"action": "accept"}}) + self.assertEqual(result, [{"number": 10, "action": "accept"}]) + + def test_from_device_sorted_numerically_not_lexically(self): + result = _rules_from_device({"20": {}, "9": {}, "100": {}}) + self.assertEqual([r["number"] for r in result], [9, 20, 100]) + + def test_source_destination_round_trip(self): + raw = {"20": {"source": {"address": "10.0.0.0/8"}, "destination": {"port": "22"}}} + result = _rules_from_device(raw) + self.assertEqual(result[0]["source"], {"address": "10.0.0.0/8"}) + self.assertEqual(result[0]["destination"], {"port": "22"}) + + +class TestHookFilterToDeviceFromDevice(unittest.TestCase): + def test_bare_hook_is_presence(self): + self.assertEqual(_hook_filter_to_device({"hook": "input"}), {}) + + def test_default_action_and_description(self): + result = _hook_filter_to_device( + {"hook": "input", "default_action": "accept", "description": "desc"}, + ) + self.assertEqual(result, {"default_action": "accept", "description": "desc"}) + + def test_with_rules(self): + result = _hook_filter_to_device( + {"hook": "input", "rules": [{"number": 10, "action": "accept"}]}, + ) + self.assertEqual(result, {"rule": {"10": {"action": "accept"}}}) + def test_from_device_basic(self): + entry = _hook_filter_from_device("input", {"default-action": "accept"}) + self.assertEqual(entry, {"hook": "input", "default_action": "accept"}) + + def test_from_device_with_rules(self): + entry = _hook_filter_from_device( + "input", + {"default-action": "accept", "rule": {"10": {"action": "accept"}}}, + ) + self.assertEqual(entry["default_action"], "accept") + self.assertEqual(entry["rules"], [{"number": 10, "action": "accept"}]) -class TestVyOSFirewallInterfacesBuildCommands(unittest.TestCase): - def _have(self): - return [ +class TestWantToDevice(unittest.TestCase): + def test_empty(self): + self.assertEqual(_want_to_device([]), {}) + self.assertEqual(_want_to_device(None), {}) + + def test_afi_with_no_hooks_omitted(self): + self.assertEqual(_want_to_device([{"afi": "ipv4", "hooks": []}]), {}) + + def test_full_config(self): + config = [ { "afi": "ipv4", "hooks": [ - { - "hook": "input", - "default_action": "accept", - "rules": [ - {"number": 10, "action": "accept", "state": "established"}, - {"number": 20, "action": "drop", "state": "invalid"}, - ], - }, - {"hook": "forward", "default_action": "accept"}, + {"hook": "input", "default_action": "accept"}, ], }, ] + result = _want_to_device(config) + self.assertEqual( + result, + {"ipv4": {"input": {"filter": {"default_action": "accept"}}}}, + ) - def test_deleted_all(self): - cmds = build_commands([], self._have(), "deleted") - paths = [c[1] for c in cmds] - self.assertIn(_BASE + ["ipv4", "input", "filter"], paths) - self.assertIn(_BASE + ["ipv4", "forward", "filter"], paths) - def test_deleted_specific(self): - config = [{"afi": "ipv4", "hooks": [{"hook": "input"}]}] - cmds = build_commands(config, self._have(), "deleted") - self.assertIn(("delete", _BASE + ["ipv4", "input", "filter"]), cmds) - paths = [c[1] for c in cmds] - self.assertNotIn(_BASE + ["ipv4", "forward", "filter"], paths) +class TestDeviceToArgspecFixture(VyOSModuleTestCase): + def test_ipv4_input_with_rules(self): + have = _device_to_argspec(self.fixture) + ipv4 = next(e for e in have if e["afi"] == "ipv4") + input_hook = next(h for h in ipv4["hooks"] if h["hook"] == "input") + self.assertEqual(input_hook["default_action"], "accept") + rule20 = next(r for r in input_hook["rules"] if r["number"] == 20) + self.assertEqual(rule20["source"], {"address": "10.0.0.0/8"}) + self.assertEqual(rule20["destination"], {"port": "22"}) + + def test_sibling_module_data_never_surfaces(self): + """Regression test: firewall.ipv4.name (owned by + vyos_firewall_rules) must never appear in this module's output.""" + have = _device_to_argspec(self.fixture) + ipv4 = next(e for e in have if e["afi"] == "ipv4") + hook_names = {h["hook"] for h in ipv4["hooks"]} + self.assertEqual(hook_names, {"input", "forward", "output"}) + + def test_ipv6_present(self): + have = _device_to_argspec(self.fixture) + afis = {e["afi"] for e in have} + self.assertIn("ipv6", afis) + + 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 = _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_overridden_idempotent_against_own_fixture(self): + have = _device_to_argspec(self.fixture) + self.assertEqual(build_commands(have, self.fixture, "overridden"), []) - def test_merged_hook(self): + def test_replaced_scoped_to_named_hooks_only(self): + """replaced only touches hooks explicitly named in config -- an + omitted hook (output here) must be left alone.""" config = [ { "afi": "ipv4", @@ -112,23 +193,24 @@ class TestVyOSFirewallInterfacesBuildCommands(unittest.TestCase): { "hook": "input", "default_action": "accept", - "rules": [{"number": 10, "action": "accept", "state": "established"}], + "rules": [ + {"number": 10, "action": "accept", "state": "established"}, + { + "number": 20, + "action": "drop", + "state": "invalid", + "source": {"address": "10.0.0.0/8"}, + "destination": {"port": "22"}, + }, + ], }, ], }, ] - cmds = build_commands(config, [], "merged") - self.assertIn( - ("set", _BASE + ["ipv4", "input", "filter", "default-action", "accept"]), - cmds, - ) - self.assertIn( - ("set", _BASE + ["ipv4", "input", "filter", "rule", "10", "action", "accept"]), - cmds, - ) + self.assertEqual(build_commands(config, self.fixture, "replaced"), []) - def test_merged_idempotent(self): - have = self._have() + def test_overridden_deletes_omitted_hook(self): + """overridden is full-model: an omitted hook must be deleted.""" config = [ { "afi": "ipv4", @@ -138,30 +220,62 @@ class TestVyOSFirewallInterfacesBuildCommands(unittest.TestCase): "default_action": "accept", "rules": [ {"number": 10, "action": "accept", "state": "established"}, - {"number": 20, "action": "drop", "state": "invalid"}, + { + "number": 20, + "action": "drop", + "state": "invalid", + "source": {"address": "10.0.0.0/8"}, + "destination": {"port": "22"}, + }, ], }, - {"hook": "forward", "default_action": "accept"}, ], }, ] - cmds = build_commands(config, have, "merged") - self.assertEqual(cmds, []) + cmds = build_commands(config, self.fixture, "overridden") + self.assertIn(("delete", _BASE + ["ipv4", "output", "filter"]), cmds) + self.assertIn(("delete", _BASE + ["ipv4", "forward", "filter"]), cmds) + self.assertIn(("delete", _BASE + ["ipv6", "input", "filter"]), cmds) - def test_overridden_removes_extra_hook(self): - have = self._have() - config = [ - { - "afi": "ipv4", - "hooks": [ - {"hook": "output", "default_action": "accept"}, - ], - }, - ] - cmds = build_commands(config, have, "overridden") - paths = [c[1] for c in cmds] - self.assertIn(_BASE + ["ipv4", "input", "filter"], paths) - self.assertIn(_BASE + ["ipv4", "forward", "filter"], paths) + def test_overridden_never_touches_sibling_ruleset(self): + cmds = build_commands([], self.fixture, "overridden") + self.assertTrue(all("name" not in c[1] for c in cmds)) + + def test_deleted_never_touches_sibling_ruleset(self): + cmds = build_commands([], self.fixture, "deleted") + self.assertTrue(all("name" not in c[1] for c in cmds)) + self.assertIn(("delete", _BASE + ["ipv4", "input", "filter"]), cmds) + + def test_deleted_scoped_to_named_config(self): + cmds = build_commands( + [{"afi": "ipv4", "hooks": [{"hook": "input"}]}], + self.fixture, + "deleted", + ) + self.assertEqual(cmds, [("delete", _BASE + ["ipv4", "input", "filter"])]) + + def test_collapsed_rule_no_char_iteration_bug(self): + """A single rule with no other config collapsed to a bare string + by the device must not be iterated character-by-character.""" + raw_have = {"ipv4": {"input": {"filter": {"rule": "10"}}}} + config = [{"afi": "ipv4", "hooks": [{"hook": "input", "rules": [{"number": 10}]}]}] + self.assertEqual(build_commands(config, raw_have, "merged"), []) + + def test_merged_new_rule(self): + cmds = build_commands( + [ + { + "afi": "ipv4", + "hooks": [{"hook": "input", "rules": [{"number": 30, "action": "accept"}]}], + }, + ], + self.fixture, + "merged", + ) + self.assertIn( + ("set", _BASE + ["ipv4", "input", "filter", "rule", "30", "action", "accept"]), + cmds, + ) if __name__ == "__main__": |