Completed
Push — master ( ae1c36...646d2b )
by Humberto
25s queued 14s
created

tests.test_core.test_link   A

Complexity

Total Complexity 12

Size/Duplication

Total Lines 121
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
eloc 80
dl 0
loc 121
rs 10
c 0
b 0
f 0
wmc 12

10 Methods

Rating   Name   Duplication   Size   Complexity  
A TestLink._get_v0x04_ifaces() 0 14 1
A TestLink.test_concurrent_get_next_tag() 0 27 2
A TestLink.test_get_tag_multiple_calls() 0 8 1
A TestLink.test_init_with_null_endpoints() 0 4 2
A TestLink.test_get_next_available_tag() 0 7 1
A TestLink.test_init() 0 6 1
A TestLink.setUp() 0 3 1
A TestLink.test_next_tag__with_use_tags() 0 9 1
A TestLink.test_link_id() 0 5 1
A TestLink.test_tag_life_cicle() 0 12 1
1
"""Link tests."""
2
import logging
3
import time
4
import unittest
5
from unittest.mock import Mock
6
7
from kytos.core.interface import Interface
8
from kytos.core.link import Link
9
from kytos.core.switch import Switch
10
11
logging.basicConfig(level=logging.CRITICAL)
12
13
14
class TestLink(unittest.TestCase):
15
    """Test Links."""
16
17
    def setUp(self):
18
        """Create interface objects."""
19
        self.iface1, self.iface2 = self._get_v0x04_ifaces()
20
21
    @staticmethod
22
    def _get_v0x04_ifaces(*args, **kwargs):
23
        """Create a pair of v0x04 interfaces with optional extra arguments."""
24
        switch1 = Switch('dpid1')
25
        switch1.connection = Mock()
26
        switch1.connection.protocol.version = 0x04
27
        iface1 = Interface('interface1', 41, switch1, *args, **kwargs)
28
29
        switch2 = Switch('dpid2')
30
        switch2.connection = Mock()
31
        switch2.connection.protocol.version = 0x04
32
        iface2 = Interface('interface2', 42, switch2, *args, **kwargs)
33
34
        return iface1, iface2
35
36
    def test_init(self):
37
        """Test normal Link initialization."""
38
        link = Link(self.iface1, self.iface2)
39
        self.assertIsInstance(link, Link)
40
        self.assertIs(link.is_active(), True)
41
        self.assertIs(link.is_enabled(), False)
42
43
    def test_init_with_null_endpoints(self):
44
        """Test initialization with None as endpoints."""
45
        with self.assertRaises(ValueError):
46
            Link(None, None)
47
48
    def test_link_id(self):
49
        """Test equality of links with the same values ​​in different order."""
50
        link1 = Link(self.iface1, self.iface2)
51
        link2 = Link(self.iface2, self.iface1)
52
        self.assertEqual(link1.id, link2.id)
53
54
    def test_get_next_available_tag(self):
55
        """Test get next available tags returns different tags"""
56
        link = Link(self.iface1, self.iface2)
57
        tag = link.get_next_available_tag()
58
        next_tag = link.get_next_available_tag()
59
60
        self.assertNotEqual(tag, next_tag)
61
62
    def test_get_tag_multiple_calls(self):
63
        """Test get next available tags returns different tags"""
64
        link = Link(self.iface1, self.iface2)
65
        tag = link.get_next_available_tag()
66
        self.assertEqual(tag.value, 1)
67
68
        next_tag = link.get_next_available_tag()
69
        self.assertEqual(next_tag.value, 2)
70
71
    def test_next_tag__with_use_tags(self):
72
        """Test get next availabe tags returns different tags"""
73
        link = Link(self.iface1, self.iface2)
74
        tag = link.get_next_available_tag()
75
        self.assertEqual(tag.value, 1)
76
77
        is_available = link.is_tag_available(tag)
78
        self.assertFalse(is_available)
79
        link.use_tag(tag)
80
81
    def test_tag_life_cicle(self):
82
        """Test get next available tags returns different tags"""
83
        link = Link(self.iface1, self.iface2)
84
        tag = link.get_next_available_tag()
85
        self.assertEqual(tag.value, 1)
86
87
        is_available = link.is_tag_available(tag)
88
        self.assertFalse(is_available)
89
90
        link.make_tag_available(tag)
91
        is_available = link.is_tag_available(tag)
92
        self.assertTrue(is_available)
93
94
    def test_concurrent_get_next_tag(self):
95
        """Test get next available tags in concurrent execution"""
96
        from tests.helper import test_concurrently
97
        _link = Link(self.iface1, self.iface2)
98
99
        _i = []
100
101
        @test_concurrently(20)
102
        def test_get_next_available_tag():
103
            """Test get next availabe tags returns different tags"""
104
            _i.append(1)
105
            _i_len = len(_i)
106
            tag = _link.get_next_available_tag()
107
            time.sleep(0.0001)
108
            _link.use_tag(tag)
109
110
            next_tag = _link.get_next_available_tag()
111
            _link.use_tag(next_tag)
112
113
            self.assertNotEqual(tag, next_tag)
114
115
            # Check if in the 20 iteration the tag value is 40
116
            # It happens because we get 2 tags for every iteration
117
            if _i_len == 20:
118
                self.assertEqual(next_tag.value, 40)
119
120
        test_get_next_available_tag()
121