kytos.core.interface   F
last analyzed

Complexity

Total Complexity 87

Size/Duplication

Total Lines 472
Duplicated Lines 0 %

Test Coverage

Coverage 71.86%

Importance

Changes 0
Metric Value
eloc 224
dl 0
loc 472
ccs 143
cts 199
cp 0.7186
rs 2
c 0
b 0
f 0
wmc 87

40 Methods

Rating   Name   Duplication   Size   Complexity  
A TAG.from_dict() 0 4 1
A Interface.id() 0 9 1
A Interface.uni() 0 4 1
A TAG.__init__() 0 3 1
A TAG.from_json() 0 4 1
A TAG.as_json() 0 3 1
A TAG.as_dict() 0 3 1
A TAG.__repr__() 0 2 1
A TAG.__eq__() 0 2 1
A Interface.delete_endpoint() 0 9 2
B Interface._get_v0x04_speed() 0 14 7
A Interface.add_endpoint() 0 9 2
C Interface._get_v0x01_v0x04_speed() 0 13 9
A Interface.make_tag_available() 0 6 2
A Interface.get_endpoint() 0 14 3
A Interface._is_v0x04() 0 4 1
A Interface.__init__() 0 32 1
A UNI.__eq__() 0 4 1
A Interface.is_tag_available() 0 3 1
A Interface.set_available_tags() 0 12 2
B Interface.update_link() 0 24 7
A Interface.__repr__() 0 2 1
A Interface.as_dict() 0 44 3
A Interface.as_json() 0 20 1
A Interface.get_next_available_tag() 0 11 2
A Interface.__eq__() 0 8 3
A Interface.set_custom_speed() 0 7 1
A NNI.__init__() 0 2 1
A Interface.enable() 0 7 1
B Interface.speed() 0 23 6
A VNNI.__init__() 0 4 1
A Interface.get_hr_speed() 0 16 4
A UNI.is_valid() 0 5 2
A Interface.get_custom_speed() 0 3 1
A UNI.as_json() 0 3 1
B Interface.get_of_features_speed() 0 27 6
A Interface.use_tag() 0 9 2
A UNI.__init__() 0 3 1
A Interface.update_endpoint() 0 10 2
A UNI.as_dict() 0 4 1

How to fix   Complexity   

Complexity

Complex classes like kytos.core.interface often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

1
"""Module with main classes related to Interfaces."""
2 1
import json
3 1
import logging
4 1
from enum import IntEnum
5
6 1
from pyof.v0x01.common.phy_port import Port as PortNo01
7 1
from pyof.v0x01.common.phy_port import PortFeatures as PortFeatures01
8 1
from pyof.v0x04.common.port import PortFeatures as PortFeatures04
9 1
from pyof.v0x04.common.port import PortNo as PortNo04
10
11 1
from kytos.core.common import GenericEntity
12 1
from kytos.core.helpers import now
13
14 1
__all__ = ('Interface',)
15
16 1
LOG = logging.getLogger(__name__)
17
18
19 1
class TAGType(IntEnum):
20
    """Class that represents a TAG Type."""
21
22 1
    VLAN = 1
23 1
    VLAN_QINQ = 2
24 1
    MPLS = 3
25
26
27 1
class TAG:
28
    """Class that represents a TAG."""
29
30 1
    def __init__(self, tag_type, value):
31 1
        self.tag_type = tag_type
32 1
        self.value = value
33
34 1
    def __eq__(self, other):
35 1
        return self.tag_type == other.tag_type and self.value == other.value
36
37 1
    def as_dict(self):
38
        """Return a dictionary representating a tag object."""
39
        return {'tag_type': self.tag_type, 'value': self.value}
40
41 1
    @classmethod
42
    def from_dict(cls, tag_dict):
43
        """Return a TAG instance from python dictionary."""
44
        return cls(tag_dict.get('tag_type'), tag_dict.get('value'))
45
46 1
    @classmethod
47
    def from_json(cls, tag_json):
48
        """Return a TAG instance from json."""
49
        return cls.from_dict(json.loads(tag_json))
50
51 1
    def as_json(self):
52
        """Return a json representating a tag object."""
53
        return json.dumps(self.as_dict())
54
55 1
    def __repr__(self):
56
        return f"TAG({self.tag_type!r}, {self.value!r})"
57
58
59 1
class Interface(GenericEntity):  # pylint: disable=too-many-instance-attributes
60
    """Interface Class used to abstract the network interfaces."""
61
62
    # pylint: disable=too-many-arguments
63 1
    def __init__(self, name, port_number, switch, address=None, state=None,
64
                 features=None, speed=None):
