tests.unit.test_core.test_switch.TestSwitch.create_switch() - Code Metrics - Inspection of "Merge pull request #284 from kytos-ng/feature/upgr..." - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 15e965...99a8a8 )

by Vinicius

created 2022-09-20 13:53 UTC

TestSwitch.create_switch() A

↳ Parent: tests.unit.test_core.test_switch

Complexity

Conditions

Size

Total Lines	11
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	10
nop	0
dl	0
loc	11
rs	9.9
c	0
b	0
f	0

"""Test kytos.core.switch module."""
import json
from datetime import datetime, timezone
from socket import error as SocketError
from unittest import TestCase
from unittest.mock import MagicMock, Mock, patch

from kytos.core import Controller
from kytos.core.buffers import KytosBuffers
from kytos.core.config import KytosConfig
from kytos.core.constants import FLOOD_TIMEOUT
from kytos.core.interface import Interface
from kytos.core.switch import Switch


def get_date():
    """Return date with FLOOD_TIMEOUT+1 microseconds."""
    return datetime(2000, 1, 1, 0, 0, 0, FLOOD_TIMEOUT+1)


# pylint: disable=protected-access, too-many-public-methods
class TestSwitch(TestCase):
    """Switch tests."""

    def setUp(self):
        """Instantiate a controller."""

        self.options = KytosConfig().options['daemon']
        self.controller = Controller(self.options)
        self.controller.buffers = KytosBuffers()
        self.controller.log = Mock()

        self.switch = self.create_switch()

    @staticmethod
    def create_switch():
        """Create a new switch."""
        connection = MagicMock()
        connection.address = 'addr'
        connection.port = 'port'
        connection.protocol.version = 0x04
        switch = Switch('00:00:00:00:00:00:00:01', connection)
        switch._enabled = True
        switch.update_lastseen()
        return switch

    def test_repr(self):
        """Test repr() output."""
        expected_repr = "Switch('00:00:00:00:00:00:00:01')"
        self.assertEqual(repr(self.switch), expected_repr)

    def test_id(self):
        """Test id property."""
        self.assertEqual(self.switch.id, '00:00:00:00:00:00:00:01')

    def test_ofp_version(self):
        """Test ofp_version property."""
        self.assertEqual(self.switch.ofp_version, '0x04')

    def test_ofp_version__none(self):
        """Test ofp_version property when connection is none."""
        self.switch.connection = None
        self.assertIsNone(self.switch.ofp_version)

    def test_switch_vlan_pool_default(self):
        """Test default vlan_pool value."""
        self.assertEqual(self.options.vlan_pool, {})

    def test_switch_vlan_pool_options(self):
        """Test switch with the example from kytos.conf."""
        dpid = "00:00:00:00:00:00:00:01"
        vlan_pool = {"00:00:00:00:00:00:00:01":
                     {"1": [[1, 2], [5, 10]], "4": [[3, 4]]}}
        self.controller.switches[dpid] = self.switch
        self.options.vlan_pool = vlan_pool
        self.controller.get_switch_or_create(dpid, self.switch.connection)

        port_id = 1
        intf = self.controller.switches[dpid].interfaces[port_id]
        tag_values = [tag.value for tag in intf.available_tags]
        self.assertEqual(tag_values, [1, 5, 6, 7, 8, 9])

        port_id = 4
        intf = self.controller.switches[dpid].interfaces[port_id]
        tag_values = [tag.value for tag in intf.available_tags]
        self.assertEqual(tag_values, [3])

        # this port number doesn't exist yet.
        port_7 = 7
        intf = Interface("test", port_7, self.switch)
        # no attr filters, so should associate as it is
        self.controller.switches[dpid].update_interface(intf)
        intf_obj = self.controller.switches[dpid].interfaces[port_7]
        self.assertEqual(intf_obj, intf)
        # assert default vlan_pool range (1, 4096)
        tag_values = [tag.value for tag in intf_obj.available_tags]
        self.assertEqual(tag_values, list(range(1, 4096)))

    def test_update_description(self):
        """Test update_description method."""
        desc = MagicMock()
        desc.mfr_desc.value = 'mfr_desc'
        desc.hw_desc.value = 'hw_desc'
        desc.sw_desc.value = 'sw_desc'
        desc.serial_num.value = 'serial_num'
        desc.dp_desc.value = 'dp_desc'

        self.switch.update_description(desc)

        self.assertEqual(self.switch.description['manufacturer'], 'mfr_desc')
        self.assertEqual(self.switch.description['hardware'], 'hw_desc')
        self.assertEqual(self.switch.description['software'], 'sw_desc')
        self.assertEqual(self.switch.description['serial'], 'serial_num')
        self.assertEqual(self.switch.description['data_path'], 'dp_desc')

    def test_disable(self):
        """Test disable method."""
        intf = MagicMock()
        self.switch.interfaces = {"1": intf}

        self.switch.disable()

        intf.disable.assert_called()
        self.assertFalse(self.switch._enabled)

    def test_disconnect(self):
        """Test disconnect method."""
        self.switch.disconnect()

        self.assertIsNone(self.switch.connection)

    def test_get_interface_by_port_no(self):
        """Test get_interface_by_port_no method."""
        interface_1 = MagicMock(port_number='1')
        interface_2 = MagicMock(port_number='2')
        self.switch.interfaces = {'1': interface_1, '2': interface_2}

        expected_interface_1 = self.switch.get_interface_by_port_no('1')
        expected_interface_2 = self.switch.get_interface_by_port_no('3')

        self.assertEqual(expected_interface_1, interface_1)
        self.assertIsNone(expected_interface_2)

    def test_update_or_create_interface_case1(self):
        """Test update_or_create_interface method."""
        interface_1 = Interface(name='interface_2', port_number=2,
                                switch=self.switch)
        self.switch.interfaces = {2: interface_1}

        self.switch.update_or_create_interface(2, name='new_interface_2')
        self.assertEqual(self.switch.interfaces[2].name, 'new_interface_2')

    def test_update_or_create_interface_case2(self):
        """Test update_or_create_interface method."""
        interface_1 = Interface(name='interface_2', port_number=2,
                                switch=self.switch)
        self.switch.interfaces = {2: interface_1}

        self.switch.update_or_create_interface(3, name='new_interface_3')
        self.assertEqual(self.switch.interfaces[2].name, 'interface_2')
        self.assertEqual(self.switch.interfaces[3].name, 'new_interface_3')

    def test_update_or_create_interface_case3(self):
        """Test update_or_create_interface method."""
        interface_1 = Interface(name='interface_2', port_number=2,
                                switch=self.switch)
        self.switch.interfaces = {2: interface_1}

        self.switch.update_or_create_interface(3, name='new_interface_3')
        self.assertEqual(self.switch.interfaces[2].name, 'interface_2')
        self.assertEqual(self.switch.interfaces[3].name, 'new_interface_3')

    def test_get_flow_by_id(self):
        """Test get_flow_by_id method."""
        flow_1 = MagicMock(id='1')
        flow_2 = MagicMock(id='2')
        self.switch.flows = [flow_1, flow_2]

        expected_flow_1 = self.switch.get_flow_by_id('1')
        expected_flow_2 = self.switch.get_flow_by_id('3')

        self.assertEqual(expected_flow_1, flow_1)
        self.assertIsNone(expected_flow_2)

    def test_is_connected__true(self):
        """Test is_connected method."""
        connection = MagicMock()
        connection.is_alive.return_value = True
        connection.is_established.return_value = True
        self.switch.connection = connection

        self.assertTrue(self.switch.is_connected())

    def test_is_connected__not_connection(self):
        """Test is_connected method when connection does not exist."""
        self.switch.connection = None

        self.assertFalse(self.switch.is_connected())

    def test_is_connected__not_alive(self):
        """Test is_connected method when switch has connection timeout."""
        connection = MagicMock()
        connection.is_alive.return_value = True
        connection.is_established.return_value = True
        self.switch.connection = connection
        self.switch.lastseen = datetime(1, 1, 1, 0, 0, 0, 0, timezone.utc)

        self.assertFalse(self.switch.is_connected())

    def test_is_active(self):
        """Test is_active method."""
        self.switch.is_connected = MagicMock()
        self.switch.is_connected.return_value = True
        self.assertTrue(self.switch.is_active())
        self.switch.is_connected.return_value = False
        self.assertFalse(self.switch.is_active())

    def test_update_connection(self):
        """Test update_connection method."""
        connection = MagicMock()
        self.switch.update_connection(connection)

        self.assertEqual(self.switch.connection, connection)
        self.assertEqual(self.switch.connection.switch, self.switch)

    def test_update_features(self):
        """Test update_features method."""
        self.switch.update_features('features')

        self.assertEqual(self.switch.features, 'features')

    def test_send(self):
        """Test send method."""
        self.switch.send('buffer')

        self.switch.connection.send.assert_called_with('buffer')

    def test_send_error(self):
        """Test send method to error case."""
        self.switch.connection.send.side_effect = SocketError

        with self.assertRaises(SocketError):
            self.switch.send(b'data')

    @patch('kytos.core.switch.now', return_value=get_date())
    def test_update_lastseen(self, mock_now):
        """Test update_lastseen method."""
        self.switch.update_lastseen()

        self.assertEqual(self.switch.lastseen, mock_now.return_value)

    def test_update_interface(self):
        """Test update_interface method."""
        interface = MagicMock(port_number=1)
        self.switch.update_interface(interface)

        self.assertEqual(self.switch.interfaces[1], interface)

    def test_remove_interface(self):
        """Test remove_interface method."""
        intf = MagicMock(port_number=1)
        self.switch.interfaces[1] = intf

        self.switch.remove_interface(intf)

        self.assertEqual(self.switch.interfaces, {})

    def test_update_mac_table(self):
        """Test update_mac_table method."""
        mac = MagicMock(value='00:00:00:00:00:00')
        self.switch.update_mac_table(mac, 1)
        self.switch.update_mac_table(mac, 2)

        self.assertEqual(self.switch.mac2port[mac.value], {1, 2})

    def test_last_flood(self):
        """Test last_flood method."""
        self.switch.flood_table['hash'] = 'timestamp'
        ethernet_frame = MagicMock()
        ethernet_frame.get_hash.return_value = 'hash'

        last_flood = self.switch.last_flood(ethernet_frame)

        self.assertEqual(last_flood, 'timestamp')

    def test_last_flood__error(self):
        """Test last_flood method to error case."""
        ethernet_frame = MagicMock()
        ethernet_frame.get_hash.return_value = 'hash'

        last_flood = self.switch.last_flood(ethernet_frame)

        self.assertIsNone(last_flood)

    @patch('kytos.core.switch.now', return_value=get_date())
    def test_should_flood(self, _):
        """Test should_flood method."""
        self.switch.flood_table['hash1'] = datetime(2000, 1, 1, 0, 0, 0, 0)
        self.switch.flood_table['hash2'] = datetime(2000, 1, 1, 0, 0, 0,
                                                    FLOOD_TIMEOUT)

        ethernet_frame = MagicMock()
        ethernet_frame.get_hash.side_effect = ['hash1', 'hash2']

        should_flood_1 = self.switch.should_flood(ethernet_frame)
        should_flood_2 = self.switch.should_flood(ethernet_frame)

        self.assertTrue(should_flood_1)
        self.assertFalse(should_flood_2)

    @patch('kytos.core.switch.now', return_value=get_date())
    def test_update_flood_table(self, mock_now):
        """Test update_flood_table method."""
        ethernet_frame = MagicMock()
        ethernet_frame.get_hash.return_value = 'hash'

        self.switch.update_flood_table(ethernet_frame)

        self.assertEqual(self.switch.flood_table['hash'],
                         mock_now.return_value)

    def test_where_is_mac(self):
        """Test where_is_mac method."""
        mac = MagicMock(value='00:00:00:00:00:00')

        expected_ports_1 = self.switch.where_is_mac(mac)

        self.switch.mac2port['00:00:00:00:00:00'] = set([1, 2, 3])
        expected_ports_2 = self.switch.where_is_mac(mac)

        self.assertIsNone(expected_ports_1)
        self.assertEqual(expected_ports_2, [1, 2, 3])

    def test_as_dict(self):
        """Test as_dict method."""
        expected_dict = {'id': '00:00:00:00:00:00:00:01',
                         'name': '00:00:00:00:00:00:00:01',
                         'dpid': '00:00:00:00:00:00:00:01',
                         'connection': 'addr:port',
                         'ofp_version': '0x04',
                         'type': 'switch',
                         'manufacturer': '',
                         'serial': '',
                         'hardware': '',
                         'software': None,
                         'data_path': '',
                         'interfaces': {},
                         'metadata': {},
                         'active': True,
                         'enabled': True}
        self.assertEqual(self.switch.as_dict(), expected_dict)

    def test_as_json(self):
        """Test as_json method."""
        expected_json = json.dumps({'id': '00:00:00:00:00:00:00:01',
                                    'name': '00:00:00:00:00:00:00:01',
                                    'dpid': '00:00:00:00:00:00:00:01',
                                    'connection': 'addr:port',
                                    'ofp_version': '0x04',
                                    'type': 'switch',
                                    'manufacturer': '',
                                    'serial': '',
                                    'hardware': '',
                                    'software': None,
                                    'data_path': '',
                                    'interfaces': {},
                                    'metadata': {},
                                    'active': True,
                                    'enabled': True})

        self.assertEqual(self.switch.as_json(), expected_json)

    def test_switch_initial_lastseen(self):
        """Test lastseen attribute initialization."""
        connection = MagicMock()
        connection.protocol.version = 0x04
        switch = Switch('00:00:00:00:00:00:00:01', connection)
        self.assertEqual(switch.is_active(), False)
        self.assertEqual(switch.lastseen,
                         datetime(1, 1, 1, 0, 0, 0, 0, timezone.utc))


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

kytos-ng / kytos

Push — master ( 15e965...99a8a8 )

TestSwitch.create_switch() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like