kytos.core.interface.VNNI.__init__() - Code Metrics - Inspection of "Changed is_valid() to allow `any` and `untagged`" - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#331)

unknown

created 2023-03-07 01:23 UTC

kytos.core.interface.VNNI.init() A

↳ Parent: kytos.core.interface

Complexity

Conditions

Size

Total Lines	4
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.2963

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	4
dl	0
loc	4
rs	10
c	0
b	0
f	0
ccs	1
cts	3
cp	0.3333
crap	1.2963

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

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

from kytos.core.common import 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()

    # 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

    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,
             'link': ""
            }

        Returns:
            dict: Dictionary filled with interface attributes.

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

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

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

        Example of output:

        .. code-block:: json

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

        Returns:
            string: Json filled with interface attributes.

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


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

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

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

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

kytos-ng / kytos

Pull Request — master (#331)

kytos.core.interface.VNNI.__init__() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

kytos.core.interface.VNNI.init() A