65
        """Assign the parameters to instance attributes.
66
67
        Args:
68
            name (string): name from this interface.
69
            port_number (int): port number from this interface.
70
            switch (:class:`~.core.switch.Switch`): Switch with this interface.
71
            address (|hw_address|): Port address from this interface.
72
            state (|port_stats|): Port Stat from interface. It will be
73
            deprecated.
74
            features (|port_features|): Port feature used to calculate link
75
                utilization from this interface. It will be deprecated.
76
            speed (int, float): Interface speed in bytes per second. Defaults
77
                to what is informed by the switch. Return ``None`` if not set
78
                and switch does not inform the speed.
79
        """
80 1
        self.name = name
81 1
        self.port_number = int(port_number)
82 1
        self.switch = switch
83 1
        self.address = address
84 1
        self.state = state
85 1
        self.features = features
86 1
        self.nni = False
87 1
        self.endpoints = []
88 1
        self.stats = None
89 1
        self.link = None
90 1
        self.lldp = True
91 1
        self._custom_speed = speed
92 1
        self.set_available_tags(range(1, 4096))
0 ignored issues
show
Comprehensibility Best Practice introduced by Vinicius Arcanjo
The variable range does not seem to be defined.
Loading history...
93
94 1
        super().__init__()
95
96 1
    def __repr__(self):
97 1
        return f"Interface('{self.name}', {self.port_number}, {self.switch!r})"
98
99 1
    def __eq__(self, other):
100
        """Compare Interface class with another instance."""
101 1
        if isinstance(other, str):
102
            return self.address == other
103 1
        if isinstance(other, Interface):
104 1
            return self.port_number == other.port_number and \
105
                self.switch.dpid == other.switch.dpid
106
        return False
107
108 1
    @property
109
    def id(self):  # pylint: disable=invalid-name
110
        """Return id from Interface instance.
111
112
        Returns:
113
            string: Interface id.
114
115
        """
116
        return "{}:{}".format(self.switch.dpid, self.port_number)
117
118 1
    @property
119
    def uni(self):
120
        """Return if an interface is a user-to-network Interface."""
121
        return not self.nni
122
123 1
    def set_available_tags(self, iterable):
124
        """Set a range of VLAN tags to be used by this Interface.
125
126
        Args:
127
            iterable ([int]): range of VLANs.
128
        """
129 1
        self.available_tags = []
130
131 1
        for i in iterable:
132 1
            vlan = TAGType.VLAN
133 1
            tag = TAG(vlan, i)
134 1
            self.available_tags.append(tag)
135
136 1
    def enable(self):
137
        """Enable this interface instance.
138
139
        Also enable the switch instance this interface is attached to.
140
        """
141
        self.switch.enable()
142
        self._enabled = True
143
144 1
    def use_tag(self, tag):
145
        """Remove a specific tag from available_tags if it is there.
146
147
        Return False in case the tag is already removed.
148
        """
149 1
        if tag in self.available_tags:
150 1
            self.available_tags.remove(tag)
151 1
            return True
152 1
        return False
153
154 1
    def is_tag_available(self, tag):
155
        """Check if a tag is available."""
156 1
        return tag in self.available_tags
157
158 1
    def get_next_available_tag(self):
159
        """Get the next available tag from the interface.
160
161
        Return the next available tag if exists and remove from the
162
        available tags.
163
        If no tag is available return False.
164
        """
165 1
        try:
166 1
            return self.available_tags.pop()
167 1
        except IndexError:
168 1
            return False
169
170 1
    def make_tag_available(self, tag):
171
        """Add a specific tag in available_tags."""
172 1
        if not self.is_tag_available(tag):
173 1
            self.available_tags.append(tag)
174 1
            return True
175
        return False
176
177 1
    def get_endpoint(self, endpoint):
178
        """Return a tuple with existent endpoint, None otherwise.
179
180
        Args:
181
            endpoint(|hw_address|, :class:`.Interface`): endpoint instance.
182
183
        Returns:
184
            tuple: A tuple with endpoint and time of last update.
185
186
        """
187
        for item in self.endpoints:
188
            if endpoint == item[0]:
189
                return item
190
        return None
191
192 1
    def add_endpoint(self, endpoint):
193
        """Create a new endpoint to Interface instance.
194
195
        Args:
196
            endpoint(|hw_address|, :class:`.Interface`): A target endpoint.
197
        """
198
        exists = self.get_endpoint(endpoint)
199
        if not exists:
200
            self.endpoints.append((endpoint, now()))
201
202 1
    def delete_endpoint(self, endpoint):
203
        """Delete a existent endpoint in Interface instance.
204
205
        Args:
206
            endpoint (|hw_address|, :class:`.Interface`): A target endpoint.
207
        """
208
        exists = self.get_endpoint(endpoint)
209
        if exists:
210
            self.endpoints.remove(exists)
