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

# -*- coding: utf-8 -*-
from __future__ import absolute_import, division, print_function


__metaclass__ = type

import unittest

from ansible_collections.vyos.rest.plugins.modules.vyos_vrf import (
    _device_to_argspec,
    _proto_from_device,
    _proto_to_device,
    _protocols_from_device,
    build_commands,
)

from .base import load_fixture


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

_RAW_HAVE = load_fixture("vrf_running.json")

# ---------------------------------------------------------------------------
# _device_to_argspec — top-level VRF properties
# ---------------------------------------------------------------------------


class TestDeviceToArgspec(unittest.TestCase):
    def setUp(self):
        self.result = _device_to_argspec(_RAW_HAVE)

    def test_bind_to_all(self):
        self.assertTrue(self.result["bind_to_all"])

    def test_instances_count(self):
        self.assertEqual(len(self.result["instances"]), 2)

    def test_vrf1_properties(self):
        vrf1 = next(i for i in self.result["instances"] if i["name"] == "vrf1")
        self.assertEqual(vrf1["description"], "red")
        self.assertTrue(vrf1["disable"])
        self.assertEqual(vrf1["table_id"], 101)
        self.assertEqual(vrf1["vni"], 501)

    def test_vrf2_address_family(self):
        vrf2 = next(i for i in self.result["instances"] if i["name"] == "vrf2")
        afis = {af["afi"]: af for af in vrf2["address_family"]}
        self.assertIn("ipv4", afis)
        self.assertTrue(afis["ipv4"]["disable_forwarding"])
        self.assertTrue(afis["ipv4"]["nht_no_resolve_via_default"])
        self.assertIn("ipv6", afis)
        self.assertTrue(afis["ipv6"]["disable_forwarding"])
        self.assertTrue(afis["ipv6"]["nht_no_resolve_via_default"])

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


# ---------------------------------------------------------------------------
# _proto_from_device — BGP
# ---------------------------------------------------------------------------


class TestBgpFromDevice(unittest.TestCase):
    def setUp(self):
        self.raw = _RAW_HAVE["name"]["vrf1"]["protocols"]["bgp"]
        self.result = _proto_from_device(self.raw, "bgp")

    def test_system_as(self):
        self.assertEqual(self.result["system_as"], 65001)

    def test_neighbor_list(self):
        self.assertEqual(len(self.result["neighbor"]), 1)
        n = self.result["neighbor"][0]
        self.assertEqual(n["address"], "10.0.0.1")
        self.assertEqual(n["remote_as"], 65002)
        self.assertEqual(n["description"], "peer1")

    def test_empty_input(self):
        self.assertEqual(_proto_from_device({}, "bgp"), {})
        self.assertEqual(_proto_from_device(None, "bgp"), {})


# ---------------------------------------------------------------------------
# _proto_to_device + build_commands — BGP
# ---------------------------------------------------------------------------


class TestBgpToDevice(unittest.TestCase):
    def test_system_as_to_device(self):
        result = _proto_to_device({"system_as": 65001}, "bgp")
        self.assertIn("system-as", result)
        self.assertEqual(result["system-as"], 65001)

    def test_neighbor_to_device(self):
        result = _proto_to_device(
            {
                "system_as": 65001,
                "neighbor": [{"address": "10.0.0.1", "remote_as": 65002}],
            },
            "bgp",
        )
        self.assertIn("neighbor", result)
        self.assertIn("10.0.0.1", result["neighbor"])
        self.assertEqual(result["neighbor"]["10.0.0.1"]["remote-as"], 65002)

    def test_idempotent(self):
        want = _proto_from_device(_RAW_HAVE["name"]["vrf1"]["protocols"]["bgp"], "bgp")
        cmds = build_commands(
            {"instances": [{"name": "vrf1", "table_id": 101, "protocols": {"bgp": want}}]},
            _RAW_HAVE,
            "merged",
        )
        bgp_cmds = [c for c in cmds if "bgp" in str(c)]
        self.assertEqual(bgp_cmds, [])

    def test_add_neighbor(self):
        want_bgp = {
            "system_as": 65001,
            "neighbor": [
                {"address": "10.0.0.1", "remote_as": 65002, "description": "peer1"},
                {"address": "10.0.0.2", "remote_as": 65003},
            ],
        }
        cmds = build_commands(
            {"instances": [{"name": "vrf1", "table_id": 101, "protocols": {"bgp": want_bgp}}]},
            _RAW_HAVE,
            "merged",
        )
        paths = [c[1] for c in cmds]
        self.assertIn(
            [
                "vrf",
                "name",
                "vrf1",
                "protocols",
                "bgp",
                "neighbor",
                "10.0.0.2",
                "remote-as",
                "65003",
            ],
            paths,
        )
        # existing neighbor should not be re-set
        self.assertNotIn(
            [
                "vrf",
                "name",
                "vrf1",
                "protocols",
                "bgp",
                "neighbor",
                "10.0.0.1",
                "remote-as",
                "65002",
            ],
            paths,
        )


