summaryrefslogtreecommitdiff
path: root/tests/unit/modules/test_vyos_ospfv3.py
blob: a84041bdd79913ada67c0c9f1bdee933982cc7a6 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
# -*- coding: utf-8 -*-
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_ospfv3 import (
    build_commands,
    get_running_config,
)


def load_fixture(filename):
    fixtures_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures")
    with open(os.path.join(fixtures_dir, filename)) as f:
        return json.load(f)


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


class TestVyOSOspfv3Parse(VyOSModuleTestCase):

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

    def test_parses_parameters(self):
        self.set_running_config(self.fixture)
        result = get_running_config(self.mock_vyos)
        self.assertEqual(result["parameters"]["router_id"], "192.0.2.10")

    def test_parses_redistribute(self):
        self.set_running_config(self.fixture)
        result = get_running_config(self.mock_vyos)
        route_types = [r["route_type"] for r in result["redistribute"]]
        self.assertIn("bgp", route_types)
        self.assertIn("connected", route_types)
        connected = next(r for r in result["redistribute"] if r["route_type"] == "connected")
        self.assertEqual(connected["route_map"], "RM1")

    def test_parses_areas(self):
        self.set_running_config(self.fixture)
        result = get_running_config(self.mock_vyos)
        self.assertEqual(len(result["areas"]), 2)
        area2 = next(a for a in result["areas"] if a["area_id"] == "2")
        self.assertEqual(area2["export_list"], "export1")
        self.assertEqual(area2["import_list"], "import1")
        self.assertEqual(len(area2["range"]), 2)
        not_adv = next(r for r in area2["range"] if r["address"] == "2001:db20::/32")
        self.assertTrue(not_adv["not_advertise"])

    def test_empty_config_returns_empty(self):
        self.set_running_config({})
        result = get_running_config(self.mock_vyos)
        self.assertEqual(result, {})


class TestVyOSOspfv3BuildCommands(unittest.TestCase):

    def test_deleted_with_have(self):
        have = {"parameters": {"router_id": "192.0.2.10"}}
        cmds = build_commands({}, have, "deleted")
        self.assertEqual(cmds, [("delete", ["protocols", "ospfv3"])])

    def test_deleted_without_have(self):
        cmds = build_commands({}, {}, "deleted")
        self.assertEqual(cmds, [])

    def test_merged_parameters(self):
        config = {"parameters": {"router_id": "192.0.2.10"}}
        cmds = build_commands(config, {}, "merged")
        self.assertIn(
            ("set", ["protocols", "ospfv3", "parameters", "router-id", "192.0.2.10"]),
            cmds,
        )

    def test_merged_redistribute(self):
        config = {"redistribute": [{"route_type": "bgp"}]}
        cmds = build_commands(config, {}, "merged")
        self.assertIn(
            ("set", ["protocols", "ospfv3", "redistribute", "bgp"]),
            cmds,
        )

    def test_merged_idempotent(self):
        config = {"parameters": {"router_id": "192.0.2.10"}}
        have = {"parameters": {"router_id": "192.0.2.10"}}
        cmds = build_commands(config, have, "merged")
        self.assertEqual(cmds, [])

    def test_merged_area_range(self):
        config = {
            "areas": [{"area_id": "2", "range": [{"address": "2001:db10::/32"}]}],
        }
        cmds = build_commands(config, {}, "merged")
        self.assertIn(
            ("set", ["protocols", "ospfv3", "area", "2", "range", "2001:db10::/32"]),
            cmds,
        )

    def test_replaced_idempotent(self):
        config = {"parameters": {"router_id": "192.0.2.10"}}
        have = {"parameters": {"router_id": "192.0.2.10"}}
        cmds = build_commands(config, have, "replaced")
        self.assertEqual(cmds, [])

    def test_replaced_rebuilds_on_change(self):
        config = {"parameters": {"router_id": "192.0.2.11"}}
        have = {"parameters": {"router_id": "192.0.2.10"}}
        cmds = build_commands(config, have, "replaced")
        self.assertEqual(cmds[0], ("delete", ["protocols", "ospfv3"]))
        self.assertIn(
            ("set", ["protocols", "ospfv3", "parameters", "router-id", "192.0.2.11"]),
            cmds,
        )

    def test_merged_area_export_list(self):
        config = {"areas": [{"area_id": "2", "export_list": "export1"}]}
        cmds = build_commands(config, {}, "merged")
        self.assertIn(
            ("set", ["protocols", "ospfv3", "area", "2", "export-list", "export1"]),
            cmds,
        )


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