kytos.core.interface.Interface.__repr__() - Code Metrics - Inspection of "[Fix] available tags safe guard" - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#272)

by Vinicius

created 2022-08-19 20:42 UTC

kytos.core.interface.Interface.repr() A

↳ Parent: kytos.core.interface

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	2
rs	10
c	0
b	0
f	0
ccs	2
cts	2
cp	1
crap	1

"""Module with main classes related to Interfaces."""
import json
import logging
from enum import IntEnum
from threading import Lock

from pyof.v0x01.common.phy_port import Port as PortNo01
from pyof.v0x01.common.phy_port import PortFeatures as PortFeatures01
from pyof.v0x04.common.port import PortFeatures as PortFeatures04
from pyof.v0x04.common.port import PortNo as PortNo04

from kytos.core.common import GenericEntity
from kytos.core.helpers import now
from kytos.core.id import InterfaceID

__all__ = ('Interface',)

LOG = logging.getLogger(__name__)


class TAGType(IntEnum):
    """Class that represents a TAG Type."""

    VLAN = 1
    VLAN_QINQ = 2
    MPLS = 3


class TAG:
    """Class that represents a TAG."""

    def __init__(self, tag_type, value):
        self.tag_type = TAGType(tag_type)
        self.value = value

    def __eq__(self, other):
        if not other:
            return False
        return self.tag_type == other.tag_type and self.value == other.value

    def as_dict(self):
        """Return a dictionary representating a tag object."""
        return {'tag_type': self.tag_type, 'value': self.value}

    @classmethod
    def from_dict(cls, tag_dict):
        """Return a TAG instance from python dictionary."""
        return cls(tag_dict.get('tag_type'), tag_dict.get('value'))

    @classmethod
    def from_json(cls, tag_json):
        """Return a TAG instance from json."""
        return cls.from_dict(json.loads(tag_json))

    def as_json(self):
        """Return a json representating a tag object."""
        return json.dumps(self.as_dict())

    def __repr__(self):
        return f"TAG({self.tag_type!r}, {self.value!r})"