211
212 1
    def update_endpoint(self, endpoint):
213
        """Update or create new endpoint to Interface instance.
214
215
        Args:
216
            endpoint(|hw_address|, :class:`.Interface`): A target endpoint.
217
        """
218
        exists = self.get_endpoint(endpoint)
219
        if exists:
220
            self.delete_endpoint(endpoint)
221
        self.add_endpoint(endpoint)
222
223 1
    def update_link(self, link):
224
        """Update link for this interface in a consistent way.
225
226
        Verify of the other endpoint of the link has the same Link information
227
        attached to it, and change it if necessary.
228
229
        Warning: This method can potentially change information of other
230
        Interface instances. Use it with caution.
231
        """
232
        if self not in (link.endpoint_a, link.endpoint_b):
233
            return False
234
235
        if self.link is None or self.link != link:
236
            self.link = link
237
238
        if link.endpoint_a == self:
239
            endpoint = link.endpoint_b
240
        else:
241
            endpoint = link.endpoint_a
242
243
        if endpoint.link is None or endpoint.link != link:
244
            endpoint.link = link
245
246
        return True
247
248 1
    @property
249
    def speed(self):
250
        """Return the link speed in bytes per second, None otherwise.
251
252
        If the switch was disconnected, we have :attr:`features` and speed is
253
        still returned for common values between v0x01 and v0x04. For specific
254
        v0x04 values (40 Gbps, 100 Gbps and 1 Tbps), the connection must be
255
        active so we can make sure the protocol version is v0x04.
256
257
        Returns:
258
            int, None: Link speed in bytes per second or ``None``.
259
260
        """
261 1
        if self._custom_speed is not None:
262 1
            return self._custom_speed
263
264 1
        if self._is_v0x04() and self.port_number == PortNo04.OFPP_LOCAL:
265
            return 0
266
267 1
        if not self._is_v0x04() and self.port_number == PortNo01.OFPP_LOCAL:
268
            return 0
269
270 1
        return self.get_of_features_speed()
271
272 1
    def set_custom_speed(self, bytes_per_second):
273
        """Set a speed that overrides switch OpenFlow information.
274
275
        If ``None`` is given, :attr:`speed` becomes the one given by the
276
        switch.
277
        """
278 1
        self._custom_speed = bytes_per_second
279
280 1
    def get_custom_speed(self):
281
        """Return custom speed or ``None`` if not set."""
282
        return self._custom_speed
283
284 1
    def get_of_features_speed(self):
285
        """Return the link speed in bytes per second, None otherwise.
286
287
        If the switch was disconnected, we have :attr:`features` and speed is
288
        still returned for common values between v0x01 and v0x04. For specific
289
        v0x04 values (40 Gbps, 100 Gbps and 1 Tbps), the connection must be
290
        active so we can make sure the protocol version is v0x04.
291
292
        Returns:
293
            int, None: Link speed in bytes per second or ``None``.
294
295
        """
296 1
        speed = self._get_v0x01_v0x04_speed()
297
        # Don't use switch.is_connected() because we can have the protocol
298 1
        if speed is None and self._is_v0x04():
299 1
            speed = self._get_v0x04_speed()
300 1
        if speed is not None:
301 1
            return speed
302
        # Warn unknown speed
303
        # Use shorter switch ID with its beginning and end
304 1
        if isinstance(self.switch.id, str) and len(self.switch.id) > 20:
305
            switch_id = self.switch.id[:3] + '...' + self.switch.id[-3:]
306
        else:
307 1
            switch_id = self.switch.id
308 1
        LOG.warning("Couldn't get port %s speed, sw %s, feats %s",
309
                    self.port_number, switch_id, self.features)
310 1
        return None
311
312 1
    def _is_v0x04(self):
313
        """Whether the switch is connected using OpenFlow 1.3."""
314 1
        return self.switch.is_connected() and \
315
            self.switch.connection.protocol.version == 0x04
316
317 1
    def _get_v0x01_v0x04_speed(self):
318
        """Check against all values of v0x01. They're part of v0x04."""
319 1
        fts = self.features
320 1
        pfts = PortFeatures01
321 1
        if fts and fts & pfts.OFPPF_10GB_FD:
322 1
            return 10 * 10**9 / 8
323 1
        if fts and fts & (pfts.OFPPF_1GB_HD | pfts.OFPPF_1GB_FD):
324 1
            return 10**9 / 8
325 1
        if fts and fts & (pfts.OFPPF_100MB_HD | pfts.OFPPF_100MB_FD):
326 1
            return 100 * 10**6 / 8
327 1
        if fts and fts & (pfts.OFPPF_10MB_HD | pfts.OFPPF_10MB_FD):
328 1
            return 10 * 10**6 / 8
