path: root/tests/unit/modules/test_vyos_nat.py

# -*- 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()