Issues in test_interface.py - All Issues - Inspection of "Fix auth URLs on documentation (#1132)" - kytos/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 6ec3b9...9ccb0e )

by Humberto

created 2020-08-19 15:41 UTC

tests/unit/test_core/test_interface.py (3 issues)

Labels

Severity

Minor 3

"""Interface tests."""
import logging
import unittest
from unittest.mock import MagicMock, Mock

from pyof.v0x04.common.port import PortFeatures

from kytos.core.interface import TAG, UNI, Interface, TAGType
from kytos.core.switch import Switch

logging.basicConfig(level=logging.CRITICAL)


class TestTAG(unittest.TestCase):
    """TAG tests."""

    def setUp(self):
        """Create TAG object."""
        self.tag = TAG(1, 123)

    def test_from_dict(self):
        """Test from_dict method."""
        tag_dict = {'tag_type': 2, 'value': 456}
        tag = self.tag.from_dict(tag_dict)

        self.assertEqual(tag.tag_type, 2)
        self.assertEqual(tag.value, 456)

    def test_as_dict(self):
        """Test as_dict method."""
        self.assertEqual(self.tag.as_dict(), {'tag_type': 1, 'value': 123})

    def test_from_json(self):
        """Test from_json method."""
        tag_json = '{"tag_type": 2, "value": 456}'
        tag = self.tag.from_json(tag_json)

        self.assertEqual(tag.tag_type, 2)
        self.assertEqual(tag.value, 456)

    def test_as_json(self):
        """Test as_json method."""
        self.assertEqual(self.tag.as_json(), '{"tag_type": 1, "value": 123}')

    def test__repr__(self):
        """Test __repr__ method."""
        self.assertEqual(repr(self.tag), 'TAG(<TAGType.VLAN: 1>, 123)')


