kytos.core.interface.Interface.set_available_tags() - Code Metrics - Inspection of "Merge pull request #836 from viniarck/available_ta..." - kytos/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 912aad...faaf9a )

by Beraldo

created 2018-12-11 14:56 UTC

Interface.set_available_tags() A

↳ Parent: kytos.core.interface

Complexity

Conditions

Size

Total Lines	12
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	2.864

Importance

Changes

Metric	Value
cc	2
eloc	6
nop	2
dl	0
loc	12
ccs	2
cts	5
cp	0.4
crap	2.864
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 = tag_type
        self.value = value

    def __eq__(self, other):
        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())


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

    # pylint: disable=too-many-arguments
    def __init__(self, name, port_number, switch, address=None, state=None,
                 features=None, speed=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.
        """
        self.name = name
        self.port_number = int(port_number)
        self.switch = switch
        self.address = address
        self.state = state
        self.features = features
        self.nni = False
        self.endpoints = []
        self.stats = None
        self.link = None
        self._custom_speed = speed
        self.set_available_tags(range(1, 4096))

        super().__init__()

    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.name == other.name and \
                self.address == other.address and \
                self.switch.dpid == other.switch.dpid
        return False

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

        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."""
        for available_tag in self.available_tags:
            if tag == available_tag:
                self.available_tags.remove(available_tag)
                return True
        return False

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

    def get_next_available_tag(self):
        """Return the next available tag if exists."""
        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``.

        """
        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

        return self.get_of_features_speed()

    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()
        if speed is not None:
            return speed
        # 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 _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': {},
             '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,
                      '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

    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()}

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

kytos / kytos

Push — master ( 912aad...faaf9a )

Interface.set_available_tags() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like