path: root/tests/unit/modules/test_vyos_ha.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_ha import (
    _device_to_argspec,
    _from_device_generic,
    _normalize_groups,
    _normalize_have,
    _to_device_generic,
    _want_to_device,
)

from .base import load_fixture


_BASE = ["high-availability"]


class TestToDeviceGeneric(unittest.TestCase):
    """Test generic argspec->device converter — no field-name knowledge."""

    def test_snake_to_kebab(self):
        result = _to_device_generic({"startup_delay": 30})
        self.assertIn("startup-delay", result)
        self.assertEqual(result["startup-delay"], "30")

    def test_bool_true_becomes_presence_node(self):
        result = _to_device_generic({"disable": True})
        self.assertEqual(result["disable"], {})

    def test_bool_false_skipped(self):
        result = _to_device_generic({"no_preempt": False})
        self.assertNotIn("no-preempt", result)

    def test_none_skipped(self):
        result = _to_device_generic({"address": None, "vrid": 20})
        self.assertNotIn("address", result)
        self.assertEqual(result["vrid"], "20")

    def test_nested_dict_recurses(self):
        result = _to_device_generic({"garp": {"master_repeat": 6}})
        self.assertEqual(result["garp"]["master-repeat"], "6")

    def test_list_in_tag_node_fields(self):
        result = _to_device_generic(
            {"address": ["192.168.1.1/24", "192.168.1.2/24"]},
            tag_node_fields={"address"},
        )
        self.assertIn("192.168.1.1/24", result["address"])
        self.assertIn("192.168.1.2/24", result["address"])
        self.assertEqual(result["address"]["192.168.1.1/24"], {})

    def test_list_not_in_tag_node_fields_skipped(self):
        result = _to_device_generic({"address": ["192.168.1.1/24"]})
        self.assertNotIn("address", result)

    def test_empty_dict(self):
        self.assertEqual(_to_device_generic({}), {})
        self.assertEqual(_to_device_generic(None), {})


class TestFromDeviceGeneric(unittest.TestCase):
    """Test generic device->argspec converter — no field-name knowledge."""

    def test_kebab_to_snake(self):
        result = _from_device_generic({"startup-delay": "30"})
        self.assertIn("startup_delay", result)
        self.assertEqual(result["startup_delay"], 30)

    def test_empty_dict_becomes_true(self):
        result = _from_device_generic({"disable": {}, "no-preempt": {}})
        self.assertTrue(result["disable"])
        self.assertTrue(result["no_preempt"])

    def test_numeric_string_cast_to_int(self):
        result = _from_device_generic({"vrid": "20", "priority": "100"})
        self.assertEqual(result["vrid"], 20)
        self.assertEqual(result["priority"], 100)

    def test_non_numeric_string_unchanged(self):
        result = _from_device_generic({"interface": "eth0", "peer-address": "192.168.1.1"})
        self.assertEqual(result["interface"], "eth0")
        self.assertEqual(result["peer_address"], "192.168.1.1")

    def test_nested_dict_recurses(self):
        result = _from_device_generic({"garp": {"master-repeat": "6"}})
        self.assertEqual(result["garp"]["master_repeat"], 6)

    def test_empty(self):
        self.assertEqual(_from_device_generic({}), {})
        self.assertEqual(_from_device_generic(None), {})