# pylint: disable=protected-access, too-many-public-methods
class TestInterface(unittest.TestCase):
    """Test Interfaces."""

    def setUp(self):
        """Create interface object."""
        self.iface = self._get_v0x04_iface()

    def test_repr(self):
        """Test repr() output."""
        expected = "Interface('name', 42, Switch('dpid'))"
        self.assertEqual(repr(self.iface), expected)

    @staticmethod
    def _get_v0x04_iface(*args, **kwargs):
        """Create a v0x04 interface object with optional extra arguments."""
        switch = Switch('dpid')
        switch.connection = Mock()
        switch.connection.protocol.version = 0x04
        return Interface('name', 42, switch, *args, **kwargs)

    def test_speed_feature_none(self):
        """When port's current features is None."""
        self.iface.features = None
        self.assertIsNone(self.iface.speed)
        self.assertEqual('', self.iface.get_hr_speed())

    def test_speed_feature_zero(self):
        """When port's current features is 0. E.g. port 65534."""
        self.iface.features = 0
        self.assertIsNone(self.iface.speed)
        self.assertEqual('', self.iface.get_hr_speed())

    def test_1_tera_speed(self):
        """1Tb link."""
        self.iface.features = PortFeatures.OFPPF_1TB_FD
        self.assertEqual(10**12 / 8, self.iface.speed)
        self.assertEqual('1 Tbps', self.iface.get_hr_speed())

    def test_100_giga_speed(self):
        """100Gb link."""
        self.iface.features = PortFeatures.OFPPF_100GB_FD
        self.assertEqual(100 * 10**9 / 8, self.iface.speed)
        self.assertEqual('100 Gbps', self.iface.get_hr_speed())

    def test_40_giga_speed(self):
        """40Gb link."""
        self.iface.features = PortFeatures.OFPPF_40GB_FD
        self.assertEqual(40 * 10**9 / 8, self.iface.speed)
        self.assertEqual('40 Gbps', self.iface.get_hr_speed())

    def test_10_giga_speed(self):
        """10Gb link."""
        self.iface.features = PortFeatures.OFPPF_10GB_FD
        self.assertEqual(10 * 10**9 / 8, self.iface.speed)
        self.assertEqual('10 Gbps', self.iface.get_hr_speed())

    def test_1_giga_speed(self):
        """1Gb link."""
        self.iface.features = PortFeatures.OFPPF_1GB_FD
        self.assertEqual(10**9 / 8, self.iface.speed)
        self.assertEqual('1 Gbps', self.iface.get_hr_speed())

    def test_100_mega_speed(self):
        """100Mb link."""
        self.iface.features = PortFeatures.OFPPF_100MB_FD
        self.assertEqual(100 * 10**6 / 8, self.iface.speed)
        self.assertEqual('100 Mbps', self.iface.get_hr_speed())

    def test_10_mega_speed(self):
        """10Mb link."""
        self.iface.features = PortFeatures.OFPPF_10MB_FD
        self.assertEqual(10 * 10**6 / 8, self.iface.speed)
        self.assertEqual('10 Mbps', self.iface.get_hr_speed())

    def test_speed_setter(self):
        """Should return speed that was set and not features'."""
        expected_speed = 12345
        self.iface.features = PortFeatures.OFPPF_10MB_FD
        self.iface.set_custom_speed(expected_speed)
        actual_speed = self.iface.speed
        self.assertEqual(expected_speed, actual_speed)

    def test_speed_in_constructor(self):
        """Custom speed should override features'."""
        expected_speed = 6789
        iface = self._get_v0x04_iface(speed=expected_speed,
                                      features=PortFeatures.OFPPF_10MB_FD)
        actual_speed = iface.speed
        self.assertEqual(expected_speed, actual_speed)

    def test_remove_custom_speed(self):
        """Should return features' speed again when custom's becomes None."""
        custom_speed = 101112
        of_speed = 10 * 10**6 / 8
        iface = self._get_v0x04_iface(speed=custom_speed,
                                      features=PortFeatures.OFPPF_10MB_FD)
        self.assertEqual(custom_speed, iface.speed)
        iface.set_custom_speed(None)
        self.assertEqual(of_speed, iface.speed)

    def test_interface_available_tags(self):
        """Test available_tags on Interface class."""
        default_range = list(range(1, 4096))

        intf_values = [tag.value for tag in self.iface.available_tags]
        self.assertListEqual(intf_values, default_range)

        custom_range = list(range(100, 199))
        self.iface.set_available_tags(custom_range)
        intf_values = [tag.value for tag in self.iface.available_tags]
        self.assertListEqual(intf_values, custom_range)

    def test_all_available_tags(self):
        """Test all available_tags on Interface class."""
        max_range = 4096

        for i in range(1, max_range):

            next_tag = self.iface.get_next_available_tag()
            self.assertIs(type(next_tag), TAG)
            self.assertEqual(next_tag.value, max_range - i)

        next_tag = self.iface.get_next_available_tag()
        self.assertEqual(next_tag, False)

    def test_interface_is_tag_available(self):
        """Test is_tag_available on Interface class."""
        max_range = 4096
        for i in range(1, max_range):

            tag = TAG(TAGType.VLAN, i)

            next_tag = self.iface.is_tag_available(tag)
            self.assertTrue(next_tag)

        # test lower limit
        tag = TAG(TAGType.VLAN, 0)
        self.assertFalse(self.iface.is_tag_available(tag))
        # test upper limit
        tag = TAG(TAGType.VLAN, max_range)
        self.assertFalse(self.iface.is_tag_available(tag))

    def test_interface_use_tags(self):
        """Test all use_tag on Interface class."""

        tag = TAG(TAGType.VLAN, 100)
        # check use tag for the first time
        is_success = self.iface.use_tag(tag)
        self.assertTrue(is_success)

        # check use tag for the second time
        is_success = self.iface.use_tag(tag)
        self.assertFalse(is_success)

        # check use tag after returning the tag to the pool
        self.iface.make_tag_available(tag)
        is_success = self.iface.use_tag(tag)
        self.assertTrue(is_success)

    def test_enable(self):
        """Test enable method."""
        self.iface.switch = MagicMock()

        self.iface.enable()

        self.iface.switch.enable.assert_called()
        self.assertTrue(self.iface._enabled)

    def test_get_endpoint(self):
        """Test get_endpoint method."""
        endpoint = ('endpoint', 'time')
        self.iface.endpoints = [endpoint]

        return_endpoint = self.iface.get_endpoint('endpoint')

        self.assertEqual(return_endpoint, endpoint)

    def test_add_endpoint(self):
        """Test add_endpoint method."""
        self.iface.add_endpoint('endpoint')

        self.assertEqual(len(self.iface.endpoints), 1)

    def test_delete_endpoint(self):
        """Test delete_endpoint method."""
        endpoint = ('endpoint', 'time')
        self.iface.endpoints = [endpoint]

        self.iface.delete_endpoint('endpoint')

        self.assertEqual(len(self.iface.endpoints), 0)

    def test_update_endpoint(self):
        """Test update_endpoint method."""
        endpoint = ('endpoint', 'time')
        self.iface.endpoints = [endpoint]

        self.iface.update_endpoint('endpoint')

        self.assertEqual(len(self.iface.endpoints), 1)

    def test_update_link__none(self):
        """Test update_link method when this interface is not in link
           endpoints."""
        link = MagicMock()
        link.endpoint_a = MagicMock()
        link.endpoint_b = MagicMock()

        result = self.iface.update_link(link)

        self.assertFalse(result)

    def test_update_link__endpoint_a(self):
        """Test update_link method when this interface is the endpoint a."""
        interface = MagicMock()
        interface.link = None
        link = MagicMock()
        link.endpoint_a = self.iface
        link.endpoint_b = interface

        self.iface.update_link(link)

        self.assertEqual(self.iface.link, link)
        self.assertEqual(interface.link, link)

    def test_update_link__endpoint_b(self):
        """Test update_link method when this interface is the endpoint b."""
        interface = MagicMock()
        interface.link = None
        link = MagicMock()
        link.endpoint_a = interface
        link.endpoint_b = self.iface

        self.iface.update_link(link)

        self.assertEqual(self.iface.link, link)
        self.assertEqual(interface.link, link)