# ---------------------------------------------------------------------------
# _proto_from_device — OSPFv2
# ---------------------------------------------------------------------------


class TestOspfFromDevice(unittest.TestCase):
    def setUp(self):
        self.raw = _RAW_HAVE["name"]["vrf1"]["protocols"]["ospf"]
        self.result = _proto_from_device(self.raw, "ospf")

    def test_areas(self):
        self.assertEqual(len(self.result["areas"]), 1)
        area = self.result["areas"][0]
        self.assertEqual(area["area_id"], "0")
        self.assertIn("10.0.0.0/24", area["networks"])
        self.assertIn("172.16.0.0/24", area["networks"])

    def test_parameters(self):
        self.assertEqual(self.result["parameters"]["router_id"], "10.0.0.1")

    def test_empty_input(self):
        self.assertEqual(_proto_from_device({}, "ospf"), {})


# ---------------------------------------------------------------------------
# build_commands — OSPFv2
# ---------------------------------------------------------------------------


class TestOspfBuildCommands(unittest.TestCase):
    def test_idempotent(self):
        want = _proto_from_device(_RAW_HAVE["name"]["vrf1"]["protocols"]["ospf"], "ospf")
        cmds = build_commands(
            {"instances": [{"name": "vrf1", "table_id": 101, "protocols": {"ospf": want}}]},
            _RAW_HAVE,
            "merged",
        )
        ospf_cmds = [c for c in cmds if "ospf" in str(c)]
        self.assertEqual(ospf_cmds, [])

    def test_add_network(self):
        want_ospf = {
            "areas": [
                {"area_id": "0", "networks": ["10.0.0.0/24", "172.16.0.0/24", "192.168.0.0/24"]},
            ],
        }
        cmds = build_commands(
            {"instances": [{"name": "vrf1", "table_id": 101, "protocols": {"ospf": want_ospf}}]},
            _RAW_HAVE,
            "merged",
        )
        paths = [c[1] for c in cmds]
        self.assertIn(
            ["vrf", "name", "vrf1", "protocols", "ospf", "area", "0", "network", "192.168.0.0/24"],
            paths,
        )

    def test_add_area(self):
        want_ospf = {
            "areas": [
                {"area_id": "0", "networks": ["10.0.0.0/24", "172.16.0.0/24"]},
                {"area_id": "1", "networks": ["10.1.0.0/24"]},
            ],
        }
        cmds = build_commands(
            {"instances": [{"name": "vrf1", "table_id": 101, "protocols": {"ospf": want_ospf}}]},
            _RAW_HAVE,
            "merged",
        )
        paths = [c[1] for c in cmds]
        self.assertIn(
            ["vrf", "name", "vrf1", "protocols", "ospf", "area", "1", "network", "10.1.0.0/24"],
            paths,
        )


# ---------------------------------------------------------------------------
# _proto_from_device — static routes
# ---------------------------------------------------------------------------


class TestStaticFromDevice(unittest.TestCase):
    def setUp(self):
        self.raw = _RAW_HAVE["name"]["vrf1"]["protocols"]["static"]
        self.result = _proto_from_device(self.raw, "static")

    def test_routes(self):
        self.assertEqual(len(self.result["routes"]), 1)
        route = self.result["routes"][0]
        self.assertEqual(route["dest"], "192.168.10.0/24")
        self.assertIn("10.0.0.254", route["next_hops"])

    def test_empty_input(self):
        self.assertEqual(_proto_from_device({}, "static"), {})