class TestWantToDevice(unittest.TestCase):
    """Test argspec->device structural reshaping."""

    def test_empty(self):
        self.assertEqual(_want_to_device({}), {})
        self.assertEqual(_want_to_device(None), {})

    def test_disable(self):
        result = _want_to_device({"disable": True})
        self.assertEqual(result["disable"], {})

    def test_virtual_server_keyed_by_name(self):
        config = {"virtual_servers": [{"name": "s1", "address": "10.0.0.1", "port": 80}]}
        result = _want_to_device(config)
        self.assertIn("s1", result["virtual-server"])
        self.assertEqual(result["virtual-server"]["s1"]["address"], "10.0.0.1")
        self.assertEqual(result["virtual-server"]["s1"]["port"], "80")

    def test_real_server_keyed_by_address(self):
        config = {
            "virtual_servers": [
                {
                    "name": "s1",
                    "port": 80,
                    "protocol": "tcp",
                    "real_server": [{"address": "10.0.0.2", "port": 8080}],
                },
            ],
        }
        result = _want_to_device(config)
        rs = result["virtual-server"]["s1"]["real-server"]
        self.assertIn("10.0.0.2", rs)
        self.assertEqual(rs["10.0.0.2"]["port"], "8080")

    def test_health_check_script_nested(self):
        config = {
            "virtual_servers": [
                {
                    "name": "s1",
                    "port": 80,
                    "protocol": "tcp",
                    "real_server": [
                        {
                            "address": "10.0.0.2",
                            "health_check_script": "/check.sh",
                        },
                    ],
                },
            ],
        }
        result = _want_to_device(config)
        rs = result["virtual-server"]["s1"]["real-server"]["10.0.0.2"]
        self.assertEqual(rs["health-check"]["script"], "/check.sh")

    def test_vrrp_global_parameters(self):
        config = {
            "vrrp": {
                "global_parameters": {
                    "startup_delay": 30,
                    "garp": {"master_repeat": 6},
                },
            },
        }
        result = _want_to_device(config)
        gp = result["vrrp"]["global-parameters"]
        self.assertEqual(gp["startup-delay"], "30")
        self.assertEqual(gp["garp"]["master-repeat"], "6")

    def test_snmp_enabled_becomes_presence_node(self):
        config = {"vrrp": {"snmp": "enabled"}}
        result = _want_to_device(config)
        self.assertEqual(result["vrrp"]["snmp"], {})

    def test_snmp_disabled_not_in_want(self):
        config = {"vrrp": {"snmp": "disabled"}}
        result = _want_to_device(config)
        self.assertNotIn("snmp", result.get("vrrp", {}))

    def test_group_keyed_by_name(self):
        config = {
            "vrrp": {
                "groups": [
                    {
                        "name": "g1",
                        "vrid": 20,
                        "interface": "eth0",
                        "priority": 100,
                    },
                ],
            },
        }
        result = _want_to_device(config)
        g = result["vrrp"]["group"]["g1"]
        self.assertEqual(g["vrid"], "20")
        self.assertEqual(g["interface"], "eth0")

    def test_group_address_tag_node(self):
        config = {
            "vrrp": {
                "groups": [
                    {
                        "name": "g1",
                        "vrid": 1,
                        "address": ["192.168.1.1/24", "192.168.1.2/24"],
                    },
                ],
            },
        }
        result = _want_to_device(config)
        addr = result["vrrp"]["group"]["g1"]["address"]
        self.assertIn("192.168.1.1/24", addr)
        self.assertEqual(addr["192.168.1.1/24"], {})

    def test_group_bool_fields(self):
        config = {
            "vrrp": {
                "groups": [
                    {
                        "name": "g1",
                        "vrid": 1,
                        "disable": True,
                        "no_preempt": True,
                        "rfc3768_compatibility": False,
                    },
                ],
            },
        }
        result = _want_to_device(config)
        g = result["vrrp"]["group"]["g1"]
        self.assertEqual(g["disable"], {})
        self.assertEqual(g["no-preempt"], {})
        self.assertNotIn("rfc3768-compatibility", g)

    def test_group_track_interface_tag_node(self):
        config = {
            "vrrp": {
                "groups": [
                    {
                        "name": "g1",
                        "vrid": 1,
                        "track": {"interface": ["eth0", "eth1"]},
                    },
                ],
            },
        }
        result = _want_to_device(config)
        track = result["vrrp"]["group"]["g1"]["track"]
        self.assertIn("eth0", track["interface"])
        self.assertIn("eth1", track["interface"])

    def test_sync_group_member_tag_node(self):
        config = {"vrrp": {"sync_groups": [{"name": "sg1", "member": ["g1", "g2"]}]}}
        result = _want_to_device(config)
        member = result["vrrp"]["sync-group"]["sg1"]["member"]
        self.assertIn("g1", member)
        self.assertIn("g2", member)