class TestUNI(unittest.TestCase):
    """UNI tests."""

    def setUp(self):
        """Create UNI object."""
        switch = MagicMock()
        switch.dpid = '00:00:00:00:00:00:00:01'
        interface = Interface('name', 1, switch)
        user_tag = TAG(1, 123)
        self.uni = UNI(interface, user_tag)

    def test__eq__(self):
        """Test __eq__ method."""
        user_tag = TAG(2, 456)
        interface = Interface('name', 2, MagicMock())
        other = UNI(interface, user_tag)

        self.assertFalse(self.uni.__eq__(other))

    def test_is_valid(self):
        """Test is_valid method for a valid, invalid and none tag."""
        self.assertTrue(self.uni.is_valid())

        with self.assertRaises(ValueError):
            TAG(999999, 123)

        self.uni.user_tag = None
        self.assertTrue(self.uni.is_valid())

    def test_as_dict(self):
        """Test as_dict method."""
        expected_dict = {'interface_id': '00:00:00:00:00:00:00:01:1',
                         'tag': {'tag_type': 1, 'value': 123}}
        self.assertEqual(self.uni.as_dict(), expected_dict)

    def test_as_json(self):
        """Test as_json method."""
        expected_json = '{"interface_id": "00:00:00:00:00:00:00:01:1", ' + \
                        '"tag": {"tag_type": 1, "value": 123}}'
        self.assertEqual(self.uni.as_json(), expected_json)