329 1
        return None
330
331 1
    def _get_v0x04_speed(self):
332
        """Check against higher enums of v0x04.
333
334
        Must be called after :meth:`get_v0x01_speed` returns ``None``.
335
        """
336 1
        fts = self.features
337 1
        pfts = PortFeatures04
338 1
        if fts and fts & pfts.OFPPF_1TB_FD:
339 1
            return 10**12 / 8
340 1
        if fts and fts & pfts.OFPPF_100GB_FD:
341 1
            return 100 * 10**9 / 8
342 1
        if fts and fts & pfts.OFPPF_40GB_FD:
343 1
            return 40 * 10**9 / 8
344 1
        return None
345
346 1
    def get_hr_speed(self):
347
        """Return Human-Readable string for link speed.
348
349
        Returns:
350
            string: String with link speed. e.g: '350 Gbps' or '350 Mbps'.
351
352
        """
353 1
        speed = self.speed
354 1
        if speed is None:
355 1
            return ''
356 1
        speed *= 8
357 1
        if speed == 10**12:
358 1
            return '1 Tbps'
359 1
        if speed >= 10**9:
360 1
            return '{} Gbps'.format(round(speed / 10**9))
361 1
        return '{} Mbps'.format(round(speed / 10**6))
362
363 1
    def as_dict(self):
364
        """Return a dictionary with Interface attributes.
365
366
        Speed is in bytes/sec. Example of output (100 Gbps):
367
368
        .. code-block:: python3
369
370
            {'id': '00:00:00:00:00:00:00:01:2',
371
             'name': 'eth01',
372
             'port_number': 2,
373
             'mac': '00:7e:04:3b:c2:a6',
374
             'switch': '00:00:00:00:00:00:00:01',
375
             'type': 'interface',
376
             'nni': False,
377
             'uni': True,
378
             'speed': 12500000000,
379
             'metadata': {},
380
             'lldp': True,
381
             'active': True,
382
             'enabled': False,
383
             'link': ""
384
            }
385
386
        Returns:
387
            dict: Dictionary filled with interface attributes.
388
389
        """
390
        iface_dict = {'id': self.id,
391
                      'name': self.name,
392
                      'port_number': self.port_number,
393
                      'mac': self.address,
394
                      'switch': self.switch.dpid,
395
                      'type': 'interface',
396
                      'nni': self.nni,
397
                      'uni': self.uni,
398
                      'speed': self.speed,
399
                      'metadata': self.metadata,
400
                      'lldp': self.lldp,
401
                      'active': self.is_active(),
402
                      'enabled': self.is_enabled(),
403
                      'link': self.link.id if self.link else ""}
404
        if self.stats:
405
            iface_dict['stats'] = self.stats.as_dict()
406
        return iface_dict
407
408 1
    def as_json(self):
409
        """Return a json with Interfaces attributes.
410
411
        Example of output:
412
413
        .. code-block:: json
414
415
            {"mac": "00:7e:04:3b:c2:a6",
416
             "switch": "00:00:00:00:00:00:00:01",
417
             "type": "interface",
418
             "name": "eth01",
419
             "id": "00:00:00:00:00:00:00:01:2",
420
             "port_number": 2,
421
             "speed": "350 Mbps"}
422
423
        Returns:
424
            string: Json filled with interface attributes.
425
426
        """
427
        return json.dumps(self.as_dict())
428
429
430 1
class UNI:
431
    """Class that represents an User-to-Network Interface."""
432
433 1
    def __init__(self, interface, user_tag=None):
434
        self.user_tag = user_tag
435
        self.interface = interface
436
437 1
    def __eq__(self, other):
438
        """Override the default implementation."""
439
        return (self.user_tag == other.user_tag and
440
                self.interface == other.interface)
441
442 1
    def is_valid(self):
443
        """Check if TAG is possible for this interface TAG pool."""
444
        if self.user_tag:
445
            return self.interface.is_tag_available(self.user_tag)
446
        return True
447
448 1
    def as_dict(self):
449
        """Return a dict representating a UNI object."""
450
        return {'interface_id': self.interface.id,
451
                'tag': self.user_tag.as_dict()}
452
453 1
    def as_json(self):
454
        """Return a json representating a UNI object."""
455
        return json.dumps(self.as_dict())
456
457
458 1
class NNI:
459
    """Class that represents an Network-to-Network Interface."""
460
461 1
    def __init__(self, interface):
462
        self.interface = interface
463
464
465 1
class VNNI(NNI):
466
    """Class that represents an Virtual Network-to-Network Interface."""
467
468 1
    def __init__(self, service_tag, *args, **kwargs):
469
        self.service_tag = service_tag
470
471
        super().__init__(*args, **kwargs)
472