1
|
|
|
"""Module to test the Path class.""" |
2
|
|
|
import sys |
3
|
|
|
from unittest import TestCase |
4
|
|
|
from unittest.mock import patch, Mock |
5
|
|
|
|
6
|
|
|
from kytos.core.common import EntityStatus |
7
|
|
|
|
8
|
|
|
# pylint: disable=wrong-import-position |
9
|
|
|
|
10
|
|
|
sys.path.insert(0, "/var/lib/kytos/napps/..") |
11
|
|
|
# pylint: enable=wrong-import-position |
12
|
|
|
from napps.kytos.mef_eline.exceptions import InvalidPath # NOQA pycodestyle |
13
|
|
|
from napps.kytos.mef_eline.models import Path # NOQA pycodestyle |
14
|
|
|
from napps.kytos.mef_eline.tests.helpers import ( |
15
|
|
|
MockResponse, |
16
|
|
|
get_link_mocked, |
17
|
|
|
get_mocked_requests, |
18
|
|
|
) # NOQA pycodestyle |
19
|
|
|
|
20
|
|
|
|
21
|
|
|
class TestPath(TestCase): |
22
|
|
|
"""Class to test path methods.""" |
23
|
|
|
|
24
|
|
|
def test_is_affected_by_link_1(self): |
25
|
|
|
"""Test method is affected by link.""" |
26
|
|
|
path = Path() |
27
|
|
|
self.assertIs(path.is_affected_by_link(), False) |
28
|
|
|
|
29
|
|
|
def test_link_affected_by_interface_1(self): |
30
|
|
|
"""Test method to get the link using an interface.""" |
31
|
|
|
link1 = Mock() |
32
|
|
|
link1.endpoint_a = "a" |
33
|
|
|
link1.endpoint_b = "b" |
34
|
|
|
link2 = Mock() |
35
|
|
|
link2.endpoint_a = "c" |
36
|
|
|
link2.endpoint_b = "d" |
37
|
|
|
path = Path([link1, link2]) |
38
|
|
|
self.assertIsNone(path.link_affected_by_interface()) |
39
|
|
|
|
40
|
|
|
def test_link_affected_by_interface_2(self): |
41
|
|
|
"""Test method to get the link using an interface.""" |
42
|
|
|
link1 = Mock() |
43
|
|
|
link1.endpoint_a = "a" |
44
|
|
|
link1.endpoint_b = "b" |
45
|
|
|
link2 = Mock() |
46
|
|
|
link2.endpoint_a = "c" |
47
|
|
|
link2.endpoint_b = "d" |
48
|
|
|
path = Path([link1, link2]) |
49
|
|
|
self.assertEqual(path.link_affected_by_interface("a"), link1) |
50
|
|
|
|
51
|
|
|
def test_status_case_1(self): |
52
|
|
|
"""Test if empty link is DISABLED.""" |
53
|
|
|
current_path = Path() |
54
|
|
|
self.assertEqual(current_path.status, EntityStatus.DISABLED) |
55
|
|
|
|
56
|
|
|
@patch("requests.get", side_effect=get_mocked_requests) |
57
|
|
|
def test_status_case_2(self, requests_mocked): |
58
|
|
|
# pylint: disable=unused-argument |
59
|
|
|
"""Test if link status is DOWN.""" |
60
|
|
|
link1 = get_link_mocked() |
61
|
|
|
link2 = get_link_mocked() |
62
|
|
|
link1.id = "def" |
63
|
|
|
link2.id = "abc" |
64
|
|
|
links = [link1, link2] |
65
|
|
|
current_path = Path(links) |
66
|
|
|
self.assertEqual(current_path.status, EntityStatus.DOWN) |
67
|
|
|
|
68
|
|
|
def test_status_case_3(self): |
69
|
|
|
"""Test if link status is DISABLED.""" |
70
|
|
|
links = [] |
71
|
|
|
current_path = Path(links) |
72
|
|
|
self.assertEqual(current_path.status, EntityStatus.DISABLED) |
73
|
|
|
|
74
|
|
|
# This method will be used by the mock to replace requests.get |
75
|
|
|
def _mocked_requests_get_status_case_4(self): |
76
|
|
|
# pylint: disable=no-self-use |
77
|
|
|
return MockResponse( |
78
|
|
|
{ |
79
|
|
|
"links": { |
80
|
|
|
"abc": {"active": True, "enabled": True}, |
81
|
|
|
"def": {"active": True, "enabled": True}, |
82
|
|
|
} |
83
|
|
|
}, |
84
|
|
|
200, |
85
|
|
|
) |
86
|
|
|
|
87
|
|
|
@patch("requests.get", side_effect=_mocked_requests_get_status_case_4) |
88
|
|
|
def test_status_case_4(self, requests_mocked): |
89
|
|
|
# pylint: disable=unused-argument |
90
|
|
|
"""Test if link status is UP.""" |
91
|
|
|
link1 = get_link_mocked() |
92
|
|
|
link2 = get_link_mocked() |
93
|
|
|
link1.id = "def" |
94
|
|
|
link2.id = "abc" |
95
|
|
|
links = [link1, link2] |
96
|
|
|
current_path = Path(links) |
97
|
|
|
self.assertEqual(current_path.status, EntityStatus.UP) |
98
|
|
|
|
99
|
|
|
# This method will be used by the mock to replace requests.get |
100
|
|
|
def _mocked_requests_get_status_case_5(self): |
101
|
|
|
# pylint: disable=no-self-use |
102
|
|
|
return MockResponse( |
103
|
|
|
{ |
104
|
|
|
"links": { |
105
|
|
|
"abc": {"active": True, "enabled": True}, |
106
|
|
|
"def": {"active": False, "enabled": False}, |
107
|
|
|
} |
108
|
|
|
}, |
109
|
|
|
200, |
110
|
|
|
) |
111
|
|
|
|
112
|
|
|
@patch("requests.get", side_effect=_mocked_requests_get_status_case_5) |
113
|
|
|
def test_status_case_5(self, requests_mocked): |
114
|
|
|
# pylint: disable=unused-argument |
115
|
|
|
"""Test if link status is UP.""" |
116
|
|
|
link1 = get_link_mocked() |
117
|
|
|
link2 = get_link_mocked() |
118
|
|
|
link1.id = "def" |
119
|
|
|
link2.id = "abc" |
120
|
|
|
links = [link1, link2] |
121
|
|
|
current_path = Path(links) |
122
|
|
|
self.assertEqual(current_path.status, EntityStatus.DISABLED) |
123
|
|
|
|
124
|
|
|
# This method will be used by the mock to replace requests.get |
125
|
|
|
def _mocked_requests_get_status_case_6(self): |
126
|
|
|
# pylint: disable=no-self-use |
127
|
|
|
return MockResponse( |
128
|
|
|
{ |
129
|
|
|
"links": { |
130
|
|
|
"abc": {"active": False, "enabled": False}, |
131
|
|
|
"def": {"active": False, "enabled": True}, |
132
|
|
|
} |
133
|
|
|
}, |
134
|
|
|
200, |
135
|
|
|
) |
136
|
|
|
|
137
|
|
|
@patch("requests.get", side_effect=_mocked_requests_get_status_case_6) |
138
|
|
|
def test_status_case_6(self, requests_mocked): |
139
|
|
|
# pylint: disable=unused-argument |
140
|
|
|
"""Test if link status is UP.""" |
141
|
|
|
link1 = get_link_mocked() |
142
|
|
|
link2 = get_link_mocked() |
143
|
|
|
link1.id = "def" |
144
|
|
|
link2.id = "abc" |
145
|
|
|
links = [link1, link2] |
146
|
|
|
current_path = Path(links) |
147
|
|
|
self.assertEqual(current_path.status, EntityStatus.DISABLED) |
148
|
|
|
|
149
|
|
|
def test_compare_same_paths(self): |
150
|
|
|
"""Test compare paths with same links.""" |
151
|
|
|
links = [ |
152
|
|
|
get_link_mocked( |
153
|
|
|
endpoint_a_port=9, endpoint_b_port=10, metadata={"s_vlan": 5} |
154
|
|
|
), |
155
|
|
|
get_link_mocked( |
156
|
|
|
endpoint_a_port=11, endpoint_b_port=12, metadata={"s_vlan": 6} |
157
|
|
|
), |
158
|
|
|
] |
159
|
|
|
|
160
|
|
|
path_1 = Path(links) |
161
|
|
|
path_2 = Path(links) |
162
|
|
|
self.assertEqual(path_1, path_2) |
163
|
|
|
|
164
|
|
|
def test_compare_different_paths(self): |
165
|
|
|
"""Test compare paths with different links.""" |
166
|
|
|
links_1 = [ |
167
|
|
|
get_link_mocked( |
168
|
|
|
endpoint_a_port=9, endpoint_b_port=10, metadata={"s_vlan": 5} |
169
|
|
|
), |
170
|
|
|
get_link_mocked( |
171
|
|
|
endpoint_a_port=11, endpoint_b_port=12, metadata={"s_vlan": 6} |
172
|
|
|
), |
173
|
|
|
] |
174
|
|
|
links_2 = [ |
175
|
|
|
get_link_mocked( |
176
|
|
|
endpoint_a_port=12, endpoint_b_port=11, metadata={"s_vlan": 5} |
177
|
|
|
), |
178
|
|
|
get_link_mocked( |
179
|
|
|
endpoint_a_port=14, endpoint_b_port=16, metadata={"s_vlan": 11} |
180
|
|
|
), |
181
|
|
|
] |
182
|
|
|
|
183
|
|
|
path_1 = Path(links_1) |
184
|
|
|
path_2 = Path(links_2) |
185
|
|
|
self.assertNotEqual(path_1, path_2) |
186
|
|
|
|
187
|
|
|
def test_as_dict(self): |
188
|
|
|
"""Test path as dict.""" |
189
|
|
|
links = [ |
190
|
|
|
get_link_mocked(link_dict={"id": 3}), |
191
|
|
|
get_link_mocked(link_dict={"id": 2}), |
192
|
|
|
] |
193
|
|
|
|
194
|
|
|
current_path = Path(links) |
195
|
|
|
expected_dict = [{"id": 3}, {"id": 2}] |
196
|
|
|
self.assertEqual(expected_dict, current_path.as_dict()) |
197
|
|
|
|
198
|
|
|
def test_is_valid(self): |
199
|
|
|
"""Test is_valid method.""" |
200
|
|
|
switch1 = "00:00:00:00:00:00:00:01" |
201
|
|
|
switch2 = "00:00:00:00:00:00:00:02" |
202
|
|
|
switch3 = "00:00:00:00:00:00:00:03" |
203
|
|
|
switch4 = "00:00:00:00:00:00:00:04" |
204
|
|
|
switch5 = "00:00:00:00:00:00:00:05" |
205
|
|
|
switch6 = "00:00:00:00:00:00:00:06" |
206
|
|
|
|
207
|
|
|
links1 = [ |
208
|
|
|
get_link_mocked(switch_a=switch1, switch_b=switch2), |
209
|
|
|
get_link_mocked(switch_a=switch2, switch_b=switch3), |
210
|
|
|
get_link_mocked(switch_a=switch3, switch_b=switch4), |
211
|
|
|
get_link_mocked(switch_a=switch4, switch_b=switch5), |
212
|
|
|
get_link_mocked(switch_a=switch5, switch_b=switch6), |
213
|
|
|
] |
214
|
|
|
|
215
|
|
|
links2 = [ |
216
|
|
|
get_link_mocked(switch_a=switch1, switch_b=switch2), |
217
|
|
|
get_link_mocked(switch_a=switch3, switch_b=switch2), |
218
|
|
|
get_link_mocked(switch_a=switch3, switch_b=switch4), |
219
|
|
|
] |
220
|
|
|
|
221
|
|
|
for links, switch_a, switch_z, expected in ( |
222
|
|
|
(links1, switch1, switch6, True), |
223
|
|
|
(links2, switch1, switch4, False), |
224
|
|
|
(links1, switch2, switch6, False), |
225
|
|
|
): |
226
|
|
|
with self.subTest( |
227
|
|
|
links=links, |
228
|
|
|
switch_a=switch_a, |
229
|
|
|
switch_z=switch_z, |
230
|
|
|
expected=expected, |
231
|
|
|
): |
232
|
|
|
path = Path(links) |
233
|
|
|
if expected: |
234
|
|
|
self.assertEqual( |
235
|
|
|
path.is_valid(switch_a, switch_z), expected |
236
|
|
|
) |
237
|
|
|
else: |
238
|
|
|
with self.assertRaises(InvalidPath): |
239
|
|
|
path.is_valid(switch_a, switch_z) |
240
|
|
|
|