class Interface(GenericEntity):  # pylint: disable=too-many-instance-attributes
    """Interface Class used to abstract the network interfaces."""

    # pylint: disable=too-many-arguments, too-many-public-methods
    def __init__(self, name, port_number, switch, address=None, state=None,
                 features=None, speed=None, config=None):
        """Assign the parameters to instance attributes.

        Args:
            name (string): name from this interface.
            port_number (int): port number from this interface.
            switch (:class:`~.core.switch.Switch`): Switch with this interface.
            address (|hw_address|): Port address from this interface.
            state (|port_stats|): Port Stat from interface. It will be
            deprecated.
            features (|port_features|): Port feature used to calculate link
                utilization from this interface. It will be deprecated.
            speed (int, float): Interface speed in bytes per second. Defaults
                to what is informed by the switch. Return ``None`` if not set
                and switch does not inform the speed.
            config(|port_config|): Port config used to indicate interface
                behavior. In general, the port config bits are set by the
                controller and are not changed by the switch. Options
                are: administratively down, ignore received packets, drop
                forwarded packets, and/or do not send packet-in messages.
        """
        self.name = name
        self.port_number = int(port_number)
        self.switch = switch
        self.address = address
        self.state = state
        self.features = features
        self.config = config
        self.nni = False
        self.endpoints = []
        self.stats = None
        self.link = None
        self.lldp = True
        self._id = InterfaceID(switch.id, port_number)
        self._custom_speed = speed
        self._tag_lock = Lock()
        self.set_available_tags(range(1, 4096))

        super().__init__()

    def __repr__(self):
        return f"Interface('{self.name}', {self.port_number}, {self.switch!r})"

    def __eq__(self, other):
        """Compare Interface class with another instance."""
        if isinstance(other, str):
            return self.address == other
        if isinstance(other, Interface):
            return self.port_number == other.port_number and \
                self.switch.dpid == other.switch.dpid
        return False

    @property
    def id(self):  # pylint: disable=invalid-name
        """Return id from Interface instance.

        Returns:
            string: Interface id.

        """
        return self._id

    @property
    def uni(self):
        """Return if an interface is a user-to-network Interface."""
        return not self.nni

    def set_available_tags(self, iterable):
        """Set a range of VLAN tags to be used by this Interface.

        Args:
            iterable ([int]): range of VLANs.
        """
        with self._tag_lock:
            self.available_tags = []

            for i in iterable:
                vlan = TAGType.VLAN
                tag = TAG(vlan, i)
                self.available_tags.append(tag)

    def enable(self):
        """Enable this interface instance.

        Also enable the switch instance this interface is attached to.
        """
        self.switch.enable()
        self._enabled = True

    def use_tag(self, tag):
        """Remove a specific tag from available_tags if it is there.

        Return False in case the tag is already removed.
        """
        try:
            with self._tag_lock:
                self.available_tags.remove(tag)
        except ValueError:
            return False
        return True

    def is_tag_available(self, tag):
        """Check if a tag is available."""
        with self._tag_lock:
            return tag in self.available_tags

    def get_next_available_tag(self):
        """Get the next available tag from the interface.

        Return the next available tag if exists and remove from the
        available tags.
        If no tag is available return False.
        """
        try:
            with self._tag_lock:
                return self.available_tags.pop()
        except IndexError:
            return False

    def make_tag_available(self, tag):
        """Add a specific tag in available_tags."""
        if not self.is_tag_available(tag) and isinstance(tag, TAG):
            with self._tag_lock:
                self.available_tags.append(tag)
            return True
        return False

    def get_endpoint(self, endpoint):
        """Return a tuple with existent endpoint, None otherwise.

        Args:
            endpoint(|hw_address|, :class:`.Interface`): endpoint instance.

        Returns:
            tuple: A tuple with endpoint and time of last update.

        """
        for item in self.endpoints:
            if endpoint == item[0]:
                return item
        return None

    def add_endpoint(self, endpoint):
        """Create a new endpoint to Interface instance.

        Args:
            endpoint(|hw_address|, :class:`.Interface`): A target endpoint.
        """
        exists = self.get_endpoint(endpoint)
        if not exists:
            self.endpoints.append((endpoint, now()))

    def delete_endpoint(self, endpoint):
        """Delete a existent endpoint in Interface instance.

        Args:
            endpoint (|hw_address|, :class:`.Interface`): A target endpoint.
        """
        exists = self.get_endpoint(endpoint)
        if exists:
            self.endpoints.remove(exists)

    def update_endpoint(self, endpoint):
        """Update or create new endpoint to Interface instance.

        Args:
            endpoint(|hw_address|, :class:`.Interface`): A target endpoint.
        """
        exists = self.get_endpoint(endpoint)
        if exists:
            self.delete_endpoint(endpoint)
        self.add_endpoint(endpoint)

    def update_link(self, link):
        """Update link for this interface in a consistent way.

        Verify of the other endpoint of the link has the same Link information
        attached to it, and change it if necessary.

        Warning: This method can potentially change information of other
        Interface instances. Use it with caution.
        """
        if self not in (link.endpoint_a, link.endpoint_b):
            return False

        if self.link is None or self.link != link:
            self.link = link

        if link.endpoint_a == self:
            endpoint = link.endpoint_b
        else:
            endpoint = link.endpoint_a

        if endpoint.link is None or endpoint.link != link:
            endpoint.link = link

        return True

    @property
    def speed(self):
        """Return the link speed in bytes per second, None otherwise.

        If the switch was disconnected, we have :attr:`features` and speed is
        still returned for common values between v0x01 and v0x04. For specific
        v0x04 values (40 Gbps, 100 Gbps and 1 Tbps), the connection must be
        active so we can make sure the protocol version is v0x04.

        Returns:
            int, None: Link speed in bytes per second or ``None``.

        """
        speed = self.get_of_features_speed()

        if speed is not None:
            return speed

        if self._custom_speed is not None:
            return self._custom_speed

        if self._is_v0x04() and self.port_number == PortNo04.OFPP_LOCAL:
            return 0

        if not self._is_v0x04() and self.port_number == PortNo01.OFPP_LOCAL:
            return 0

        # Warn unknown speed
        # Use shorter switch ID with its beginning and end
        if isinstance(self.switch.id, str) and len(self.switch.id) > 20:
            switch_id = self.switch.id[:3] + '...' + self.switch.id[-3:]
        else:
            switch_id = self.switch.id
        LOG.warning("Couldn't get port %s speed, sw %s, feats %s",
                    self.port_number, switch_id, self.features)

        return None

    def set_custom_speed(self, bytes_per_second):
        """Set a speed that overrides switch OpenFlow information.

        If ``None`` is given, :attr:`speed` becomes the one given by the
        switch.
        """
        self._custom_speed = bytes_per_second

    def get_custom_speed(self):
        """Return custom speed or ``None`` if not set."""
        return self._custom_speed

    def get_of_features_speed(self):
        """Return the link speed in bytes per second, None otherwise.

        If the switch was disconnected, we have :attr:`features` and speed is
        still returned for common values between v0x01 and v0x04. For specific
        v0x04 values (40 Gbps, 100 Gbps and 1 Tbps), the connection must be
        active so we can make sure the protocol version is v0x04.

        Returns:
            int, None: Link speed in bytes per second or ``None``.

        """
        speed = self._get_v0x01_v0x04_speed()
        # Don't use switch.is_connected() because we can have the protocol
        if speed is None and self._is_v0x04():
            speed = self._get_v0x04_speed()
        return speed

    def _is_v0x04(self):
        """Whether the switch is connected using OpenFlow 1.3."""
        return self.switch.is_connected() and \
            self.switch.connection.protocol.version == 0x04

    def _get_v0x01_v0x04_speed(self):
        """Check against all values of v0x01. They're part of v0x04."""
        fts = self.features
        pfts = PortFeatures01
        if fts and fts & pfts.OFPPF_10GB_FD:
            return 10 * 10**9 / 8
        if fts and fts & (pfts.OFPPF_1GB_HD | pfts.OFPPF_1GB_FD):
            return 10**9 / 8
        if fts and fts & (pfts.OFPPF_100MB_HD | pfts.OFPPF_100MB_FD):
            return 100 * 10**6 / 8
        if fts and fts & (pfts.OFPPF_10MB_HD | pfts.OFPPF_10MB_FD):
            return 10 * 10**6 / 8
        return None

    def _get_v0x04_speed(self):
        """Check against higher enums of v0x04.

        Must be called after :meth:`get_v0x01_speed` returns ``None``.
        """
        fts = self.features
        pfts = PortFeatures04
        if fts and fts & pfts.OFPPF_1TB_FD:
            return 10**12 / 8
        if fts and fts & pfts.OFPPF_100GB_FD:
            return 100 * 10**9 / 8
        if fts and fts & pfts.OFPPF_40GB_FD:
            return 40 * 10**9 / 8
        return None

    def get_hr_speed(self):
        """Return Human-Readable string for link speed.

        Returns:
            string: String with link speed. e.g: '350 Gbps' or '350 Mbps'.

        """
        speed = self.speed
        if speed is None:
            return ''
        speed *= 8
        if speed == 10**12:
            return '1 Tbps'
        if speed >= 10**9:
            return '{} Gbps'.format(round(speed / 10**9))
        return '{} Mbps'.format(round(speed / 10**6))

    def as_dict(self):
        """Return a dictionary with Interface attributes.

        Speed is in bytes/sec. Example of output (100 Gbps):

        .. code-block:: python3

            {'id': '00:00:00:00:00:00:00:01:2',
             'name': 'eth01',
             'port_number': 2,
             'mac': '00:7e:04:3b:c2:a6',
             'switch': '00:00:00:00:00:00:00:01',
             'type': 'interface',
             'nni': False,
             'uni': True,
             'speed': 12500000000,
             'metadata': {},
             'lldp': True,
             'active': True,
             'enabled': False,
             'link': ""
            }

        Returns:
            dict: Dictionary filled with interface attributes.

        """
        iface_dict = {'id': self.id,
                      'name': self.name,
                      'port_number': self.port_number,
                      'mac': self.address,
                      'switch': self.switch.dpid,
                      'type': 'interface',
                      'nni': self.nni,
                      'uni': self.uni,
                      'speed': self.speed,
                      'metadata': self.metadata,
                      'lldp': self.lldp,
                      'active': self.is_active(),
                      'enabled': self.is_enabled(),
                      'link': self.link.id if self.link else ""}
        if self.stats:
            iface_dict['stats'] = self.stats.as_dict()
        return iface_dict

    @classmethod
    def from_dict(cls, interface_dict):
        """Return a Interface instance from python dictionary."""
        return cls(interface_dict.get('name'),
                   interface_dict.get('port_number'),
                   interface_dict.get('switch'),
                   interface_dict.get('address'),
                   interface_dict.get('state'),
                   interface_dict.get('features'),
                   interface_dict.get('speed'))

    def as_json(self):
        """Return a json with Interfaces attributes.

        Example of output:

        .. code-block:: json

            {"mac": "00:7e:04:3b:c2:a6",
             "switch": "00:00:00:00:00:00:00:01",
             "type": "interface",
             "name": "eth01",
             "id": "00:00:00:00:00:00:00:01:2",
             "port_number": 2,
             "speed": "350 Mbps"}

        Returns:
            string: Json filled with interface attributes.

        """
        return json.dumps(self.as_dict())