class TestDeviceToArgspec(unittest.TestCase):
    """Test device->argspec structural reshaping."""

    def test_empty(self):
        self.assertEqual(_device_to_argspec({}), {})
        self.assertEqual(_device_to_argspec(None), {})

    def test_disable_presence_node(self):
        result = _device_to_argspec({"disable": {}})
        self.assertTrue(result["disable"])

    def test_virtual_server_list(self):
        raw = {"virtual-server": {"s1": {"address": "10.0.0.1", "port": "80"}}}
        result = _device_to_argspec(raw)
        self.assertEqual(len(result["virtual_servers"]), 1)
        self.assertEqual(result["virtual_servers"][0]["name"], "s1")
        self.assertEqual(result["virtual_servers"][0]["port"], 80)

    def test_real_server_list(self):
        raw = {
            "virtual-server": {
                "s1": {
                    "port": "80",
                    "real-server": {"10.0.0.2": {"port": "8080"}},
                },
            },
        }
        result = _device_to_argspec(raw)
        rs = result["virtual_servers"][0]["real_server"]
        self.assertEqual(rs[0]["address"], "10.0.0.2")
        self.assertEqual(rs[0]["port"], 8080)

    def test_health_check_script_extracted(self):
        raw = {
            "virtual-server": {
                "s1": {
                    "port": "80",
                    "real-server": {"10.0.0.2": {"health-check": {"script": "/check.sh"}}},
                },
            },
        }
        result = _device_to_argspec(raw)
        rs = result["virtual_servers"][0]["real_server"][0]
        self.assertEqual(rs["health_check_script"], "/check.sh")

    def test_snmp_presence_node_becomes_enabled(self):
        raw = {"vrrp": {"snmp": {}}}
        result = _device_to_argspec(raw)
        self.assertEqual(result["vrrp"]["snmp"], "enabled")

    def test_global_parameters_converted(self):
        raw = {
            "vrrp": {
                "global-parameters": {
                    "startup-delay": "30",
                    "garp": {"master-repeat": "6"},
                },
            },
        }
        result = _device_to_argspec(raw)
        gp = result["vrrp"]["global_parameters"]
        self.assertEqual(gp["startup_delay"], 30)
        self.assertEqual(gp["garp"]["master_repeat"], 6)

    def test_group_list_sorted(self):
        raw = {"vrrp": {"group": {"z1": {"vrid": "2"}, "a1": {"vrid": "1"}}}}
        result = _device_to_argspec(raw)
        names = [g["name"] for g in result["vrrp"]["groups"]]
        self.assertEqual(names[0], "a1")
        self.assertEqual(names[1], "z1")

    def test_group_address_tag_node_to_list(self):
        raw = {"vrrp": {"group": {"g1": {"address": {"192.168.1.1/24": {}}}}}}
        result = _device_to_argspec(raw)
        self.assertEqual(result["vrrp"]["groups"][0]["address"], ["192.168.1.1/24"])

    def test_group_single_address_string_to_list(self):
        raw = {"vrrp": {"group": {"g1": {"address": "192.168.1.1/24"}}}}
        result = _device_to_argspec(raw)
        self.assertEqual(result["vrrp"]["groups"][0]["address"], ["192.168.1.1/24"])

    def test_group_bool_presence_nodes(self):
        raw = {"vrrp": {"group": {"g1": {"disable": {}, "no-preempt": {}}}}}
        result = _device_to_argspec(raw)
        g = result["vrrp"]["groups"][0]
        self.assertTrue(g["disable"])
        self.assertTrue(g["no_preempt"])

    def test_sync_group_member_string_to_list(self):
        raw = {"vrrp": {"sync-group": {"sg1": {"member": "g1"}}}}
        result = _device_to_argspec(raw)
        self.assertEqual(result["vrrp"]["sync_groups"][0]["member"], ["g1"])

    def test_sync_group_member_dict_to_list(self):
        raw = {"vrrp": {"sync-group": {"sg1": {"member": {"g1": {}, "g2": {}}}}}}
        result = _device_to_argspec(raw)
        self.assertEqual(sorted(result["vrrp"]["sync_groups"][0]["member"]), ["g1", "g2"])


