TestSwitch.test_update_connection()   A
last analyzed

Complexity

Conditions 1

Size

Total Lines 7
Code Lines 5

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 5
CRAP Score 1

Importance

Changes 0
Metric Value
cc 1
eloc 5
nop 1
dl 0
loc 7
ccs 5
cts 5
cp 1
crap 1
rs 10
c 0
b 0
f 0
1
"""Test kytos.core.switch module."""
2 1
import json
3 1
from datetime import datetime, timezone
4 1
from socket import error as SocketError
5 1
from unittest.mock import MagicMock, Mock, patch
6
7 1
import pytest
8
9 1
from kytos.core import Controller
10 1
from kytos.core.common import EntityStatus
11 1
from kytos.core.config import KytosConfig
12 1
from kytos.core.constants import FLOOD_TIMEOUT
13 1
from kytos.core.interface import Interface
14 1
from kytos.core.switch import Switch
15
16
17 1
def get_date():
18
    """Return date with FLOOD_TIMEOUT+1 microseconds."""
19 1
    return datetime(2000, 1, 1, 0, 0, 0, FLOOD_TIMEOUT+1)
20
21
22
# pylint: disable=protected-access, too-many-public-methods
23 1
class TestSwitch:
24
    """Switch tests."""
25
26 1
    def setup_method(self):
27
        """Instantiate a controller."""
28
29 1
        self.options = KytosConfig().options['daemon']
30 1
        self.controller = Controller(self.options)
31 1
        self.controller.log = Mock()
32 1
        self.controller._buffers = MagicMock()
33 1
        self.switch = self.create_switch()
34
35 1
    @staticmethod
36 1
    def create_switch():
37
        """Create a new switch."""
38 1
        connection = MagicMock()
39 1
        connection.address = 'addr'
40 1
        connection.port = 'port'
41 1
        connection.protocol.version = 0x04
42 1
        switch = Switch('00:00:00:00:00:00:00:01', connection)
43 1
        switch._enabled = True
44 1
        switch.update_lastseen()
45 1
        return switch
46
47 1
    def test_repr(self):
48
        """Test repr() output."""
49 1
        expected_repr = "Switch('00:00:00:00:00:00:00:01')"
50 1
        assert repr(self.switch) == expected_repr
51
52 1
    def test_id(self):
53
        """Test id property."""
54 1
        assert self.switch.id == '00:00:00:00:00:00:00:01'
55
56 1
    def test_ofp_version(self):
57
        """Test ofp_version property."""
58 1
        assert self.switch.ofp_version == '0x04'
59
60 1
    def test_ofp_version__none(self):
61
        """Test ofp_version property when connection is none."""
62 1
        self.switch.connection = None
63 1
        assert self.switch.ofp_version is None
64
65 1
    def test_update_description(self):
66
        """Test update_description method."""
67 1
        desc = MagicMock()
68 1
        desc.mfr_desc.value = 'mfr_desc'
69 1
        desc.hw_desc.value = 'hw_desc'
70 1
        desc.sw_desc.value = 'sw_desc'
71 1
        desc.serial_num.value = 'serial_num'
72 1
        desc.dp_desc.value = 'dp_desc'
73
74 1
        self.switch.update_description(desc)
75
76 1
        assert self.switch.description['manufacturer'] == 'mfr_desc'
77 1
        assert self.switch.description['hardware'] == 'hw_desc'
78 1
        assert self.switch.description['software'] == 'sw_desc'
79 1
        assert self.switch.description['serial'] == 'serial_num'
80 1
        assert self.switch.description['data_path'] == 'dp_desc'
81
82 1
    def test_disable(self):
83
        """Test disable method."""
84 1
        intf = MagicMock()
85 1
        self.switch.interfaces = {"1": intf}
86
87 1
        self.switch.disable()
88
89 1
        intf.disable.assert_called()
90 1
        assert not self.switch._enabled
91
92 1
    def test_disconnect(self):
93
        """Test disconnect method."""
94 1
        self.switch.disconnect()
95
96 1
        assert self.switch.connection is None
97
98 1
    def test_get_interface_by_port_no(self):
99
        """Test get_interface_by_port_no method."""
100 1
        interface_1 = MagicMock(port_number='1')
101 1
        interface_2 = MagicMock(port_number='2')
102 1
        self.switch.interfaces = {'1': interface_1, '2': interface_2}
103
104 1
        expected_interface_1 = self.switch.get_interface_by_port_no('1')
105 1
        expected_interface_2 = self.switch.get_interface_by_port_no('3')
106
107 1
        assert expected_interface_1 == interface_1
108 1
        assert expected_interface_2 is None
109
110 1
    def test_update_or_create_interface_case1(self):
111
        """Test update_or_create_interface method."""
112 1
        interface_1 = Interface(name='interface_2', port_number=2,
113
                                switch=self.switch)
114 1
        self.switch.interfaces = {2: interface_1}
115
116 1
        self.switch.update_or_create_interface(2, name='new_interface_2')
