diff options
| author | omnom62 <omnom62@outlook.com> | 2026-07-06 14:54:12 +1000 |
|---|---|---|
| committer | John Estabrook <jestabro@vyos.io> | 2026-08-21 14:02:19 -0500 |
| commit | 45dc95873fd906c582fbbd5e6ca3838caf867399 (patch) | |
| tree | 71b3e53e0080da60abf9fd2381705971dfe4d507 /tests/unit/modules/test_vyos_nat.py | |
| parent | 7a6b5e4f3a7a021cfa75faa7bf833741dfc09cff (diff) | |
| download | rest.vyos-T8989_wave4.tar.gz rest.vyos-T8989_wave4.zip | |
T8989: wave4 vyos_command, dict_op refactorT8989_wave4
* T8989: vyos_command module
* T8989: vyos_command module UAT and SIT
* T8989: vyos_command changelog
* T8989: vyos_command linter
* T8989: vyos_config module
* T8989: vyos_config module changelog
* T8989: Wave 4 vyos_config module with integration and unit tests
* T8323: vyos_system module
* T8332: vyos_system SIT and UAT
* T8323: vyos_vlan module
* T8323: vyos_vlan module
* T8323: vyos_vlan module SIT and UAT
* T8323: vyos_system module
* T8989: Wave 4 vyos_vlan reworked with dict_op engine
* T8989: Fix dict_op single-value string list handling, add vyos_system integration tests
* T8989: logging_global refactor
* T8989: migrate ntp_global, logging_global, firewall_global to dict_op engine
* T8989: vyos_nat module for REST API collection
* T8989: vyos_nat module for REST API collection, linter fixes
* T8989: vyos_ha module for REST API collection
* T8989: vyos_ha module for REST API collection
* T8989: vyos_ha module sanity and linter fixes
* T8989: vyos_ha module sanity and linter fixes
* T8989: vyos_ha module linter fixes
* T8989: vyos.rest AI comment fixes
* T8323: vyos_nat AI comment fixes
* T8989 ai fixes
* T8989: vyos_bgp_address_family dict_op
* T8989: vyos_bgp_address_family vyos_bgp_global dict_op
* T8989: dict_op refactor for firewall_*, nat, user
* T8989: dict_op refactor for firewall_*, nat, user
* T8989: dict_op refactor for ntp_global, ha
* T8989: snmp_server dict_op refactor
* T8989: snmp_server dict_op refactor
* T8989: route_map dict_op refactor
Diffstat (limited to 'tests/unit/modules/test_vyos_nat.py')
| -rw-r--r-- | tests/unit/modules/test_vyos_nat.py | 558 |
1 files changed, 558 insertions, 0 deletions
diff --git a/tests/unit/modules/test_vyos_nat.py b/tests/unit/modules/test_vyos_nat.py new file mode 100644 index 0000000..9393e03 --- /dev/null +++ b/tests/unit/modules/test_vyos_nat.py @@ -0,0 +1,558 @@ +# -*- 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_nat import ( + _cgnat_from_device, + _cgnat_to_device, + _device_to_argspec, + _normalize_nat_have, + _rules_from_device, + _rules_to_device, + _want_to_device, +) + +from .base import load_fixture + + +def _load_nat_fixture(): + return load_fixture("nat_running.json") + + +class TestRulesToDevice(unittest.TestCase): + """Keys stay snake_case here -- dict_op does kebab translation itself + at comparison time, so _rules_to_device must not do it manually.""" + + def test_simple_source_rule(self): + rules = [ + { + "id": 100, + "outbound_interface": {"name": "eth0"}, + "translation": {"address": "masquerade"}, + }, + ] + result = _rules_to_device(rules) + self.assertIn("100", result) + self.assertEqual(result["100"]["outbound_interface"]["name"], "eth0") + self.assertEqual(result["100"]["translation"]["address"], "masquerade") + + def test_bool_fields_become_presence_nodes(self): + rules = [ + { + "id": 100, + "log": True, + "exclude": True, + "translation": {"address": "masquerade"}, + }, + ] + result = _rules_to_device(rules) + self.assertEqual(result["100"]["log"], {}) + self.assertEqual(result["100"]["exclude"], {}) + + def test_false_bool_not_emitted(self): + rules = [{"id": 100, "log": False, "translation": {"address": "masquerade"}}] + result = _rules_to_device(rules) + self.assertNotIn("log", result["100"]) + + def test_destination_rule_with_port(self): + rules = [ + { + "id": 200, + "protocol": "tcp", + "inbound_interface": {"name": "eth0"}, + "destination": {"address": "198.51.100.10", "port": "80"}, + "translation": {"address": "192.168.1.10", "port": "8080"}, + }, + ] + result = _rules_to_device(rules) + self.assertEqual(result["200"]["protocol"], "tcp") + self.assertEqual(result["200"]["destination"]["address"], "198.51.100.10") + self.assertEqual(result["200"]["destination"]["port"], "80") + + def test_static_rule_inbound_interface_string(self): + """inbound_interface is a genuine union type: a plain string for + static NAT (confirmed vyos-1x: bare leafNode), a dict for source/ + destination NAT (confirmed: node with name/group children). No + special-casing needed either way -- autoclean passes a string + through unchanged and recurses into a dict identically.""" + rules = [ + { + "id": 300, + "inbound_interface": "eth0", + "destination": {"address": "198.51.100.20"}, + "translation": {"address": "192.168.1.20"}, + }, + ] + result = _rules_to_device(rules) + self.assertEqual(result["300"]["inbound_interface"], "eth0") + + def test_multiple_rules_keyed_by_id(self): + rules = [ + {"id": 100, "translation": {"address": "masquerade"}}, + {"id": 200, "translation": {"address": "masquerade"}}, + ] + result = _rules_to_device(rules) + self.assertIn("100", result) + self.assertIn("200", result) + + def test_none_values_not_emitted(self): + rules = [{"id": 100, "description": None, "translation": {"address": "masquerade"}}] + result = _rules_to_device(rules) + self.assertNotIn("description", result["100"]) + + def test_load_balance_hash_stays_a_plain_list(self): + """hash is a multi-value leafNode (confirmed <multi/>), not a tag + node -- it must pass through as a plain list untouched, letting + dict_op's own native list handling manage it.""" + rules = [ + { + "id": 100, + "load_balance": {"hash": ["source-address", "random"]}, + "translation": {"address": "masquerade"}, + }, + ] + result = _rules_to_device(rules) + self.assertEqual(result["100"]["load_balance"]["hash"], ["source-address", "random"]) + + def test_load_balance_backend_reshaped_to_tag_node(self): + """backend IS a genuine tag node (confirmed: nested "weight" + leaf), unlike hash -- this one needs the structural reshape.""" + rules = [ + { + "id": 100, + "load_balance": {"backend": [{"ip": "192.168.1.10", "weight": 50}]}, + "translation": {"address": "masquerade"}, + }, + ] + result = _rules_to_device(rules) + self.assertEqual(result["100"]["load_balance"]["backend"], {"192.168.1.10": {"weight": 50}}) + + def test_load_balance_backend_without_weight_is_bare_presence(self): + """Regression check for the autoclean-based simplification: a + backend entry with no weight must still produce a bare presence + node, matching the previous manual if/else exactly.""" + rules = [ + { + "id": 100, + "load_balance": {"backend": [{"ip": "192.168.1.10"}]}, + "translation": {"address": "masquerade"}, + }, + ] + result = _rules_to_device(rules) + self.assertEqual(result["100"]["load_balance"]["backend"], {"192.168.1.10": {}}) + + def test_nat64_translation_pool_reshaped_to_tag_node(self): + rules = [ + {"id": 10, "translation": {"pool": [{"id": 1, "address": "192.168.100.10"}]}}, + ] + result = _rules_to_device(rules) + self.assertEqual(result["10"]["translation"]["pool"], {"1": {"address": "192.168.100.10"}}) + + +class TestRulesFromDevice(unittest.TestCase): + def test_simple_rule(self): + raw = { + "100": { + "outbound-interface": {"name": "eth0"}, + "translation": {"address": "masquerade"}, + }, + } + result = _rules_from_device(raw) + self.assertEqual(len(result), 1) + self.assertEqual(result[0]["id"], 100) + self.assertEqual(result[0]["outbound_interface"]["name"], "eth0") + + def test_rules_sorted_by_id(self): + raw = { + "200": {"translation": {"address": "masquerade"}}, + "100": {"translation": {"address": "masquerade"}}, + } + result = _rules_from_device(raw) + self.assertEqual(result[0]["id"], 100) + self.assertEqual(result[1]["id"], 200) + + def test_presence_node_becomes_bool(self): + raw = {"100": {"log": {}, "translation": {"address": "masquerade"}}} + result = _rules_from_device(raw) + self.assertTrue(result[0]["log"]) + + def test_static_inbound_interface_string(self): + raw = { + "300": { + "inbound-interface": "eth0", + "destination": {"address": "198.51.100.20"}, + "translation": {"address": "192.168.1.20"}, + }, + } + result = _rules_from_device(raw) + self.assertEqual(result[0]["inbound_interface"], "eth0") + + def test_load_balance_hash_single_value_collapse(self): + """The device can collapse a single-value multi-leaf to a bare + string; this must come back as a 1-element list, not a string, + to match the field's real (list) type.""" + raw = {"100": {"load_balance": {}, "load-balance": {"hash": "random"}}} + # (duplicate key above is just illustrating intent; real call:) + raw = {"100": {"load-balance": {"hash": "random"}}} + result = _rules_from_device(raw) + self.assertEqual(result[0]["load_balance"]["hash"], ["random"]) + + def test_load_balance_backend_from_tag_node(self): + raw = {"100": {"load-balance": {"backend": {"192.168.1.10": {"weight": "50"}}}}} + result = _rules_from_device(raw) + self.assertEqual( + result[0]["load_balance"]["backend"], + [{"ip": "192.168.1.10", "weight": 50}], + ) + + def test_nat64_pool_from_tag_node(self): + raw = {"10": {"translation": {"pool": {"1": {"address": "192.168.100.10"}}}}} + result = _rules_from_device(raw) + self.assertEqual(result[0]["translation"]["pool"], [{"id": 1, "address": "192.168.100.10"}]) + + def test_empty_returns_empty(self): + self.assertEqual(_rules_from_device({}), []) + self.assertEqual(_rules_from_device(None), []) + + +class TestCgnat(unittest.TestCase): + """The core bug-fix area: external pool range is a genuine tag node + (nested "seq" leaf), internal pool range is a plain multi-value leaf + -- confirmed against vyos-1x schema, and previously conflated.""" + + def test_external_pool_range_is_tag_node_with_seq(self): + cgnat = { + "pool": { + "external": [ + {"name": "EXT1", "range": [{"value": "203.0.113.1-203.0.113.10", "seq": 1}]}, + ], + }, + } + result = _cgnat_to_device(cgnat) + self.assertEqual( + result["pool"]["external"]["EXT1"]["range"], + {"203.0.113.1-203.0.113.10": {"seq": 1}}, + ) + + def test_external_pool_range_without_seq(self): + cgnat = {"pool": {"external": [{"name": "EXT1", "range": [{"value": "203.0.113.1-.10"}]}]}} + result = _cgnat_to_device(cgnat) + self.assertEqual(result["pool"]["external"]["EXT1"]["range"], {"203.0.113.1-.10": {}}) + + def test_internal_pool_range_stays_a_plain_list(self): + cgnat = {"pool": {"internal": [{"name": "INT1", "range": ["10.0.0.0/24", "10.0.1.0/24"]}]}} + result = _cgnat_to_device(cgnat) + self.assertEqual( + result["pool"]["internal"]["INT1"]["range"], + ["10.0.0.0/24", "10.0.1.0/24"], + ) + + def test_internal_pool_multi_value_range_from_device_not_dropped(self): + """Regression test for the confirmed data-loss bug: the previous + implementation only checked isinstance(str)/isinstance(dict) for + internal pool range and silently dropped it whenever the device + returned the real shape for >1 value -- a plain list.""" + raw = {"pool": {"internal": {"INT1": {"range": ["10.0.0.0/24", "10.0.1.0/24"]}}}} + result = _cgnat_from_device(raw) + pool = result["pool"]["internal"][0] + self.assertEqual(pool["range"], ["10.0.0.0/24", "10.0.1.0/24"]) + + def test_internal_pool_single_value_range_collapse(self): + raw = {"pool": {"internal": {"INT1": {"range": "10.0.2.0/24"}}}} + result = _cgnat_from_device(raw) + self.assertEqual(result["pool"]["internal"][0]["range"], ["10.0.2.0/24"]) + + def test_external_pool_range_from_device_with_seq(self): + raw = {"pool": {"external": {"EXT1": {"range": {"203.0.113.1-.10": {"seq": "1"}}}}}} + result = _cgnat_from_device(raw) + rng = result["pool"]["external"][0]["range"] + self.assertEqual(rng, [{"value": "203.0.113.1-.10", "seq": 1}]) + + def test_log_allocation_generic_presence(self): + result = _cgnat_to_device({"log_allocation": True}) + self.assertEqual(result["log_allocation"], {}) + + def test_cgnat_rule_generic(self): + cgnat = {"rule": [{"id": 1, "destination": {"group": {"address_group": "CGNAT-DST"}}}]} + result = _cgnat_to_device(cgnat) + self.assertEqual( + result["rule"]["1"]["destination"]["group"]["address_group"], + "CGNAT-DST", + ) + + def test_empty(self): + self.assertEqual(_cgnat_to_device({}), {}) + self.assertEqual(_cgnat_from_device({}), {}) + + +class TestWantToDevice(unittest.TestCase): + def test_empty(self): + self.assertEqual(_want_to_device({}), {}) + self.assertEqual(_want_to_device(None), {}) + + def test_source_nat(self): + config = { + "nat": { + "source": { + "rule": [ + { + "id": 100, + "outbound_interface": {"name": "eth0"}, + "translation": {"address": "masquerade"}, + }, + ], + }, + }, + } + result = _want_to_device(config) + self.assertIn("100", result["nat"]["source"]["rule"]) + + def test_nat64_pools(self): + config = { + "nat64": { + "source": { + "rule": [ + { + "id": 10, + "translation": {"pool": [{"id": 1, "address": "192.168.100.10"}]}, + }, + ], + }, + }, + } + result = _want_to_device(config) + rule = result["nat64"]["source"]["rule"]["10"] + self.assertIn("1", rule["translation"]["pool"]) + + def test_nat66(self): + config = { + "nat66": { + "source": { + "rule": [{"id": 10, "outbound_interface": {"name": "eth0"}}], + }, + }, + } + result = _want_to_device(config) + self.assertIn("10", result["nat66"]["source"]["rule"]) + + def test_nat66_destination_and_source_via_dispatch_table(self): + """nat66 has no cgnat, and only destination/source (no static) -- + exercised via _NAT_TYPE_SECTIONS, not hand-written per-type + blocks.""" + config = { + "nat66": { + "destination": {"rule": [{"id": 10, "protocol": "tcp"}]}, + "source": {"rule": [{"id": 20}]}, + }, + } + result = _want_to_device(config) + self.assertIn("10", result["nat66"]["destination"]["rule"]) + self.assertIn("20", result["nat66"]["source"]["rule"]) + self.assertNotIn("cgnat", result["nat66"]) + + def test_cgnat_in_want(self): + config = {"nat": {"cgnat": {"log_allocation": True}}} + result = _want_to_device(config) + self.assertEqual(result["nat"]["cgnat"]["log_allocation"], {}) + + +class TestDeviceToArgspec(unittest.TestCase): + def test_empty(self): + self.assertEqual(_device_to_argspec({}), {}) + self.assertEqual(_device_to_argspec(None), {}) + + def test_source_rule(self): + raw = {"nat": {"source": {"rule": {"100": {"translation": {"address": "masquerade"}}}}}} + result = _device_to_argspec(raw) + self.assertEqual(result["nat"]["source"]["rule"][0]["id"], 100) + + def test_nat64_pools_parsed(self): + raw = { + "nat64": { + "source": { + "rule": {"10": {"translation": {"pool": {"1": {"address": "192.168.100.10"}}}}}, + }, + }, + } + result = _device_to_argspec(raw) + pools = result["nat64"]["source"]["rule"][0]["translation"]["pool"] + self.assertEqual(pools[0]["id"], 1) + + def test_static_inbound_interface_string(self): + raw = {"nat": {"static": {"rule": {"300": {"inbound-interface": "eth0"}}}}} + result = _device_to_argspec(raw) + self.assertEqual(result["nat"]["static"]["rule"][0]["inbound_interface"], "eth0") + + +class TestDeviceToArgspecFixture(unittest.TestCase): + def setUp(self): + self.fixture = _load_nat_fixture() + + def test_source_rules_parsed(self): + result = _device_to_argspec(self.fixture) + ids = [r["id"] for r in result["nat"]["source"]["rule"]] + self.assertIn(100, ids) + self.assertIn(101, ids) + + def test_destination_rule_parsed(self): + result = _device_to_argspec(self.fixture) + rule = result["nat"]["destination"]["rule"][0] + self.assertEqual(rule["id"], 200) + self.assertEqual(rule["destination"]["port"], "80") + + def test_static_rule_parsed(self): + result = _device_to_argspec(self.fixture) + rule = result["nat"]["static"]["rule"][0] + self.assertEqual(rule["inbound_interface"], "eth0") + + def test_nat64_pools_parsed(self): + result = _device_to_argspec(self.fixture) + pools = result["nat64"]["source"]["rule"][0]["translation"]["pool"] + self.assertEqual(pools[0]["port"], "1-65535") + + def test_description_parsed(self): + result = _device_to_argspec(self.fixture) + rule100 = next(r for r in result["nat"]["source"]["rule"] if r["id"] == 100) + self.assertEqual(rule100["description"], "Source rule 100") + + def test_hash_single_value_collapse_from_fixture(self): + result = _device_to_argspec(self.fixture) + rule100 = next(r for r in result["nat"]["source"]["rule"] if r["id"] == 100) + self.assertEqual(rule100["load_balance"]["hash"], ["random"]) + + def test_backend_from_fixture(self): + result = _device_to_argspec(self.fixture) + rule100 = next(r for r in result["nat"]["source"]["rule"] if r["id"] == 100) + backends = {b["ip"]: b.get("weight") for b in rule100["load_balance"]["backend"]} + self.assertEqual(backends["192.168.1.10"], 50) + self.assertEqual(backends["192.168.1.11"], None) + + def test_cgnat_internal_pool_range_not_dropped(self): + """The actual regression this whole refactor was triggered by.""" + result = _device_to_argspec(self.fixture) + pool = result["nat"]["cgnat"]["pool"]["internal"][0] + self.assertEqual(pool["range"], ["10.0.0.0/24", "10.0.1.0/24"]) + + def test_cgnat_external_pool_range_with_seq(self): + result = _device_to_argspec(self.fixture) + pool = result["nat"]["cgnat"]["pool"]["external"][0] + self.assertEqual(pool["range"], [{"value": "203.0.113.1-203.0.113.10", "seq": 1}]) + + def test_cgnat_rule_parsed(self): + result = _device_to_argspec(self.fixture) + rule = result["nat"]["cgnat"]["rule"][0] + self.assertEqual(rule["destination"]["group"]["address_group"], "CGNAT-DST") + + +class TestDictOpNat(unittest.TestCase): + """End-to-end command generation, exactly as main() calls it.""" + + def test_merged_adds_source_rule(self): + want = _want_to_device( + { + "nat": { + "source": { + "rule": [ + { + "id": 100, + "outbound_interface": {"name": "eth0"}, + "translation": {"address": "masquerade"}, + }, + ], + }, + }, + }, + ) + cmds = dict_op(want.get("nat", {}), {}, ["nat"], op="set") + paths = [c[1] for c in cmds] + self.assertIn(["nat", "source", "rule", "100", "outbound-interface", "name", "eth0"], paths) + self.assertIn( + ["nat", "source", "rule", "100", "translation", "address", "masquerade"], + paths, + ) + + def test_merged_idempotent_against_fixture(self): + fixture = _load_nat_fixture() + have = _device_to_argspec(fixture) + want = _want_to_device({"nat": have.get("nat", {})}).get("nat", {}) + norm_have = _normalize_nat_have(fixture, "nat") + cmds = dict_op(want, norm_have, ["nat"], op="set") + self.assertEqual(cmds, []) + + def test_nat64_idempotent_against_fixture(self): + fixture = _load_nat_fixture() + have = _device_to_argspec(fixture) + want = _want_to_device({"nat64": have.get("nat64", {})}).get("nat64", {}) + norm_have = _normalize_nat_have(fixture, "nat64") + cmds = dict_op(want, norm_have, ["nat64"], op="set") + self.assertEqual(cmds, []) + + def test_cgnat_idempotent_against_fixture_including_ranges(self): + """The real proof the bug is fixed: idempotency now holds even + though it involves both the tag-node (external) and plain-list + (internal) range shapes at once.""" + fixture = _load_nat_fixture() + have = _device_to_argspec(fixture) + want = _want_to_device({"nat": have.get("nat", {})}).get("nat", {}) + norm_have = _normalize_nat_have(fixture, "nat") + cmds = dict_op(want, norm_have, ["nat"], op="set") + self.assertEqual(cmds, []) + + def test_replaced_purges_stale_internal_range_member(self): + """The internal-pool range being a plain list means removing a + member under 'replaced' relies on dict_op's list-purge handling + (fixed earlier this session) -- confirmed it applies here too.""" + raw_have = { + "cgnat": {"pool": {"internal": {"INT1": {"range": ["10.0.0.0/24", "10.0.1.0/24"]}}}}, + } + want = _want_to_device( + { + "nat": { + "cgnat": {"pool": {"internal": [{"name": "INT1", "range": ["10.0.0.0/24"]}]}}, + }, + }, + )["nat"] + norm_have = _normalize_nat_have({"nat": raw_have}, "nat") + cmds = dict_op(want, norm_have, ["nat"], op="purge") + self.assertIn( + ("delete", ["nat", "cgnat", "pool", "internal", "INT1", "range", "10.0.1.0/24"]), + cmds, + ) + + def test_overridden_deletes_entire_omitted_section(self): + """overridden is full-model: a section entirely omitted from + want (not just a rule within it) must be deleted, via the same + single dict_op purge call main() uses -- no manual section-scan + loop needed.""" + raw_have = { + "destination": {"rule": {"200": {"protocol": "tcp"}}}, + "source": {"rule": {"100": {}}}, + } + nat_want = _want_to_device( + {"nat": {"source": {"rule": [{"id": 100}]}}}, + )["nat"] + norm_have = _normalize_nat_have({"nat": raw_have}, "nat") + cmds = dict_op(nat_want, norm_have, ["nat"], op="purge") + self.assertIn(("delete", ["nat", "destination"]), cmds) + self.assertTrue(all(c[1] != ["nat", "source"] for c in cmds)) + + def test_overridden_full_wipe_deletes_each_section_individually(self): + """Empty want under overridden purges every present section -- + granular per-section deletes, not one blanket delete of the + whole nat_type (that distinction only matters for vyos_bgp_global, + where system-as's device-model constraint forces atomicity; NAT + has no equivalent cross-field constraint).""" + raw_have = {"destination": {"rule": {"200": {}}}, "source": {"rule": {"100": {}}}} + norm_have = _normalize_nat_have({"nat": raw_have}, "nat") + cmds = dict_op({}, norm_have, ["nat"], op="purge") + self.assertIn(("delete", ["nat", "destination"]), cmds) + self.assertIn(("delete", ["nat", "source"]), cmds) + + +if __name__ == "__main__": + unittest.main() |
