kytos.core.interface.Interface.status_reason() - Code Metrics - Inspection of "Merge pull request #384 from kytos-ng/feature/stat..." - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( f84184...cafc34 )

by Vinicius

created 2023-05-30 19:32 UTC

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

↳ Parent: kytos.core.interface

Complexity

Conditions

Size

Total Lines	10
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	8
nop	1
dl	0
loc	10
rs	10
c	0
b	0
f	0

"""Module with main classes related to Interfaces."""
import json
import logging
import operator
from collections import OrderedDict
from enum import IntEnum
from functools import reduce
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 EntityStatus, 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."""

    status_funcs = OrderedDict()
    status_reason_funcs = OrderedDict()

    # 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

    @property

    def status(self):
        """Return the current status of the Entity."""
        state = super().status
        if state == EntityStatus.DISABLED:
            return state

        for status_func in self.status_funcs.values():
            if status_func(self) == EntityStatus.DOWN:
                return EntityStatus.DOWN
        return state

    @classmethod
    def register_status_func(cls, name: str, func):
        """Register status func given its name and a callable at setup time."""
        cls.status_funcs[name] = func

    @classmethod
    def register_status_reason_func(cls, name: str, func):
        """Register status reason func given its name
        and a callable at setup time."""
        cls.status_reason_funcs[name] = func

    @property
    def status_reason(self):
        """Return the reason behind the current status of the entity."""
        return reduce(
            operator.or_,
            map(
                lambda x: x(self),
                self.status_reason_funcs.values()
            ),
            super().status_reason
        )

    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 f"{round(speed / 10**9)} Gbps"
        return f"{round(speed / 10**6)} Mbps"

    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,
             'status': 'DISABLED',
             '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(),
            'status': self.status.value,
            'status_reason': sorted(self.status_reason),
            '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_reserved_valid_tag(self) -> bool:
        """Check if TAG string is possible"""
        reserved_tag = {"any", "untagged"}
        if self.user_tag.value in reserved_tag:
            return True
        return False

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

kytos-ng / kytos

Push — master ( f84184...cafc34 )

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

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like