Completed
Push — master ( 76cc93...764801 )
by Beraldo
14s queued 11s
created

tests.test_core.test_switch   A

Complexity

Total Complexity 4

Size/Duplication

Total Lines 65
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
eloc 43
dl 0
loc 65
rs 10
c 0
b 0
f 0
wmc 4

4 Methods

Rating   Name   Duplication   Size   Complexity  
A TestSwitch.test_switch_set_options_vlan_pool() 0 32 1
A TestSwitch.test_switch_vlan_pool_default() 0 3 1
A TestSwitch.setUp() 0 9 1
A TestSwitch.tearDown() 0 3 1
1
"""Test kytos.core.switch module."""
2
import asyncio
3
from unittest import TestCase
4
from unittest.mock import Mock
5
6
from kytos.core import Controller
7
from kytos.core.config import KytosConfig
8
from kytos.core.interface import Interface
9
from kytos.core.switch import Switch
10
11
12
class TestSwitch(TestCase):
13
    """Switch tests."""
14
15
    def setUp(self):
16
        """Instantiate a controller."""
17
18
        self.loop = asyncio.new_event_loop()
19
        asyncio.set_event_loop(None)
20
21
        self.options = KytosConfig().options['daemon']
22
        self.controller = Controller(self.options, loop=self.loop)
23
        self.controller.log = Mock()
24
25
    def tearDown(self):
26
        """TearDown."""
27
        self.loop.close()
28
29
    def test_switch_vlan_pool_default(self):
30
        """Test default vlan_pool value."""
31
        self.assertEqual(self.options.vlan_pool, {})
32
33
    def test_switch_set_options_vlan_pool(self):
34
        """Test switch with the example from kytos.conf."""
35
        dpid = "00:00:00:00:00:00:00:01"
36
        vlan_pool_json = '{"00:00:00:00:00:00:00:01": {"1": [[1, 2], [5, 10]], \
37
            "4": [[3, 4]]}}'
38
        switch = Switch(dpid)
39
        self.controller.switches[dpid] = switch
40
        self.options.vlan_pool = vlan_pool_json
41
        switch.connection = Mock()
42
        switch.connection.protocol.version = 0x04
43
        self.controller.get_switch_or_create(dpid, switch.connection)
44
45
        port_id = 1
46
        intf = self.controller.switches[dpid].interfaces[port_id]
47
        tag_values = [tag.value for tag in intf.available_tags]
48
        self.assertEqual(tag_values, [1, 5, 6, 7, 8, 9])
49
50
        port_id = 4
51
        intf = self.controller.switches[dpid].interfaces[port_id]
52
        tag_values = [tag.value for tag in intf.available_tags]
53
        self.assertEqual(tag_values, [3])
54
55
        # this port number doesn't exist yet.
56
        port_7 = 7
57
        intf = Interface("test", port_7, switch)
58
        # no attr filters, so should associate as it is
59
        self.controller.switches[dpid].update_interface(intf)
60
        intf_obj = self.controller.switches[dpid].interfaces[port_7]
61
        self.assertEqual(intf_obj, intf)
62
        # assert default vlan_pool range (1, 4096)
63
        tag_values = [tag.value for tag in intf_obj.available_tags]
64
        self.assertEqual(tag_values, list(range(1, 4096)))
65