117 1
        assert self.switch.interfaces[2].name == 'new_interface_2'
118
119 1
    def test_update_or_create_interface_case2(self):
120
        """Test update_or_create_interface method."""
121 1
        interface_1 = Interface(name='interface_2', port_number=2,
122
                                switch=self.switch)
123 1
        self.switch.interfaces = {2: interface_1}
124
125 1
        self.switch.update_or_create_interface(3, name='new_interface_3')
126 1
        assert self.switch.interfaces[2].name == 'interface_2'
127 1
        assert self.switch.interfaces[3].name == 'new_interface_3'
128
129 1
    def test_update_or_create_interface_case3(self):
130
        """Test update_or_create_interface method."""
131 1
        interface_1 = Interface(name='interface_2', port_number=2,
132
                                switch=self.switch)
133 1
        self.switch.interfaces = {2: interface_1}
134
135 1
        self.switch.update_or_create_interface(3, name='new_interface_3')
136 1
        assert self.switch.interfaces[2].name == 'interface_2'
137 1
        assert self.switch.interfaces[3].name == 'new_interface_3'
138
139 1
    def test_get_flow_by_id(self):
140
        """Test get_flow_by_id method."""
141 1
        flow_1 = MagicMock(id='1')
142 1
        flow_2 = MagicMock(id='2')
143 1
        self.switch.flows = [flow_1, flow_2]
144
145 1
        expected_flow_1 = self.switch.get_flow_by_id('1')
146 1
        expected_flow_2 = self.switch.get_flow_by_id('3')
147
148 1
        assert expected_flow_1 == flow_1
149 1
        assert expected_flow_2 is None
150
151 1
    def test_is_connected__true(self):
152
        """Test is_connected method."""
153 1
        connection = MagicMock()
154 1
        connection.is_alive.return_value = True
155 1
        connection.is_established.return_value = True
156 1
        self.switch.connection = connection
157
158 1
        assert self.switch.is_connected()
159
160 1
    def test_is_connected__not_connection(self):
161
        """Test is_connected method when connection does not exist."""
162 1
        self.switch.connection = None
163
164 1
        assert not self.switch.is_connected()
165
166 1
    def test_is_connected__not_alive(self):
167
        """Test is_connected method when switch has connection timeout."""
168 1
        connection = MagicMock()
169 1
        connection.is_alive.return_value = True
170 1
        connection.is_established.return_value = True
171 1
        self.switch.connection = connection
172 1
        self.switch.lastseen = datetime(1, 1, 1, 0, 0, 0, 0, timezone.utc)
173
174 1
        assert not self.switch.is_connected()
175
176 1
    def test_is_active(self):
177
        """Test is_active method."""
178 1
        self.switch.is_connected = MagicMock()
179 1
        self.switch.is_connected.return_value = True
180 1
        assert self.switch.is_active()
181 1
        self.switch.is_connected.return_value = False
182 1
        assert not self.switch.is_active()
183
184 1
    def test_update_connection(self):
185
        """Test update_connection method."""
186 1
        connection = MagicMock()
187 1
        self.switch.update_connection(connection)
188
189 1
        assert self.switch.connection == connection
190 1
        assert self.switch.connection.switch == self.switch
191
192 1
    def test_update_features(self):
193
        """Test update_features method."""
194 1
        self.switch.update_features('features')
195
196 1
        assert self.switch.features == 'features'
197
198 1
    def test_send(self):
199
        """Test send method."""
200 1
        self.switch.send('buffer')
201
202 1
        self.switch.connection.send.assert_called_with('buffer')
203
204 1
    def test_send_error(self):
205
        """Test send method to error case."""
206 1
        self.switch.connection.send.side_effect = SocketError
207
208 1
        with pytest.raises(SocketError):
209 1
            self.switch.send(b'data')
210
211 1
    @patch('kytos.core.switch.now', return_value=get_date())
212 1
    def test_update_lastseen(self, mock_now):
213
        """Test update_lastseen method."""
214 1
        self.switch.update_lastseen()
215
216 1
        assert self.switch.lastseen == mock_now.return_value
217
218 1
    def test_update_interface(self):
219
        """Test update_interface method."""
220 1
        interface = MagicMock(port_number=1)
221 1
        self.switch.update_interface(interface)
222
223 1
        assert self.switch.interfaces[1] == interface
224
225 1
    def test_remove_interface(self):
226
        """Test remove_interface method."""
227 1
        intf = MagicMock(port_number=1)
228 1
        self.switch.interfaces[1] = intf
229
230 1
        self.switch.remove_interface(intf)
231
232 1
        assert not self.switch.interfaces
233
234 1
    def test_update_mac_table(self):
235
        """Test update_mac_table method."""
236 1
        mac = MagicMock(value='00:00:00:00:00:00')
237 1
        self.switch.update_mac_table(mac, 1)
238 1
        self.switch.update_mac_table(mac, 2)
