kytos.core.interface.TAG.__repr__() - Code Metrics - Inspection of "Merge pull request #867 from ajoaoff/viniarck-vlan..." - kytos/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 76cc93...764801 )

by Beraldo

created 2019-01-31 16:32 UTC

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

↳ Parent: kytos.core.interface

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	2
ccs	1
cts	1
cp	1
crap	1
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())

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

kytos / kytos

Push — master ( 76cc93...764801 )

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

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

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