kytos.core.interface

Complexity

Total Complexity

106

Size/Duplication

Total Lines	543
Duplicated Lines	2.03 %

Test Coverage

Coverage

94.72%

Importance

Changes

0

45 Methods

Rating	Name	Duplication	Size	Complexity
B	Interface._get_v0x04_speed()	0	14	7
A	Interface.add_endpoint()	0	9	2
A	Interface.id()	0	9	1
A	Interface.__eq__()	0	8	3
A	Interface.set_custom_speed()	0	7	1
A	Interface.status()	11	11	4
A	Interface.delete_endpoint()	0	9	2
C	Interface._get_v0x01_v0x04_speed()	0	13	9
A	Interface.make_tag_available()	0	7	4
A	Interface.get_endpoint()	0	14	3
A	TAG.from_dict()	0	4	1
A	TAG.__repr__()	0	2	1
A	Interface._is_v0x04()	0	4	1
A	Interface.__init__()	0	40	1
A	UNI.__eq__()	0	4	1
A	Interface.register_status_func()	0	4	1
A	Interface.is_tag_available()	0	4	2
A	Interface.set_available_tags()	0	13	3
B	Interface.update_link()	0	24	7
A	Interface.uni()	0	4	1
A	Interface.__repr__()	0	2	1
A	Interface.as_dict()	0	44	3
A	Interface.as_json()	0	20	1
A	TAG.__init__()	0	3	1
A	Interface.get_next_available_tag()	0	12	3
A	TAG.from_json()	0	4	1
A	TAG.__eq__()	0	4	2
A	Interface.enable()	0	7	1
C	Interface.speed()	0	37	9
A	Interface.get_hr_speed()	0	16	4
A	Interface.get_custom_speed()	0	3	1
A	TAG.as_json()	0	3	1
A	Interface.get_of_features_speed()	0	17	3
A	Interface.use_tag()	0	11	3
A	UNI.__init__()	0	3	1
A	Interface.update_endpoint()	0	10	2
A	Interface.from_dict()	0	10	1
A	TAG.as_dict()	0	3	1
A	UNI._is_reserved_valid_tag()	0	6	2
A	UNI.from_dict()	0	5	1
A	NNI.__init__()	0	2	1
A	VNNI.__init__()	0	4	1
A	UNI.is_valid()	0	8	4
A	UNI.as_json()	0	3	1
A	UNI.as_dict()	0	5	2

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

This code seems to be duplicated in your project.

    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)