Passed
Pull Request — master (#84)
by Vinicius
05:23 queued 01:50
created

build.tests.unit.test_topo_controller   A

Complexity

Total Complexity 31

Size/Duplication

Total Lines 303
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
eloc 206
dl 0
loc 303
rs 9.92
c 0
b 0
f 0
wmc 31

29 Methods

Rating   Name   Duplication   Size   Complexity  
A TestTopoController.test_enable_switch() 0 8 1
A TestTopoController.test_add_link_metadata() 0 10 1
A TestTopoController.test_add_switch_metadata() 0 9 1
A TestTopoController.test_boostrap_index_existent() 0 13 1
A TestTopoController.test_get_topology() 0 7 1
A TestTopoController.test_get_links() 0 7 1
A TestTopoController.test_enable_link() 0 8 1
A TestTopoController.test_enable_interface() 0 8 1
A TestTopoController.test_get_interfaces() 0 10 1
A TestTopoController.test_bulk_upsert_interface_details() 0 13 2
A TestTopoController.test_delete_switch_metadata() 0 9 1
A TestTopoController.test_deactivate_interface() 0 8 1
A TestTopoController.test_delete_interface_metadata_key() 0 9 1
A TestTopoController.test_disable_interface() 0 8 1
A TestTopoController.test_add_interface_metadata() 0 9 1
A TestTopoController.test_disable_switch() 0 8 1
A TestTopoController.test_boostrap_index() 0 10 1
A TestTopoController.test_activate_interface() 0 8 1
A TestTopoController.test_upsert_switch() 0 10 2
A TestTopoController.test_get_switches() 0 8 1
A TestTopoController.test_disable_interface_lldp() 0 8 1
A TestTopoController.test_get_interfaces_details() 0 6 1
A TestTopoController.test_enable_interface_lldp() 0 8 1
A TestTopoController.test_disable_link() 0 8 1
A TestTopoController.test_delete_link_metadata_key() 0 9 1
A TestTopoController.test_boostrap_indexes() 0 11 1
A TestTopoController.test_deactivate_switch() 0 8 1
A TestTopoController.test_upsert_link() 0 12 1
A TestTopoController.setUp() 0 6 1
1
"""Module to TopoController."""
2
3
from unittest import TestCase
4
from unittest.mock import MagicMock, patch
5
6
from pymongo.operations import UpdateOne
7
8
from napps.kytos.topology.controllers import TopoController
9
from napps.kytos.topology.db.client import bootstrap_index
10
from napps.kytos.topology.db.models import LinkDoc, SwitchDoc
11
12
13
class TestTopoController(TestCase):  # pylint: disable=too-many-public-methods
14
    """Test the Main class."""
15
16
    def setUp(self) -> None:
17
        """Execute steps before each tests."""
18
        self.topo = TopoController(MagicMock())
19
        self.dpid = "00:00:00:00:00:00:00:01"
20
        self.interface_id = f"{self.dpid}:1"
21
        self.link_id = "some_id"
22
23
    @patch("napps.kytos.topology.controllers.bootstrap_index")
24
    def test_boostrap_indexes(self, mock) -> None:
25
        """Test_boostrap_indexes."""
26
        self.topo.bootstrap_indexes()
27
28
        expected_indexes = [
29
            ("switches", "interfaces.id", 1),
30
            ("links", "endpoints.id", 1),
31
        ]
32
        indexes = [(v[0][1], v[0][2], v[0][3]) for v in mock.call_args_list]
33
        assert expected_indexes == indexes
34
35
    @staticmethod
36
    def test_boostrap_index_existent() -> None:
37
        """test_boostrap_index.
38
39
        This test will be moved to kytos core onde db client moves too.
40
        """
41
        db_mock = MagicMock()
42
        db_mock["napps"].index_information.return_value = {
43
            "_id_": {"v": 2, "key": [("_id", 1)]},
44
            "interfaces.id_1": {"v": 2, "key": [("interfaces.id", 1)]},
45
        }
46
        bootstrap_index(db_mock, "napps", "interfaces.id", 1)
47
        assert db_mock["napps"].create_index.call_count == 0
48
49
    @staticmethod
50
    def test_boostrap_index() -> None:
51
        """test_boostrap_index.
52
53
        This test will be moved to kytos core onde db client moves too.
54
        """
55
        db_mock = MagicMock()
56
        db_mock["napps"].index_information.return_value = {}
57
        bootstrap_index(db_mock, "switches", "interfaces.id", 1)
58
        assert db_mock["napps"].create_index.call_count == 1
59
60
    def test_get_topology(self) -> None:
61
        """Test_get_topology."""
62
        self.topo.get_switches = MagicMock()
63
        self.topo.get_links = MagicMock()
64
        assert "topology" in self.topo.get_topology()
65
        assert self.topo.get_switches.call_count == 1
66
        assert self.topo.get_links.call_count == 1
67
68
    def test_get_links(self) -> None:
69
        """test_get_links."""
70
        assert "links" in self.topo.get_links()
71
        assert self.topo.db.links.aggregate.call_count == 1
72
        arg = self.topo.db.links.aggregate.call_args[0]
73
        assert arg[0] == [{"$sort": {"_id": 1}},
74
                          {"$project": LinkDoc.projection()}]
75
76
    def test_get_switches(self) -> None:
77
        """test_get_switches."""
78
        assert "switches" in self.topo.get_switches()
79
        assert self.topo.db.switches.aggregate.call_count == 1
80
        arg = self.topo.db.switches.aggregate.call_args[0]
81
        assert arg[0] == [
82
            {"$sort": {"_id": 1}},
83
            {"$project": SwitchDoc.projection()},
84
        ]
85
86
    def test_get_interfaces(self) -> None:
87
        """test_get_interfaces."""
88
        assert "interfaces" in self.topo.get_interfaces()
89
        assert self.topo.db.switches.aggregate.call_count == 1
90
        arg = self.topo.db.switches.aggregate.call_args[0]
91
        assert arg[0] == [
92
            {"$sort": {"_id": 1}},
93
            {"$project": {"interfaces": 1, "_id": 0}},
94
            {"$unwind": "$interfaces"},
95
            {"$replaceRoot": {"newRoot": "$interfaces"}},
96
        ]
97
98
    def test_enable_switch(self) -> None:
99
        """test_enable_switch."""
100
        self.topo.enable_switch(self.dpid)
101
102
        self.topo.db.switches.find_one_and_update.assert_called()
103
        arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0]