class UNI:
    """Class that represents an User-to-Network Interface."""

    def __init__(self, interface, user_tag=None):
        self.user_tag = user_tag
        self.interface = interface

    def __eq__(self, other):
        """Override the default implementation."""
        return (self.user_tag == other.user_tag and
                self.interface == other.interface)

    def is_valid(self):
        """Check if TAG is possible for this interface TAG pool."""
        if self.user_tag:
            return self.interface.is_tag_available(self.user_tag)
        return True

    def as_dict(self):
        """Return a dict representating a UNI object."""
        return {
            'interface_id': self.interface.id,
            'tag': self.user_tag.as_dict() if self.user_tag else None
            }

    @classmethod
    def from_dict(cls, uni):
        """Return a Uni instance from python dictionary."""
        return cls(uni.get('interface'),
                   uni.get('user_tag'))

    def as_json(self):
        """Return a json representating a UNI object."""
        return json.dumps(self.as_dict())


class NNI:
    """Class that represents an Network-to-Network Interface."""

    def __init__(self, interface):
        self.interface = interface


class VNNI(NNI):
    """Class that represents an Virtual Network-to-Network Interface."""

    def __init__(self, service_tag, *args, **kwargs):
        self.service_tag = service_tag

        super().__init__(*args, **kwargs)


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

kytos-ng / kytos

Pull Request — master (#272)

kytos.core.interface.Interface.__repr__() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

kytos.core.interface.Interface.repr() A