1			"""Interface tests."""
2			import logging
3			import unittest
4			from unittest.mock import MagicMock, Mock
5
6			from pyof.v0x04.common.port import PortFeatures
7
8			from kytos.core.interface import TAG, UNI, Interface, TAGType
9			from kytos.core.switch import Switch
10
11			logging.basicConfig(level=logging.CRITICAL)
12
13
14			class TestTAG(unittest.TestCase):
15			"""TAG tests."""
16
17			def setUp(self):
18			"""Create TAG object."""
19			self.tag = TAG(1, 123)
20
21			def test_from_dict(self):
22			"""Test from_dict method."""
23			tag_dict = {'tag_type': 2, 'value': 456}
24			tag = self.tag.from_dict(tag_dict)
25
26			self.assertEqual(tag.tag_type, 2)
27			self.assertEqual(tag.value, 456)
28
29			def test_as_dict(self):
30			"""Test as_dict method."""
31			self.assertEqual(self.tag.as_dict(), {'tag_type': 1, 'value': 123})
32
33			def test_from_json(self):
34			"""Test from_json method."""
35			tag_json = '{"tag_type": 2, "value": 456}'
36			tag = self.tag.from_json(tag_json)
37
38			self.assertEqual(tag.tag_type, 2)
39			self.assertEqual(tag.value, 456)
40
41			def test_as_json(self):
42			"""Test as_json method."""
43			self.assertEqual(self.tag.as_json(), '{"tag_type": 1, "value": 123}')
44
45			def test__repr__(self):
46			"""Test __repr__ method."""
47			self.assertEqual(repr(self.tag), 'TAG(<TAGType.VLAN: 1>, 123)')
48
49
50			# pylint: disable=protected-access, too-many-public-methods
51			class TestInterface(unittest.TestCase):
52			"""Test Interfaces."""
53
54			def setUp(self):
55			"""Create interface object."""
56			self.iface = self._get_v0x04_iface()
57
58			def test_repr(self):
59			"""Test repr() output."""
60			expected = "Interface('name', 42, Switch('dpid'))"
61			self.assertEqual(repr(self.iface), expected)
62
63			@staticmethod
64			def _get_v0x04_iface(args, *kwargs):
65			"""Create a v0x04 interface object with optional extra arguments."""
66			switch = Switch('dpid')
67			switch.connection = Mock()
68			switch.connection.protocol.version = 0x04
69			return Interface('name', 42, switch, args, *kwargs)
70
71			def test_speed_feature_none(self):
72			"""When port's current features is None."""
73			self.iface.features = None
74			self.assertIsNone(self.iface.speed)
75			self.assertEqual('', self.iface.get_hr_speed())
76
77			def test_speed_feature_zero(self):
78			"""When port's current features is 0. E.g. port 65534."""
79			self.iface.features = 0
80			self.assertIsNone(self.iface.speed)
81			self.assertEqual('', self.iface.get_hr_speed())
82
83			def test_1_tera_speed(self):
84			"""1Tb link."""
85			self.iface.features = PortFeatures.OFPPF_1TB_FD
86			self.assertEqual(10**12 / 8, self.iface.speed)
87			self.assertEqual('1 Tbps', self.iface.get_hr_speed())
88
89			def test_100_giga_speed(self):
90			"""100Gb link."""
91			self.iface.features = PortFeatures.OFPPF_100GB_FD
92			self.assertEqual(100 * 10**9 / 8, self.iface.speed)
93			self.assertEqual('100 Gbps', self.iface.get_hr_speed())
94
95			def test_40_giga_speed(self):
96			"""40Gb link."""
97			self.iface.features = PortFeatures.OFPPF_40GB_FD
98			self.assertEqual(40 * 10**9 / 8, self.iface.speed)
99			self.assertEqual('40 Gbps', self.iface.get_hr_speed())
100
101			def test_10_giga_speed(self):
102			"""10Gb link."""
103			self.iface.features = PortFeatures.OFPPF_10GB_FD
104			self.assertEqual(10 * 10**9 / 8, self.iface.speed)
105			self.assertEqual('10 Gbps', self.iface.get_hr_speed())
106
107			def test_1_giga_speed(self):
108			"""1Gb link."""
109			self.iface.features = PortFeatures.OFPPF_1GB_FD
110			self.assertEqual(10**9 / 8, self.iface.speed)
111			self.assertEqual('1 Gbps', self.iface.get_hr_speed())
112
113			def test_100_mega_speed(self):
114			"""100Mb link."""
115			self.iface.features = PortFeatures.OFPPF_100MB_FD
116			self.assertEqual(100 * 10**6 / 8, self.iface.speed)
117			self.assertEqual('100 Mbps', self.iface.get_hr_speed())
118
119			def test_10_mega_speed(self):
120			"""10Mb link."""
121			self.iface.features = PortFeatures.OFPPF_10MB_FD
122			self.assertEqual(10 * 10**6 / 8, self.iface.speed)
123			self.assertEqual('10 Mbps', self.iface.get_hr_speed())
124
125			def test_speed_setter(self):
126			"""Should return speed that was set and not features'."""
127			expected_speed = 12345
128			self.iface.features = PortFeatures.OFPPF_10MB_FD
129			self.iface.set_custom_speed(expected_speed)
130			actual_speed = self.iface.speed
131			self.assertEqual(expected_speed, actual_speed)
132
133			def test_speed_in_constructor(self):
134			"""Custom speed should override features'."""
135			expected_speed = 6789
136			iface = self._get_v0x04_iface(speed=expected_speed,
137			features=PortFeatures.OFPPF_10MB_FD)
138			actual_speed = iface.speed
139			self.assertEqual(expected_speed, actual_speed)
140
141			def test_remove_custom_speed(self):
142			"""Should return features' speed again when custom's becomes None."""
143			custom_speed = 101112
144			of_speed = 10 * 10**6 / 8
145			iface = self._get_v0x04_iface(speed=custom_speed,
146			features=PortFeatures.OFPPF_10MB_FD)
147			self.assertEqual(custom_speed, iface.speed)
148			iface.set_custom_speed(None)
149			self.assertEqual(of_speed, iface.speed)
150
151			def test_interface_available_tags(self):
152			"""Test available_tags on Interface class."""
153			default_range = list(range(1, 4096))
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-04-13 14:12 UTC by Report Bug Copy Issue Report Show Similar Issues like this The variable `range` does not seem to be defined. Loading history...
154			intf_values = [tag.value for tag in self.iface.available_tags]
155			self.assertListEqual(intf_values, default_range)
156
157			custom_range = list(range(100, 199))
158			self.iface.set_available_tags(custom_range)
159			intf_values = [tag.value for tag in self.iface.available_tags]
160			self.assertListEqual(intf_values, custom_range)
161
162			def test_all_available_tags(self):
163			"""Test all available_tags on Interface class."""
164			max_range = 4096
165
166			for i in range(1, max_range):
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-07-02 16:13 UTC by Report Bug Copy Issue Report Show Similar Issues like this The variable `range` does not seem to be defined. Loading history...
167			next_tag = self.iface.get_next_available_tag()
168			self.assertIs(type(next_tag), TAG)
169			self.assertEqual(next_tag.value, max_range - i)
170
171			next_tag = self.iface.get_next_available_tag()
172			self.assertEqual(next_tag, False)
173
174			def test_interface_is_tag_available(self):
175			"""Test is_tag_available on Interface class."""
176			max_range = 4096
177			for i in range(1, max_range):
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-07-02 16:13 UTC by Report Bug Copy Issue Report Show Similar Issues like this The variable `range` does not seem to be defined. Loading history...
178			tag = TAG(TAGType.VLAN, i)
179
180			next_tag = self.iface.is_tag_available(tag)
181			self.assertTrue(next_tag)
182
183			# test lower limit
184			tag = TAG(TAGType.VLAN, 0)
185			self.assertFalse(self.iface.is_tag_available(tag))
186			# test upper limit
187			tag = TAG(TAGType.VLAN, max_range)
188			self.assertFalse(self.iface.is_tag_available(tag))
189
190			def test_interface_use_tags(self):
191			"""Test all use_tag on Interface class."""
192
193			tag = TAG(TAGType.VLAN, 100)
194			# check use tag for the first time
195			is_success = self.iface.use_tag(tag)
196			self.assertTrue(is_success)
197
198			# check use tag for the second time
199			is_success = self.iface.use_tag(tag)
200			self.assertFalse(is_success)
201
202			# check use tag after returning the tag to the pool
203			self.iface.make_tag_available(tag)
204			is_success = self.iface.use_tag(tag)
205			self.assertTrue(is_success)
206
207			def test_enable(self):
208			"""Test enable method."""
209			self.iface.switch = MagicMock()
210
211			self.iface.enable()
212
213			self.iface.switch.enable.assert_called()
214			self.assertTrue(self.iface._enabled)
215
216			def test_get_endpoint(self):
217			"""Test get_endpoint method."""
218			endpoint = ('endpoint', 'time')
219			self.iface.endpoints = [endpoint]
220
221			return_endpoint = self.iface.get_endpoint('endpoint')
222
223			self.assertEqual(return_endpoint, endpoint)
224
225			def test_add_endpoint(self):
226			"""Test add_endpoint method."""
227			self.iface.add_endpoint('endpoint')
228
229			self.assertEqual(len(self.iface.endpoints), 1)
230
231			def test_delete_endpoint(self):
232			"""Test delete_endpoint method."""
233			endpoint = ('endpoint', 'time')
234			self.iface.endpoints = [endpoint]
235
236			self.iface.delete_endpoint('endpoint')
237
238			self.assertEqual(len(self.iface.endpoints), 0)
239
240			def test_update_endpoint(self):
241			"""Test update_endpoint method."""
242			endpoint = ('endpoint', 'time')
243			self.iface.endpoints = [endpoint]
244
245			self.iface.update_endpoint('endpoint')
246
247			self.assertEqual(len(self.iface.endpoints), 1)
248
249			def test_update_link__none(self):
250			"""Test update_link method when this interface is not in link
251			endpoints."""
252			link = MagicMock()
253			link.endpoint_a = MagicMock()
254			link.endpoint_b = MagicMock()
255
256			result = self.iface.update_link(link)
257
258			self.assertFalse(result)
259
260			def test_update_link__endpoint_a(self):
261			"""Test update_link method when this interface is the endpoint a."""
262			interface = MagicMock()
263			interface.link = None
264			link = MagicMock()
265			link.endpoint_a = self.iface
266			link.endpoint_b = interface
267
268			self.iface.update_link(link)
269
270			self.assertEqual(self.iface.link, link)
271			self.assertEqual(interface.link, link)
272
273			def test_update_link__endpoint_b(self):
274			"""Test update_link method when this interface is the endpoint b."""
275			interface = MagicMock()
276			interface.link = None
277			link = MagicMock()
278			link.endpoint_a = interface
279			link.endpoint_b = self.iface
280
281			self.iface.update_link(link)
282
283			self.assertEqual(self.iface.link, link)
284			self.assertEqual(interface.link, link)
285
286
287			class TestUNI(unittest.TestCase):
288			"""UNI tests."""
289
290			def setUp(self):
291			"""Create UNI object."""
292			switch = MagicMock()
293			switch.dpid = '00:00:00:00:00:00:00:01'
294			interface = Interface('name', 1, switch)
295			user_tag = TAG(1, 123)
296			self.uni = UNI(interface, user_tag)
297
298			def test__eq__(self):
299			"""Test __eq__ method."""
300			user_tag = TAG(2, 456)
301			interface = Interface('name', 2, MagicMock())
302			other = UNI(interface, user_tag)
303
304			self.assertFalse(self.uni.__eq__(other))
305
306			def test_is_valid(self):
307			"""Test is_valid method for a valid, invalid and none tag."""
308			self.assertTrue(self.uni.is_valid())
309
310			with self.assertRaises(ValueError):
311			TAG(999999, 123)
312
313			self.uni.user_tag = None
314			self.assertTrue(self.uni.is_valid())
315
316			def test_as_dict(self):
317			"""Test as_dict method."""
318			expected_dict = {'interface_id': '00:00:00:00:00:00:00:01:1',
319			'tag': {'tag_type': 1, 'value': 123}}
320			self.assertEqual(self.uni.as_dict(), expected_dict)
321
322			def test_as_json(self):
323			"""Test as_json method."""
324			expected_json = '{"interface_id": "00:00:00:00:00:00:00:01:1", ' + \
325			'"tag": {"tag_type": 1, "value": 123}}'
326			self.assertEqual(self.uni.as_json(), expected_json)
327

kytos / kytos

Push — master ( 6ec3b9...9ccb0e )

tests/unit/test_core/test_interface.py (3 issues)

Labels

Severity

Introduced By

Duplication Side-by-Side

Filter issues like