104
        assert arg1 == {"_id": self.dpid}
105
        assert arg2["$set"]["enabled"]
106
107
    def test_disable_switch(self) -> None:
108
        """test_disable_switch."""
109
        self.topo.disable_switch(self.dpid)
110
111
        self.topo.db.switches.find_one_and_update.assert_called()
112
        arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0]
113
        assert arg1 == {"_id": self.dpid}
114
        assert not arg2["$set"]["enabled"]
115
116
    def test_deactivate_switch(self) -> None:
117
        """test_deactivate_switch."""
118
        self.topo.deactivate_switch(self.dpid)
119
120
        self.topo.db.switches.find_one_and_update.assert_called()
121
        arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0]
122
        assert arg1 == {"_id": self.dpid}
123
        assert not arg2["$set"]["active"]
124
125
    def test_add_switch_metadata(self) -> None:
126
        """test_add_switch_metadata."""
127
        metadata = {"some": "value"}
128
        self.topo.add_switch_metadata(self.dpid, metadata)
129
130
        self.topo.db.switches.find_one_and_update.assert_called()
131
        arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0]
132
        assert arg1 == {"_id": self.dpid}
133
        assert arg2["$set"]["metadata.some"] == "value"
134
135
    def test_delete_switch_metadata(self) -> None:
136
        """test_delete_switch_metadata."""
137
        key = "some"
138
        self.topo.delete_switch_metadata_key(self.dpid, key)
139
140
        self.topo.db.switches.find_one_and_update.assert_called()
141
        arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0]
142
        assert arg1 == {"_id": self.dpid}
143
        assert arg2["$unset"][f"metadata.{key}"] == ""
144
145
    def test_enable_interface(self) -> None:
146
        """test_enable_interface."""
147
        self.topo.enable_interface(self.interface_id)
148
149
        self.topo.db.switches.find_one_and_update.assert_called()
150
        arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0]
151
        assert arg1 == {"interfaces.id": self.interface_id}
152
        assert arg2["$set"]["interfaces.$.enabled"]
153
154
    def test_disable_interface(self) -> None:
155
        """test_disable_interface."""
156
        self.topo.disable_interface(self.interface_id)
157
158
        self.topo.db.switches.find_one_and_update.assert_called()
159
        arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0]
160
        assert arg1 == {"interfaces.id": self.interface_id}
161
        assert not arg2["$set"]["interfaces.$.enabled"]
162
163
    def test_activate_interface(self) -> None:
164
        """test_activate_interface."""
165
        self.topo.activate_interface(self.interface_id)
166
167
        self.topo.db.switches.find_one_and_update.assert_called()
168
        arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0]
169
        assert arg1 == {"interfaces.id": self.interface_id}
170
        assert arg2["$set"]["interfaces.$.active"]
171
172
    def test_deactivate_interface(self) -> None:
173
        """test_deactivate_interface."""
174
        self.topo.deactivate_interface(self.interface_id)
175
176
        self.topo.db.switches.find_one_and_update.assert_called()
177
        arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0]
178
        assert arg1 == {"interfaces.id": self.interface_id}
179
        assert not arg2["$set"]["interfaces.$.active"]
180
181
    def test_enable_interface_lldp(self) -> None:
182
        """test_enable_interface_lldp."""
183
        self.topo.enable_interface_lldp(self.interface_id)
184
185
        self.topo.db.switches.find_one_and_update.assert_called()
186
        arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0]
187
        assert arg1 == {"interfaces.id": self.interface_id}
188
        assert arg2["$set"]["interfaces.$.lldp"]
189
190
    def test_disable_interface_lldp(self) -> None:
191
        """test_disable_interface_lldp."""
192
        self.topo.disable_interface_lldp(self.interface_id)