239
240 1
        assert self.switch.mac2port[mac.value] == {1, 2}
241
242 1
    def test_last_flood(self):
243
        """Test last_flood method."""
244 1
        self.switch.flood_table['hash'] = 'timestamp'
245 1
        ethernet_frame = MagicMock()
246 1
        ethernet_frame.get_hash.return_value = 'hash'
247
248 1
        last_flood = self.switch.last_flood(ethernet_frame)
249
250 1
        assert last_flood == 'timestamp'
251
252 1
    def test_last_flood__error(self):
253
        """Test last_flood method to error case."""
254 1
        ethernet_frame = MagicMock()
255 1
        ethernet_frame.get_hash.return_value = 'hash'
256
257 1
        last_flood = self.switch.last_flood(ethernet_frame)
258
259 1
        assert last_flood is None
260
261 1
    @patch('kytos.core.switch.now', return_value=get_date())
262 1
    def test_should_flood(self, _):
263
        """Test should_flood method."""
264 1
        self.switch.flood_table['hash1'] = datetime(2000, 1, 1, 0, 0, 0, 0)
265 1
        self.switch.flood_table['hash2'] = datetime(2000, 1, 1, 0, 0, 0,
266
                                                    FLOOD_TIMEOUT)
267
268 1
        ethernet_frame = MagicMock()
269 1
        ethernet_frame.get_hash.side_effect = ['hash1', 'hash2']
270
271 1
        should_flood_1 = self.switch.should_flood(ethernet_frame)
272 1
        should_flood_2 = self.switch.should_flood(ethernet_frame)
273
274 1
        assert should_flood_1
275 1
        assert not should_flood_2
276
277 1
    @patch('kytos.core.switch.now', return_value=get_date())
278 1
    def test_update_flood_table(self, mock_now):
279
        """Test update_flood_table method."""
280 1
        ethernet_frame = MagicMock()
281 1
        ethernet_frame.get_hash.return_value = 'hash'
282
283 1
        self.switch.update_flood_table(ethernet_frame)
284
285 1
        assert self.switch.flood_table['hash'] == mock_now.return_value
286
287 1
    def test_where_is_mac(self):
288
        """Test where_is_mac method."""
289 1
        mac = MagicMock(value='00:00:00:00:00:00')
290
291 1
        expected_ports_1 = self.switch.where_is_mac(mac)
292
293 1
        self.switch.mac2port['00:00:00:00:00:00'] = set([1, 2, 3])
294 1
        expected_ports_2 = self.switch.where_is_mac(mac)
295
296 1
        assert expected_ports_1 is None
297 1
        assert expected_ports_2 == [1, 2, 3]
298
299 1
    def test_as_dict_as_json(self):
300
        """Test as_dict and as_json method."""
301 1
        expected_dict = {
302
            'id': '00:00:00:00:00:00:00:01',
303
            'name': '00:00:00:00:00:00:00:01',
304
            'dpid': '00:00:00:00:00:00:00:01',
305
            'connection': 'addr:port',
306
            'ofp_version': '0x04',
307
            'type': 'switch',
308
            'manufacturer': '',
309
            'serial': '',
310
            'hardware': '',
311
            'software': None,
312
            'data_path': '',
313
            'interfaces': {},
314
            'metadata': {},
315
            'active': True,
316
            'enabled': True,
317
            'status': 'UP',
318
            'status_reason': []
319
        }
320 1
        assert self.switch.as_dict() == expected_dict
321
322 1
        expected_json = json.dumps(expected_dict)
323
324 1
        assert self.switch.as_json() == expected_json
325
326 1
    def test_switch_initial_lastseen(self):
327
        """Test lastseen attribute initialization."""
328 1
        connection = MagicMock()
329 1
        connection.protocol.version = 0x04
330 1
        switch = Switch('00:00:00:00:00:00:00:01', connection)
331 1
        assert switch.is_active() is False
332 1
        assert switch.lastseen == datetime(1, 1, 1, 0, 0, 0, 0, timezone.utc)
333
334 1
    def test_status_funcs(self) -> None:
335
        """Test status_funcs."""
336 1
        self.switch.enable()
337 1
        self.switch.activate()
338 1
        assert self.switch.is_active()
339 1
        Switch.register_status_func(
340
            "some_napp_some_func",
341
            lambda switch: EntityStatus.DOWN
342
        )
343 1
        Switch.register_status_reason_func(
344
            "some_napp_some_func",
345
            lambda iface: {'test_status'}
346
        )
347 1
        assert self.switch.status == EntityStatus.DOWN
348 1
        assert self.switch.status_reason == {'test_status'}
349 1
        Switch.register_status_func(
350
            "some_napp_some_func",
351
            lambda iface: None
352
        )
353 1
        Switch.register_status_reason_func(
354
            "some_napp_some_func",
355
            lambda iface: set()
356
        )
357 1
        assert self.switch.status == EntityStatus.UP
358
        assert self.switch.status_reason == set()
359