kytos.core.interface.Interface.delete_endpoint() - Code Metrics - kytos/kytos - Measure and Improve Code Quality continuously with Scrutinizer

kytos.core.interface.Interface.delete_endpoint() A
last analyzed 2021-05-31 22:05 UTC

↳ Parent: kytos.core.interface

Complexity

Conditions

Size

Total Lines	9
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	2

Importance

Changes

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

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

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

__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._custom_speed = speed
        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 "{}:{}".format(self.switch.dpid, self.port_number)

    @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.
        """
        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:
            self.available_tags.remove(tag)
        except ValueError:
            return False
        return True

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

kytos / kytos

kytos.core.interface.Interface.delete_endpoint() A last analyzed 2021-05-31 22:05 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

kytos.core.interface.Interface.delete_endpoint() A
last analyzed 2021-05-31 22:05 UTC