193
194
        self.topo.db.switches.find_one_and_update.assert_called()
195
        arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0]
196
        assert arg1 == {"interfaces.id": self.interface_id}
197
        assert not arg2["$set"]["interfaces.$.lldp"]
198
199
    def test_add_interface_metadata(self) -> None:
200
        """test_add_interface_metadata."""
201
        metadata = {"some": "value"}
202
        self.topo.add_interface_metadata(self.interface_id, metadata)
203
204
        self.topo.db.switches.find_one_and_update.assert_called()
205
        arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0]
206
        assert arg1 == {"interfaces.id": self.interface_id}
207
        assert arg2["$set"]["interfaces.$.metadata.some"] == "value"
208
209
    def test_delete_interface_metadata_key(self) -> None:
210
        """test_delete_interface_metadata."""
211
        key = "some"
212
        self.topo.delete_interface_metadata_key(self.interface_id, key)
213
214
        self.topo.db.switches.find_one_and_update.assert_called()
215
        arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0]
216
        assert arg1 == {"interfaces.id": self.interface_id}
217
        assert arg2["$unset"][f"interfaces.$.metadata.{key}"] == ""
218
219
    def test_enable_link(self) -> None:
220
        """test_enable_link."""
221
        self.topo.enable_link(self.link_id)
222
223
        self.topo.db.links.find_one_and_update.assert_called()
224
        arg1, arg2 = self.topo.db.links.find_one_and_update.call_args[0]
225
        assert arg1 == {"_id": self.link_id}
226
        assert arg2["$set"]["enabled"]
227
228
    def test_disable_link(self) -> None:
229
        """test_disable_link."""
230
        self.topo.disable_link(self.link_id)
231
232
        self.topo.db.links.find_one_and_update.assert_called()
233
        arg1, arg2 = self.topo.db.links.find_one_and_update.call_args[0]
234
        assert arg1 == {"_id": self.link_id}
235
        assert not arg2["$set"]["enabled"]
236
237
    def test_add_link_metadata(self) -> None:
238
        """test_add_link_metadata."""
239
        key = "some_key"
240
        value = "some_value"
241
        self.topo.add_link_metadata(self.link_id, {key: value})
242
243
        self.topo.db.links.find_one_and_update.assert_called()
244
        arg1, arg2 = self.topo.db.links.find_one_and_update.call_args[0]
245
        assert arg1 == {"_id": self.link_id}
246
        assert arg2["$set"][f"metadata.{key}"] == value
247
248
    def test_delete_link_metadata_key(self) -> None:
249
        """test_delete_link_metadata_key."""
250
        key = "some_key"
251
        self.topo.delete_link_metadata_key(self.link_id, key)
252
253
        self.topo.db.links.find_one_and_update.assert_called()
254
        arg1, arg2 = self.topo.db.links.find_one_and_update.call_args[0]
255
        assert arg1 == {"_id": self.link_id}
256
        assert arg2["$unset"][f"metadata.{key}"] == ""
257
258
    def test_get_interfaces_details(self) -> None:
259
        """test_get_insterfaces_details."""
260
        interfaces_ids = ["1", "2", "3"]
261
        self.topo.get_interfaces_details(interfaces_ids)
262
        self.topo.db.interface_details.aggregate.assert_called_with(
263
            [{"$match": {"_id": {"$in": interfaces_ids}}}]
264
        )
265
266
    def test_bulk_upsert_interface_details(self) -> None:
267
        """test_bulk_upsert_interface_details."""
268
        ids_details = [
269
            ("1", {"_id": "1", "available_vlans": [1]}),
270
            ("2", {"_id": "2", "available_vlans": [2]}),
271
        ]
272
        self.topo.bulk_upsert_interface_details(ids_details)
273
274
        self.topo.db_client.start_session.assert_called()
275
        arg1 = self.topo.db.interface_details.bulk_write.call_args[0]
276
        assert len(arg1[0]) == len(ids_details)
277
        for item in arg1[0]:
278
            assert isinstance(item, UpdateOne)
279
280
    def test_upsert_switch(self) -> None:
281
        """test_upsert_switch."""
282
        switch_dict = {"enabled": True, "active": True, "_id": self.dpid}
283
        self.topo.upsert_switch(self.dpid, switch_dict)
284
285
        self.topo.db.switches.find_one_and_update.assert_called()
286
        arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0]
287
        assert arg1 == {"_id": self.dpid}
288
        for key, value in switch_dict.items():
289
            assert arg2["$set"][key] == value
290
291
    def test_upsert_link(self) -> None:
292
        """test_upsert_link."""
293
        link_dict = {
294
            "_id": self.link_id,
295
            "enabled": True,
296
            "active": True,
297
            "endpoint_a": {"id": "00:00:00:00:00:00:00:01:1"},
298
            "endpoint_b": {"id": "00:00:00:00:00:00:00:02:01"},
299
        }
300
        self.topo.upsert_link(self.link_id, link_dict)
301
        assert self.topo.db.switches.find_one_and_update.call_count == 2
302
        assert self.topo.db.links.find_one_and_update.call_count == 1
303