class TestFixture(unittest.TestCase):
    """Test against real fixture."""

    def setUp(self):
        self.fixture = load_fixture("ha_running.json")

    def test_disable_parsed(self):
        result = _device_to_argspec(self.fixture)
        self.assertTrue(result["disable"])

    def test_virtual_server_parsed(self):
        result = _device_to_argspec(self.fixture)
        vs = result["virtual_servers"][0]
        self.assertEqual(vs["name"], "s1")
        self.assertEqual(vs["address"], "10.10.10.5")
        self.assertEqual(vs["port"], 80)
        self.assertEqual(vs["real_server"][0]["address"], "10.10.50.2")
        self.assertEqual(vs["real_server"][0]["port"], 8443)

    def test_global_parameters_parsed(self):
        result = _device_to_argspec(self.fixture)
        gp = result["vrrp"]["global_parameters"]
        self.assertEqual(gp["startup_delay"], 30)
        self.assertEqual(gp["garp"]["master_repeat"], 6)

    def test_snmp_parsed(self):
        result = _device_to_argspec(self.fixture)
        self.assertEqual(result["vrrp"]["snmp"], "enabled")

    def test_groups_parsed(self):
        result = _device_to_argspec(self.fixture)
        groups = {g["name"]: g for g in result["vrrp"]["groups"]}
        self.assertIn("g1", groups)
        self.assertIn("g2", groups)
        self.assertEqual(groups["g1"]["interface"], "eth0")
        self.assertEqual(groups["g1"]["vrid"], 20)
        self.assertIn("192.168.1.100/24", groups["g1"]["address"])
        self.assertTrue(groups["g1"]["no_preempt"])
        # g2 single address string -> list
        self.assertEqual(groups["g2"]["address"], ["192.168.2.100/24"])

    def test_sync_group_parsed(self):
        result = _device_to_argspec(self.fixture)
        sg = result["vrrp"]["sync_groups"][0]
        self.assertEqual(sg["name"], "sg1")
        self.assertEqual(sg["member"], ["g1"])
        self.assertEqual(sg["health_check"]["failure_count"], 5)
        self.assertEqual(sg["health_check"]["ping"], "192.168.1.1")


class TestDictOpHa(unittest.TestCase):
    """Test dict_op with HA device shapes."""

    def test_merged_adds_vrrp_group(self):
        want = _want_to_device(
            {
                "vrrp": {
                    "groups": [
                        {"name": "g1", "vrid": 20, "interface": "eth0", "priority": 100},
                    ],
                },
            },
        )
        cmds = dict_op(want, {}, _BASE, op="set")
        paths = [c[1] for c in cmds]
        self.assertIn(_BASE + ["vrrp", "group", "g1", "vrid", "20"], paths)
        self.assertIn(_BASE + ["vrrp", "group", "g1", "interface", "eth0"], paths)

    def test_merged_idempotent(self):
        config = {
            "vrrp": {
                "groups": [
                    {"name": "g1", "vrid": 20, "interface": "eth0"},
                ],
            },
        }
        want = _want_to_device(config)
        norm_have = _normalize_have(
            {
                "vrrp": {
                    "group": {
                        "g1": {"vrid": "20", "interface": "eth0"},
                    },
                },
            },
        )
        cmds = dict_op(want, norm_have, _BASE, op="set")
        self.assertEqual(cmds, [])

    def test_purge_removes_extra_group(self):
        want = _want_to_device(
            {
                "vrrp": {
                    "groups": [
                        {"name": "g1", "vrid": 20, "interface": "eth0"},
                    ],
                },
            },
        )
        norm_have = _normalize_have(
            {
                "vrrp": {
                    "group": {
                        "g1": {"vrid": "20", "interface": "eth0"},
                        "g2": {"vrid": "21", "interface": "eth1"},
                    },
                },
            },
        )
        cmds = dict_op(
            want.get("vrrp", {}),
            norm_have.get("vrrp", {}),
            _BASE + ["vrrp"],
            op="purge",
        )
        paths = [c[1] for c in cmds]
        self.assertIn(_BASE + ["vrrp", "group", "g2"], paths)
        self.assertNotIn(_BASE + ["vrrp", "group", "g1"], paths)

    def test_normalize_have_member_string(self):
        raw = {"vrrp": {"sync-group": {"sg1": {"member": "g1"}}}}
        norm = _normalize_have(raw)
        self.assertIsInstance(norm["vrrp"]["sync-group"]["sg1"]["member"], dict)
        self.assertIn("g1", norm["vrrp"]["sync-group"]["sg1"]["member"])

    def test_normalize_groups_address_string(self):
        norm = {"vrrp": {"group": {"g1": {"address": "192.168.1.1/24"}}}}
        _normalize_groups(norm)
        self.assertIsInstance(norm["vrrp"]["group"]["g1"]["address"], dict)
        self.assertIn("192.168.1.1/24", norm["vrrp"]["group"]["g1"]["address"])


if __name__ == "__main__":
    unittest.main()