1 | """Interface tests.""" |
||
2 | import logging |
||
3 | import unittest |
||
4 | from unittest.mock import Mock |
||
5 | |||
6 | from pyof.v0x04.common.port import PortFeatures |
||
7 | |||
8 | from kytos.core.interface import TAG, Interface, TAGType |
||
9 | from kytos.core.switch import Switch |
||
10 | |||
11 | logging.basicConfig(level=logging.CRITICAL) |
||
12 | |||
13 | |||
14 | class TestInterface(unittest.TestCase): |
||
15 | """Test Interfaces.""" |
||
16 | |||
17 | def setUp(self): |
||
18 | """Create interface object.""" |
||
19 | self.iface = self._get_v0x04_iface() |
||
20 | |||
21 | def test_repr(self): |
||
22 | """Test repr() output.""" |
||
23 | expected = "Interface('name', 42, Switch('dpid'))" |
||
24 | self.assertEqual(repr(self.iface), expected) |
||
25 | |||
26 | @staticmethod |
||
27 | def _get_v0x04_iface(*args, **kwargs): |
||
28 | """Create a v0x04 interface object with optional extra arguments.""" |
||
29 | switch = Switch('dpid') |
||
30 | switch.connection = Mock() |
||
31 | switch.connection.protocol.version = 0x04 |
||
32 | return Interface('name', 42, switch, *args, **kwargs) |
||
33 | |||
34 | def test_speed_feature_none(self): |
||
35 | """When port's current features is None.""" |
||
36 | self.iface.features = None |
||
37 | self.assertIsNone(self.iface.speed) |
||
38 | self.assertEqual('', self.iface.get_hr_speed()) |
||
39 | |||
40 | def test_speed_feature_zero(self): |
||
41 | """When port's current features is 0. E.g. port 65534.""" |
||
42 | self.iface.features = 0 |
||
43 | self.assertIsNone(self.iface.speed) |
||
44 | self.assertEqual('', self.iface.get_hr_speed()) |
||
45 | |||
46 | def test_1_tera_speed(self): |
||
47 | """1Tb link.""" |
||
48 | self.iface.features = PortFeatures.OFPPF_1TB_FD |
||
49 | self.assertEqual(10**12 / 8, self.iface.speed) |
||
50 | self.assertEqual('1 Tbps', self.iface.get_hr_speed()) |
||
51 | |||
52 | def test_100_giga_speed(self): |
||
53 | """100Gb link.""" |
||
54 | self.iface.features = PortFeatures.OFPPF_100GB_FD |
||
55 | self.assertEqual(100 * 10**9 / 8, self.iface.speed) |
||
56 | self.assertEqual('100 Gbps', self.iface.get_hr_speed()) |
||
57 | |||
58 | def test_40_giga_speed(self): |
||
59 | """40Gb link.""" |
||
60 | self.iface.features = PortFeatures.OFPPF_40GB_FD |
||
61 | self.assertEqual(40 * 10**9 / 8, self.iface.speed) |
||
62 | self.assertEqual('40 Gbps', self.iface.get_hr_speed()) |
||
63 | |||
64 | def test_10_giga_speed(self): |
||
65 | """10Gb link.""" |
||
66 | self.iface.features = PortFeatures.OFPPF_10GB_FD |
||
67 | self.assertEqual(10 * 10**9 / 8, self.iface.speed) |
||
68 | self.assertEqual('10 Gbps', self.iface.get_hr_speed()) |
||
69 | |||
70 | def test_1_giga_speed(self): |
||
71 | """1Gb link.""" |
||
72 | self.iface.features = PortFeatures.OFPPF_1GB_FD |
||
73 | self.assertEqual(10**9 / 8, self.iface.speed) |
||
74 | self.assertEqual('1 Gbps', self.iface.get_hr_speed()) |
||
75 | |||
76 | def test_100_mega_speed(self): |
||
77 | """100Mb link.""" |
||
78 | self.iface.features = PortFeatures.OFPPF_100MB_FD |
||
79 | self.assertEqual(100 * 10**6 / 8, self.iface.speed) |
||
80 | self.assertEqual('100 Mbps', self.iface.get_hr_speed()) |
||
81 | |||
82 | def test_10_mega_speed(self): |
||
83 | """10Mb link.""" |
||
84 | self.iface.features = PortFeatures.OFPPF_10MB_FD |
||
85 | self.assertEqual(10 * 10**6 / 8, self.iface.speed) |
||
86 | self.assertEqual('10 Mbps', self.iface.get_hr_speed()) |
||
87 | |||
88 | def test_speed_setter(self): |
||
89 | """Should return speed that was set and not features'.""" |
||
90 | expected_speed = 12345 |
||
91 | self.iface.features = PortFeatures.OFPPF_10MB_FD |
||
92 | self.iface.set_custom_speed(expected_speed) |
||
93 | actual_speed = self.iface.speed |
||
94 | self.assertEqual(expected_speed, actual_speed) |
||
95 | |||
96 | def test_speed_in_constructor(self): |
||
97 | """Custom speed should override features'.""" |
||
98 | expected_speed = 6789 |
||
99 | iface = self._get_v0x04_iface(speed=expected_speed, |
||
100 | features=PortFeatures.OFPPF_10MB_FD) |
||
101 | actual_speed = iface.speed |
||
102 | self.assertEqual(expected_speed, actual_speed) |
||
103 | |||
104 | def test_remove_custom_speed(self): |
||
105 | """Should return features' speed again when custom's becomes None.""" |
||
106 | custom_speed = 101112 |
||
107 | of_speed = 10 * 10**6 / 8 |
||
108 | iface = self._get_v0x04_iface(speed=custom_speed, |
||
109 | features=PortFeatures.OFPPF_10MB_FD) |
||
110 | self.assertEqual(custom_speed, iface.speed) |
||
111 | iface.set_custom_speed(None) |
||
112 | self.assertEqual(of_speed, iface.speed) |
||
113 | |||
114 | def test_interface_available_tags(self): |
||
115 | """Test available_tags on Interface class.""" |
||
116 | default_range = [vlan for vlan in range(1, 4096)] |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
Loading history...
|
|||
117 | intf_values = [tag.value for tag in self.iface.available_tags] |
||
118 | self.assertListEqual(intf_values, default_range) |
||
119 | |||
120 | custom_range = [vlan for vlan in range(100, 199)] |
||
121 | self.iface.set_available_tags(custom_range) |
||
122 | intf_values = [tag.value for tag in self.iface.available_tags] |
||
123 | self.assertListEqual(intf_values, custom_range) |
||
124 | |||
125 | def test_all_available_tags(self): |
||
126 | """Test all available_tags on Interface class.""" |
||
127 | max_range = 4096 |
||
128 | |||
129 | for i in range(1, max_range): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
130 | next_tag = self.iface.get_next_available_tag() |
||
131 | self.assertIs(type(next_tag), TAG) |
||
132 | self.assertEqual(next_tag.value, max_range - i) |
||
133 | |||
134 | next_tag = self.iface.get_next_available_tag() |
||
135 | self.assertEqual(next_tag, False) |
||
136 | |||
137 | def test_interface_is_tag_available(self): |
||
138 | """Test is_tag_available on Interface class.""" |
||
139 | max_range = 4096 |
||
140 | for i in range(1, max_range): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
141 | tag = TAG(TAGType.VLAN, i) |
||
142 | |||
143 | next_tag = self.iface.is_tag_available(tag) |
||
144 | self.assertTrue(next_tag) |
||
145 | |||
146 | # test lower limit |
||
147 | tag = TAG(TAGType.VLAN, 0) |
||
148 | self.assertFalse(self.iface.is_tag_available(tag)) |
||
149 | # test upper limit |
||
150 | tag = TAG(TAGType.VLAN, max_range) |
||
151 | self.assertFalse(self.iface.is_tag_available(tag)) |
||
152 | |||
153 | def test_interface_use_tags(self): |
||
154 | """Test all use_tag on Interface class.""" |
||
155 | |||
156 | tag = TAG(TAGType.VLAN, 100) |
||
157 | # check use tag for the first time |
||
158 | is_success = self.iface.use_tag(tag) |
||
159 | self.assertTrue(is_success) |
||
160 | |||
161 | # check use tag for the second time |
||
162 | is_success = self.iface.use_tag(tag) |
||
163 | self.assertFalse(is_success) |
||
164 | |||
165 | # check use tag after returning the tag to the pool |
||
166 | self.iface.make_tag_available(tag) |
||
167 | is_success = self.iface.use_tag(tag) |
||
168 | self.assertTrue(is_success) |
||
169 |