diff options
Diffstat (limited to 'tests/unit/modules/test_vyos_route_maps.py')
| -rw-r--r-- | tests/unit/modules/test_vyos_route_maps.py | 506 |
1 files changed, 374 insertions, 132 deletions
diff --git a/tests/unit/modules/test_vyos_route_maps.py b/tests/unit/modules/test_vyos_route_maps.py index 97d814b..4eec600 100644 --- a/tests/unit/modules/test_vyos_route_maps.py +++ b/tests/unit/modules/test_vyos_route_maps.py @@ -4,189 +4,431 @@ 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_route_maps import ( - _want_to_api_match, - _want_to_api_set, + ARGUMENT_SPEC, + _derive_key_field, + _device_to_argspec, + _keyed_list_from_device, + _keyed_list_to_device, + _match_from_device, + _match_to_device, + _rule_entry_from_device, + _rule_entry_to_device, + _seed_route_map_placeholders, + _set_from_device, + _set_to_device, + _want_to_device, build_commands, get_running_config, ) +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 = ["policy", "route-map"] 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 - + self.fixture = load_fixture("route_maps_running.json") + self.mock_vyos.get_config = MagicMock(return_value=self.fixture) -class TestVyOSRouteMapsGetRunning(VyOSModuleTestCase): - def setUp(self): - super().setUp() - self.fixture = load_fixture("route_maps_running.json") +class TestGetRunningConfig(VyOSModuleTestCase): + def test_unwraps_route_map_wrapper_key(self): + """Confirmed against the pre-existing fixture (built from real + device data): the REST API wraps the response in an extra + "route-map" key even when querying at the policy/route-map + path itself -- the same defensive-unwrap pattern every other + module this session needed for its own top-level get_config.""" + result = get_running_config(self.mock_vyos) + self.assertIn("RM-TEST-EXPORT-POLICY", result) + self.assertNotIn("route-map", result) - def test_unwraps_route_map_nesting(self): - """API returns {"route-map": {"NAME": {...}}} — must unwrap.""" - self.set_running_config(self.fixture) + def test_no_wrapper_key_passes_through(self): + self.mock_vyos.get_config = MagicMock(return_value={"RM1": {"rule": {}}}) result = get_running_config(self.mock_vyos) - names = [e["route_map"] for e in result] + self.assertIn("RM1", 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_route_map_key(self): + opts = ARGUMENT_SPEC["config"]["options"] + self.assertEqual(_derive_key_field(opts), "route_map") + + def test_derives_sequence_key(self): + entry_opts = ARGUMENT_SPEC["config"]["options"]["entries"]["options"] + self.assertEqual(_derive_key_field(entry_opts), "sequence") + + def test_raises_if_none_required(self): + with self.assertRaises(ValueError): + _derive_key_field({"a": {"type": "str"}}) + + def test_raises_if_more_than_one_required(self): + with self.assertRaises(ValueError): + _derive_key_field({"a": {"required": True}, "b": {"required": True}}) + + +class TestKeyedListHelper(unittest.TestCase): + def test_to_device_default_transform_is_autoclean(self): + result = _keyed_list_to_device([{"route_map": "RM1", "description": "x"}], "route_map") + self.assertEqual(result, {"RM1": {"description": "x"}}) + + def test_from_device_default_transform_is_from_device(self): + result = _keyed_list_from_device({"RM1": {"description": "x"}}, "route_map") + self.assertEqual(result, [{"route_map": "RM1", "description": "x"}]) + + def test_empty(self): + self.assertEqual(_keyed_list_to_device([], "route_map"), {}) + self.assertEqual(_keyed_list_from_device({}, "route_map"), []) + + +class TestMatchToDeviceFromDevice(unittest.TestCase): + """Most match options are fully generic; only prefix_list/ + prefix_list6 and ip/ipv6 nexthop matching are genuine structural + exceptions (confirmed against vyos-1x: the device nests these + deeper than the argspec).""" + + def test_simple_fields_generic(self): + result = _match_to_device({"peer": "192.0.2.1", "protocol": "bgp", "metric": 100}) + self.assertEqual(result, {"peer": "192.0.2.1", "protocol": "bgp", "metric": 100}) + + def test_prefix_list_nested_two_levels(self): + result = _match_to_device({"prefix_list": "PL1"}) + self.assertEqual(result, {"ip": {"address": {"prefix-list": "PL1"}}}) + + def test_prefix_list6_nested_two_levels(self): + result = _match_to_device({"prefix_list6": "PL6"}) + self.assertEqual(result, {"ipv6": {"address": {"prefix-list": "PL6"}}}) + + def test_ip_nexthop_extra_nesting_level(self): + result = _match_to_device( + {"ip": {"nexthop_address": "10.0.0.1", "nexthop_prefix_list": "PL2"}}, + ) + self.assertEqual( + result, + {"ip": {"nexthop": {"address": "10.0.0.1", "prefix-list": "PL2"}}}, + ) + + def test_ipv6_nexthop_extra_nesting_level(self): + result = _match_to_device({"ipv6": {"nexthop_address": "2001:db8::1"}}) + self.assertEqual(result, {"ipv6": {"nexthop": {"address": "2001:db8::1"}}}) + + def test_from_device_prefix_list(self): + entry = _match_from_device({"ip": {"address": {"prefix-list": "PL1"}}}) + self.assertEqual(entry["prefix_list"], "PL1") + + def test_from_device_nexthop(self): + entry = _match_from_device({"ip": {"nexthop": {"address": "10.0.0.1"}}}) + self.assertEqual(entry["ip"], {"nexthop_address": "10.0.0.1"}) + + def test_from_device_generic_fields(self): + entry = _match_from_device({"peer": "192.0.2.1", "protocol": "bgp"}) + self.assertEqual(entry, {"peer": "192.0.2.1", "protocol": "bgp"}) + + def test_empty(self): + self.assertEqual(_match_to_device({}), {}) + self.assertEqual(_match_to_device(None), {}) + self.assertEqual(_match_from_device({}), {}) + self.assertEqual(_match_from_device(None), {}) + + +class TestSetToDeviceFromDevice(unittest.TestCase): + """as_path_* collapse onto one nested device node. community/ + large_community/ipv6_next_hop are fully generic once modeled as + real nested dicts. "as_" is a genuine Python-keyword-collision + rename, nested inside aggregator specifically.""" + + def test_atomic_aggregate_fully_generic(self): + result = _set_to_device({"atomic_aggregate": True}) + self.assertEqual(result, {"atomic_aggregate": {}}) + + def test_as_path_options_collapse_onto_one_node(self): + result = _set_to_device( + {"as_path_exclude": "111", "as_path_prepend": "65001", "as_path_prepend_last_as": 2}, + ) + self.assertEqual( + result["as-path"], + {"exclude": "111", "prepend": "65001", "prepend-last-as": 2}, + ) + + def test_aggregator_as_rename(self): + """Regression test for the real bug caught this session: "as_" + is nested inside "aggregator", not a top-level set field -- a + flat rename map applied only at the top level misses it + entirely.""" + result = _set_to_device({"aggregator": {"as_": 100, "ip": "10.0.0.5"}}) + self.assertEqual(result, {"aggregator": {"as": 100, "ip": "10.0.0.5"}}) + + def test_aggregator_as_only(self): + result = _set_to_device({"aggregator": {"as_": 100}}) + self.assertEqual(result, {"aggregator": {"as": 100}}) + + def test_community_add_stays_a_plain_list(self): + result = _set_to_device({"community": {"add": ["no-export", "no-advertise"]}}) + self.assertEqual(result["community"], {"add": ["no-export", "no-advertise"]}) + + def test_large_community_none_presence(self): + result = _set_to_device({"large_community": {"none": True}}) + self.assertEqual(result["large_community"], {"none": {}}) + + def test_ipv6_next_hop_generic(self): + result = _set_to_device({"ipv6_next_hop": {"global": "2001:db8::1"}}) + self.assertEqual(result["ipv6_next_hop"], {"global": "2001:db8::1"}) + + def test_ipv6_next_hop_valueless_options(self): + result = _set_to_device({"ipv6_next_hop": {"peer_address": True, "prefer_global": True}}) + self.assertEqual( + result["ipv6_next_hop"], + {"peer_address": {}, "prefer_global": {}}, + ) + + def test_from_device_community_add(self): + entry = _set_from_device({"community": {"add": ["no-export"]}}) + self.assertEqual(entry["community"], {"add": ["no-export"]}) + + def test_from_device_large_community_none(self): + entry = _set_from_device({"large-community": {"none": {}}}) + self.assertEqual(entry["large_community"], {"none": True}) + + def test_from_device_as_path(self): + entry = _set_from_device({"as-path": {"exclude": "111", "prepend-last-as": "2"}}) + self.assertEqual(entry["as_path_exclude"], "111") + self.assertEqual(entry["as_path_prepend_last_as"], 2) + + def test_from_device_aggregator_as_rename_with_int_cast(self): + entry = _set_from_device({"aggregator": {"as": "100", "ip": "10.0.0.5"}}) + self.assertEqual(entry["aggregator"]["as_"], 100) + self.assertEqual(entry["aggregator"]["ip"], "10.0.0.5") + + def test_empty(self): + self.assertEqual(_set_to_device({}), {}) + self.assertEqual(_set_to_device(None), {}) + self.assertEqual(_set_from_device({}), {}) + self.assertEqual(_set_from_device(None), {}) + + +class TestRuleEntryToDeviceFromDevice(unittest.TestCase): + def test_continue_sequence_renamed(self): + """ "continue" is a Python keyword and can't be used as a + dict() kwarg -- "continue_sequence" is the unavoidable argspec + name, handled directly at the rule level (not a set field).""" + result = _rule_entry_to_device({"continue_sequence": 20}) + self.assertEqual(result["continue"], 20) + + def test_generic_fields(self): + result = _rule_entry_to_device({"action": "permit", "call": "RM2"}) + self.assertEqual(result, {"action": "permit", "call": "RM2"}) + + def test_from_device_continue(self): + entry = _rule_entry_from_device({"continue": "20"}) + self.assertEqual(entry["continue_sequence"], 20) + + def test_from_device_bare_collapse(self): + entry = _rule_entry_from_device(None) + self.assertEqual(entry, {}) + + +class TestWantToDevice(unittest.TestCase): + def test_empty(self): + self.assertEqual(_want_to_device([]), {}) + self.assertEqual(_want_to_device(None), {}) + + def test_route_map_without_entries_omitted(self): + self.assertEqual(_want_to_device([{"route_map": "RM1"}]), {}) + + def test_keyed_by_route_map_name(self): + config = [{"route_map": "RM1", "entries": [{"sequence": 10, "action": "permit"}]}] + result = _want_to_device(config) + self.assertIn("10", result["RM1"]["rule"]) + + def test_underscore_route_map_name_stays_verbatim(self): + """Confirmed against vyos-1x: route-map names may legitimately + contain underscores. _want_to_device itself must not alter the + key -- the dict_op-level protection is tested separately in + TestBuildCommands.""" + config = [{"route_map": "my_route_map", "entries": [{"sequence": 10}]}] + result = _want_to_device(config) + self.assertIn("my_route_map", result) + + +class TestSeedRouteMapPlaceholders(unittest.TestCase): + """Regression tests for the confirmed bug: dict_op's fallback + guesses a kebab-cased device key whenever a want key is missing + from have -- correct for schema field names, wrong for a route-map + name (an opaque value that may contain an underscore). Reproduced + directly before this fix: "my_route_map" became "my-route-map" in + the generated command on first creation.""" + + def test_seeds_new_route_map_verbatim(self): + want = {"my_route_map": {"rule": {"10": {}}}} + have = {} + _seed_route_map_placeholders(want, have) + self.assertIn("my_route_map", have) + + def test_seeds_new_rule_with_none_not_empty_dict(self): + """Seeding with {} instead of None would make dict_op think a + presence-only rule already matches and skip emitting its set + command -- the same mistake caught once already this session.""" + want = {"RM1": {"rule": {"10": {}}}} + have = {"RM1": {"rule": {}}} + _seed_route_map_placeholders(want, have) + self.assertIsNone(have["RM1"]["rule"]["10"]) + + def test_does_not_overwrite_existing_entries(self): + want = {"RM1": {"rule": {"10": {}}}} + have = {"RM1": {"rule": {"10": {"action": "permit"}}}} + _seed_route_map_placeholders(want, have) + self.assertEqual(have["RM1"]["rule"]["10"], {"action": "permit"}) + + +class TestDeviceToArgspecFixture(VyOSModuleTestCase): + def test_all_route_maps_parsed(self): + raw = get_running_config(self.mock_vyos) + result = _device_to_argspec(raw) + names = [rm["route_map"] for rm in result] self.assertIn("RM-TEST-EXPORT-POLICY", names) self.assertIn("rm1", names) - # "route-map" itself must NOT appear as a route map name - self.assertNotIn("route-map", names) - - def test_parses_rule_action(self): - self.set_running_config(self.fixture) - result = get_running_config(self.mock_vyos) - rm = next(e for e in result if e["route_map"] == "RM-TEST-EXPORT-POLICY") - rule = rm["entries"][0] - self.assertEqual(rule["action"], "permit") - self.assertEqual(rule["sequence"], 10) - def test_parses_match_peer(self): - self.set_running_config(self.fixture) - result = get_running_config(self.mock_vyos) - rm = next(e for e in result if e["route_map"] == "RM-TEST-EXPORT-POLICY") + def test_prefix_list_and_nexthop_match_parsed(self): + raw = get_running_config(self.mock_vyos) + result = _device_to_argspec(raw) + rm = next(rm for rm in result if rm["route_map"] == "RM-TEST-EXPORT-POLICY") rule = rm["entries"][0] - self.assertEqual(rule["match"]["peer"], "192.0.2.32") + self.assertEqual(rule["match"]["prefix_list"], "PL-MATCH") + self.assertEqual(rule["match"]["ip"]["nexthop_address"], "10.0.0.1") - def test_parses_set_fields(self): - self.set_running_config(self.fixture) - result = get_running_config(self.mock_vyos) - rm = next(e for e in result if e["route_map"] == "RM-TEST-EXPORT-POLICY") + def test_community_add_parsed_as_list(self): + raw = get_running_config(self.mock_vyos) + result = _device_to_argspec(raw) + rm = next(rm for rm in result if rm["route_map"] == "RM-TEST-EXPORT-POLICY") rule = rm["entries"][0] - self.assertEqual(rule["set"]["metric"], "5") - self.assertEqual(rule["set"]["aggregator"]["as"], "100") - self.assertEqual(rule["set"]["as-path"]["exclude"], "111") - - def test_empty_returns_empty_list(self): - self.set_running_config({}) - result = get_running_config(self.mock_vyos) - self.assertEqual(result, []) - - -class TestVyOSRouteMapsWantToApi(unittest.TestCase): - - def test_as_path_exclude_nested(self): - """as_path_exclude maps to nested as-path.exclude.""" - result = _want_to_api_set({"as_path_exclude": "111"}) - self.assertEqual(result["as-path"]["exclude"], "111") - - def test_metric_flat(self): - result = _want_to_api_set({"metric": "5"}) - self.assertEqual(result["metric"], "5") + self.assertEqual(rule["set"]["community"]["add"], ["no-export", "no-advertise"]) - def test_aggregator_as(self): - result = _want_to_api_set({"aggregator": {"as": 100}}) - self.assertEqual(result["aggregator"]["as"], "100") + def test_large_community_none_parsed(self): + raw = get_running_config(self.mock_vyos) + result = _device_to_argspec(raw) + rm1 = next(rm for rm in result if rm["route_map"] == "rm1") + self.assertTrue(rm1["entries"][0]["set"]["large_community"]["none"]) - def test_aggregator_as_underscore(self): - """aggregator.as_ is an alias for aggregator.as.""" - result = _want_to_api_set({"aggregator": {"as_": 100}}) - self.assertEqual(result["aggregator"]["as"], "100") + def test_aggregator_as_parsed(self): + raw = get_running_config(self.mock_vyos) + result = _device_to_argspec(raw) + rm = next(rm for rm in result if rm["route_map"] == "RM-TEST-EXPORT-POLICY") + self.assertEqual(rm["entries"][0]["set"]["aggregator"]["as_"], 100) - def test_large_community_presence_node(self): - result = _want_to_api_set({"large_community": "none"}) - self.assertEqual(result["large-community"], {"none": {}}) + def test_empty_config(self): + self.assertEqual(_device_to_argspec({}), []) + self.assertEqual(_device_to_argspec(None), []) - def test_match_peer(self): - result = _want_to_api_match({"peer": "192.0.2.32"}) - self.assertEqual(result["peer"], "192.0.2.32") - -class TestVyOSRouteMapsBuildCommands(unittest.TestCase): - - def _have_empty(self): - return [] - - def _have_with_rm(self): - return [ +class TestBuildCommands(VyOSModuleTestCase): + def setUp(self): + super().setUp() + self.raw = get_running_config(self.mock_vyos) + + def test_merged_idempotent_against_own_fixture(self): + have = _device_to_argspec(self.raw) + self.assertEqual(build_commands(have, self.raw, "merged"), []) + + def test_replaced_idempotent_against_own_fixture(self): + have = _device_to_argspec(self.raw) + self.assertEqual(build_commands(have, self.raw, "replaced"), []) + + def test_overridden_idempotent_against_own_fixture(self): + have = _device_to_argspec(self.raw) + self.assertEqual(build_commands(have, self.raw, "overridden"), []) + + def test_underscore_route_map_name_not_kebab_cased_on_creation(self): + """The primary confirmed bug this session, reproduced directly + end to end before the fix: "my_route_map" became + "my-route-map" in the generated command on first creation.""" + config = [{"route_map": "my_route_map", "entries": [{"sequence": 10, "action": "permit"}]}] + cmds = build_commands(config, {}, "merged") + self.assertFalse(any("my-route-map" in str(c) for c in cmds)) + self.assertTrue(any("my_route_map" in str(c) for c in cmds)) + + def test_replaced_scoped_to_named_route_map_only(self): + have = _device_to_argspec(self.raw) + config = [ { - "route_map": "RM1", - "entries": [ - { - "sequence": 10, - "action": "permit", - "match": {"peer": "192.0.2.32"}, - "set": {"metric": "5", "as-path": {"exclude": "111"}}, - }, - ], + "route_map": "RM-TEST-EXPORT-POLICY", + "entries": have[0]["entries"], }, ] + cmds = build_commands(config, self.raw, "replaced") + self.assertEqual(cmds, []) + self.assertFalse(any("rm1" in str(c) for c in cmds)) - def test_merged_adds_new_rm(self): + def test_replaced_removes_omitted_field_full_replace_semantic(self): + """Confirms this is the intended "replaced" semantic (matching + every other module this session), not a bug: omitting a field + from a route map named in "replaced" removes it.""" config = [ { - "route_map": "RM-NEW", + "route_map": "RM-TEST-EXPORT-POLICY", "entries": [{"sequence": 10, "action": "permit"}], }, ] - cmds = build_commands(config, self._have_empty(), "merged") - paths = [c[1] for c in cmds] - self.assertIn(["policy", "route-map", "RM-NEW", "rule", "10", "action", "permit"], paths) + cmds = build_commands(config, self.raw, "replaced") + self.assertIn(("delete", _BASE + ["RM-TEST-EXPORT-POLICY", "rule", "10", "set"]), cmds) + self.assertIn(("delete", _BASE + ["RM-TEST-EXPORT-POLICY", "rule", "10", "match"]), cmds) - def test_merged_idempotent(self): + def test_overridden_deletes_omitted_route_map(self): + have = _device_to_argspec(self.raw) config = [ { - "route_map": "RM1", - "entries": [ - { - "sequence": 10, - "action": "permit", - "match": {"peer": "192.0.2.32"}, - "set": {"metric": "5", "as_path_exclude": "111"}, - }, - ], + "route_map": "RM-TEST-EXPORT-POLICY", + "entries": have[0]["entries"], }, ] - cmds = build_commands(config, self._have_with_rm(), "merged") - self.assertEqual(cmds, []) + cmds = build_commands(config, self.raw, "overridden") + self.assertIn(("delete", _BASE + ["rm1"]), cmds) + + def test_deleted_scoped_to_named_route_map(self): + cmds = build_commands([{"route_map": "rm1"}], self.raw, "deleted") + self.assertEqual(cmds, [("delete", _BASE + ["rm1"])]) - def test_deleted_no_config_deletes_all(self): - cmds = build_commands([], self._have_with_rm(), "deleted") - self.assertIn(("delete", ["policy", "route-map"]), cmds) + def test_deleted_no_config_removes_all(self): + cmds = build_commands([], self.raw, "deleted") + self.assertEqual(cmds, [("delete", _BASE)]) - def test_deleted_with_config_deletes_named(self): - config = [{"route_map": "RM1"}] - cmds = build_commands(config, self._have_with_rm(), "deleted") - self.assertIn(("delete", ["policy", "route-map", "RM1"]), cmds) + def test_deleted_named_nonexistent_is_noop(self): + cmds = build_commands([{"route_map": "NONEXISTENT"}], self.raw, "deleted") + self.assertEqual(cmds, []) + + def test_collapsed_single_rule_no_char_iteration_bug(self): + raw_have = {"RM1": {"rule": "10"}} + config = [{"route_map": "RM1", "entries": [{"sequence": 10}]}] + self.assertEqual(build_commands(config, raw_have, "merged"), []) - def test_replaced_deletes_then_resets(self): + def test_merged_new_rule_with_community(self): config = [ { - "route_map": "RM1", - "entries": [{"sequence": 10, "action": "deny"}], + "route_map": "RM-NEW", + "entries": [ + { + "sequence": 10, + "action": "permit", + "set": {"community": {"add": ["no-export"]}}, + }, + ], }, ] - cmds = build_commands(config, self._have_with_rm(), "replaced") - ops = [c[0] for c in cmds] - # delete must come before set - self.assertIn("delete", ops) - self.assertIn("set", ops) - delete_idx = ops.index("delete") - set_idx = ops.index("set") - self.assertLess(delete_idx, set_idx) - - def test_overridden_removes_extra_rm(self): - config = [{"route_map": "RM-NEW", "entries": []}] - have = self._have_with_rm() # has RM1 - cmds = build_commands(config, have, "overridden") - self.assertIn(("delete", ["policy", "route-map", "RM1"]), cmds) + cmds = build_commands(config, {}, "merged") + self.assertIn( + ("set", _BASE + ["RM-NEW", "rule", "10", "set", "community", "add", "no-export"]), + cmds, + ) if __name__ == "__main__": |