# ---------------------------------------------------------------------------
# build_commands — static routes
# ---------------------------------------------------------------------------


class TestStaticBuildCommands(unittest.TestCase):
    def test_idempotent(self):
        want = _proto_from_device(_RAW_HAVE["name"]["vrf1"]["protocols"]["static"], "static")
        cmds = build_commands(
            {"instances": [{"name": "vrf1", "table_id": 101, "protocols": {"static": want}}]},
            _RAW_HAVE,
            "merged",
        )
        static_cmds = [c for c in cmds if "static" in str(c)]
        self.assertEqual(static_cmds, [])

    def test_add_route(self):
        want_static = {
            "routes": [
                {"dest": "192.168.10.0/24", "next_hops": ["10.0.0.254"]},
                {"dest": "192.168.20.0/24", "next_hops": ["10.0.0.254"]},
            ],
        }
        cmds = build_commands(
            {
                "instances": [
                    {"name": "vrf1", "table_id": 101, "protocols": {"static": want_static}},
                ],
            },
            _RAW_HAVE,
            "merged",
        )
        paths = [c[1] for c in cmds]
        self.assertIn(
            [
                "vrf",
                "name",
                "vrf1",
                "protocols",
                "static",
                "route",
                "192.168.20.0/24",
                "next-hop",
                "10.0.0.254",
            ],
            paths,
        )


# ---------------------------------------------------------------------------
# _protocols_from_device
# ---------------------------------------------------------------------------


class TestProtocolsFromDevice(unittest.TestCase):
    def test_all_protocols(self):
        raw_vrf = _RAW_HAVE["name"]["vrf1"]
        result = _protocols_from_device(raw_vrf)
        self.assertIn("bgp", result)
        self.assertIn("ospf", result)
        self.assertIn("static", result)

    def test_no_protocols(self):
        raw_vrf = _RAW_HAVE["name"]["vrf2"]
        result = _protocols_from_device(raw_vrf)
        self.assertIsNone(result)


# ---------------------------------------------------------------------------
# build_commands — top-level VRF operations
# ---------------------------------------------------------------------------


class TestBuildCommands(unittest.TestCase):
    def test_merged_new_vrf(self):
        config = {"instances": [{"name": "vrf3", "table_id": 200, "vni": 2000}]}
        cmds = build_commands(config, _RAW_HAVE, "merged")
        paths = [c[1] for c in cmds]
        self.assertIn(["vrf", "name", "vrf3", "table", "200"], paths)
        self.assertIn(["vrf", "name", "vrf3", "vni", "2000"], paths)

    def test_merged_idempotent(self):
        config = {
            "bind_to_all": True,
            "instances": [
                {"name": "vrf1", "description": "red", "table_id": 101, "vni": 501},
            ],
        }
        cmds = build_commands(config, _RAW_HAVE, "merged")
        self.assertEqual(cmds, [])

    def test_deleted_specific_vrf(self):
        config = {"instances": [{"name": "vrf1"}]}
        cmds = build_commands(config, _RAW_HAVE, "deleted")
        self.assertIn(("delete", ["vrf", "name", "vrf1"]), cmds)
        self.assertNotIn(("delete", ["vrf", "name", "vrf2"]), cmds)

    def test_deleted_all(self):
        cmds = build_commands({}, _RAW_HAVE, "deleted")
        self.assertIn(("delete", ["vrf"]), cmds)

    def test_overridden_removes_extra_vrf(self):
        config = {"instances": [{"name": "vrf1", "table_id": 101}]}
        cmds = build_commands(config, _RAW_HAVE, "overridden")
        paths = [c[1] for c in cmds]
        self.assertIn(["vrf", "name", "vrf2"], paths)

    def test_merged_does_not_delete_unreferenced_vrf(self):
        config = {"instances": [{"name": "vrf1", "table_id": 101}]}
        cmds = build_commands(config, _RAW_HAVE, "merged")
        paths = [c[1] for c in cmds]
        self.assertNotIn(["vrf", "name", "vrf2"], paths)


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