kytos.core.link.Link.from_dict() - Code Metrics - Inspection of "Merge pull request #150 from italovalcy/add_link_r..." - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( f21663...e05eab )

by Italo Valcy

created 2021-12-09 19:56 UTC

kytos.core.link.Link.from_dict() A

↳ Parent: kytos.core.link

Complexity

Conditions

Size

Total Lines	5
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.125

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	2
dl	0
loc	5
rs	10
c	0
b	0
f	0
ccs	1
cts	2
cp	0.5
crap	1.125

"""Module with all classes related to links.

Links are low level abstractions representing connections between two
interfaces.
"""

import hashlib
import json
import random

from kytos.core.common import GenericEntity
from kytos.core.exceptions import (KytosLinkCreationError,
                                   KytosNoTagAvailableError)
from kytos.core.interface import TAGType


class Link(GenericEntity):
    """Define a link between two Endpoints."""

    def __init__(self, endpoint_a, endpoint_b):
        """Create a Link instance and set its attributes.

        Two kytos.core.interface.Interface are required as parameters.
        """
        if endpoint_a is None:
            raise KytosLinkCreationError("endpoint_a cannot be None")
        if endpoint_b is None:
            raise KytosLinkCreationError("endpoint_b cannot be None")
        self.endpoint_a = endpoint_a
        self.endpoint_b = endpoint_b
        super().__init__()

    def __hash__(self):
        return hash(self.id)

    def __repr__(self):
        return f"Link({self.endpoint_a!r}, {self.endpoint_b!r})"

    def is_enabled(self):
        """Override the is_enabled method.

        We consider a link enabled when all the interfaces are enabled.

        Returns:
            boolean: True if both interfaces are enabled, False otherwise.

        """
        return (self._enabled and self.endpoint_a.is_enabled() and
                self.endpoint_b.is_enabled())

    def is_active(self):
        """Override the is_active method.

        We consider a link active whether all the interfaces are active.

        Returns:
            boolean: True if the interfaces are active, othewrise False.

        """
        return (self._active and self.endpoint_a.is_active() and
                self.endpoint_b.is_active())

    def __eq__(self, other):
        """Check if two instances of Link are equal."""
        return self.id == other.id

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

        Returns:
            string: link id.

        """
        dpid_a = self.endpoint_a.switch.dpid
        port_a = self.endpoint_a.port_number
        dpid_b = self.endpoint_b.switch.dpid
        port_b = self.endpoint_b.port_number
        if dpid_a < dpid_b:
            elements = (dpid_a, port_a, dpid_b, port_b)
        elif dpid_a > dpid_b:
            elements = (dpid_b, port_b, dpid_a, port_a)
        elif port_a < port_b:
            elements = (dpid_a, port_a, dpid_b, port_b)
        else:
            elements = (dpid_b, port_b, dpid_a, port_a)

        str_id = "%s:%s:%s:%s" % elements
        return hashlib.sha256(str_id.encode('utf-8')).hexdigest()

    @property
    def available_tags(self):
        """Return the available tags for the link.

        Based on the endpoint tags.
        """
        return [tag for tag in self.endpoint_a.available_tags if tag in
                self.endpoint_b.available_tags]

    def use_tag(self, tag):
        """Remove a specific tag from available_tags if it is there.

        Deprecated: use only the get_next_available_tag method.
        """
        if self.is_tag_available(tag):
            self.endpoint_a.use_tag(tag)
            self.endpoint_b.use_tag(tag)
            return True
        return False

    def is_tag_available(self, tag):
        """Check if a tag is available."""
        return (self.endpoint_a.is_tag_available(tag) and
                self.endpoint_b.is_tag_available(tag))

    def get_next_available_tag(self):
        """Return the next available tag if exists."""
        # Copy the available tags because in case of error
        # we will remove and add elements to the available_tags
        available_tags_a = self.endpoint_a.available_tags.copy()
        available_tags_b = self.endpoint_b.available_tags.copy()
        random.shuffle(available_tags_a)
        random.shuffle(available_tags_b)

        for tag in available_tags_a:
            # Tag does not exist in endpoint B. Try another tag.
            if tag not in available_tags_b:
                continue

            # Tag already in use. Try another tag.
            if not self.endpoint_a.use_tag(tag):
                continue

            # Tag already in use in B. Mark the tag as available again.
            if not self.endpoint_b.use_tag(tag):
                self.endpoint_a.make_tag_available(tag)
                continue

            # Tag used successfully by both endpoints. Returning.
            return tag

        raise KytosNoTagAvailableError(self)

    def make_tag_available(self, tag):
        """Add a specific tag in available_tags."""
        if not self.is_tag_available(tag):
            self.endpoint_a.make_tag_available(tag)
            self.endpoint_b.make_tag_available(tag)
            return True
        return False

    def available_vlans(self):
        """Get all available vlans from each interface in the link."""
        vlans_a = self._get_available_vlans(self.endpoint_a)
        vlans_b = self._get_available_vlans(self.endpoint_b)
        return [vlan for vlan in vlans_a if vlan in vlans_b]

    @staticmethod
    def _get_available_vlans(endpoint):
        """Return all vlans from endpoint."""
        tags = endpoint.available_tags
        return [tag for tag in tags if tag.tag_type == TAGType.VLAN]

    def as_dict(self):
        """Return the Link as a dictionary."""
        return {'id': self.id,
                'endpoint_a': self.endpoint_a.as_dict(),
                'endpoint_b': self.endpoint_b.as_dict(),
                'metadata': self.get_metadata_as_dict(),
                'active': self.is_active(),
                'enabled': self.is_enabled()}

    def as_json(self):
        """Return the Link as a JSON string."""
        return json.dumps(self.as_dict())

    @classmethod
    def from_dict(cls, link_dict):
        """Return a Link instance from python dictionary."""
        return cls(link_dict.get('endpoint_a'),
                   link_dict.get('endpoint_b'))


1		"""Module with all classes related to links.
2
3		Links are low level abstractions representing connections between two
4		interfaces.
5		"""
6
7	2	import hashlib
8	2	import json
9	2	import random
10
11	2	from kytos.core.common import GenericEntity
12	2	from kytos.core.exceptions import (KytosLinkCreationError,
13		KytosNoTagAvailableError)
14	2	from kytos.core.interface import TAGType
15
16
17	2	class Link(GenericEntity):
18		"""Define a link between two Endpoints."""
19
20	2	def __init__(self, endpoint_a, endpoint_b):
21		"""Create a Link instance and set its attributes.
22
23		Two kytos.core.interface.Interface are required as parameters.
24		"""
25	2	if endpoint_a is None:
26	2	raise KytosLinkCreationError("endpoint_a cannot be None")
27	2	if endpoint_b is None:
28	2	raise KytosLinkCreationError("endpoint_b cannot be None")
29	2	self.endpoint_a = endpoint_a
30	2	self.endpoint_b = endpoint_b
31	2	super().__init__()
32
33	2	def __hash__(self):
34		return hash(self.id)
35
36	2	def __repr__(self):
37	2	return f"Link({self.endpoint_a!r}, {self.endpoint_b!r})"
38
39	2	def is_enabled(self):
40		"""Override the is_enabled method.
41
42		We consider a link enabled when all the interfaces are enabled.
43
44		Returns:
45		boolean: True if both interfaces are enabled, False otherwise.
46
47		"""
48	2	return (self._enabled and self.endpoint_a.is_enabled() and
49		self.endpoint_b.is_enabled())
50
51	2	def is_active(self):
52		"""Override the is_active method.
53
54		We consider a link active whether all the interfaces are active.
55
56		Returns:
57		boolean: True if the interfaces are active, othewrise False.
58
59		"""
60	2	return (self._active and self.endpoint_a.is_active() and
61		self.endpoint_b.is_active())
62
63	2	def __eq__(self, other):
64		"""Check if two instances of Link are equal."""
65	2	return self.id == other.id
66
67	2	@property
68		def id(self): # pylint: disable=invalid-name
69		"""Return id from Link intance.
70
71		Returns:
72		string: link id.
73
74		"""
75	2	dpid_a = self.endpoint_a.switch.dpid
76	2	port_a = self.endpoint_a.port_number
77	2	dpid_b = self.endpoint_b.switch.dpid
78	2	port_b = self.endpoint_b.port_number
79	2	if dpid_a < dpid_b:
80	2	elements = (dpid_a, port_a, dpid_b, port_b)
81	2	elif dpid_a > dpid_b:
82	2	elements = (dpid_b, port_b, dpid_a, port_a)
83	2	elif port_a < port_b:
84	2	elements = (dpid_a, port_a, dpid_b, port_b)
85		else:
86	2	elements = (dpid_b, port_b, dpid_a, port_a)
87
88	2	str_id = "%s:%s:%s:%s" % elements
89	2	return hashlib.sha256(str_id.encode('utf-8')).hexdigest()
90
91	2	@property
92		def available_tags(self):
93		"""Return the available tags for the link.
94
95		Based on the endpoint tags.
96		"""
97	2	return [tag for tag in self.endpoint_a.available_tags if tag in
98		self.endpoint_b.available_tags]
99
100	2	def use_tag(self, tag):
101		"""Remove a specific tag from available_tags if it is there.
102
103		Deprecated: use only the get_next_available_tag method.
104		"""
105	2	if self.is_tag_available(tag):
106	2	self.endpoint_a.use_tag(tag)
107	2	self.endpoint_b.use_tag(tag)
108	2	return True
109	2	return False
110
111	2	def is_tag_available(self, tag):
112		"""Check if a tag is available."""
113	2	return (self.endpoint_a.is_tag_available(tag) and
114		self.endpoint_b.is_tag_available(tag))
115
116	2	def get_next_available_tag(self):
117		"""Return the next available tag if exists."""
118		# Copy the available tags because in case of error
119		# we will remove and add elements to the available_tags
120	2	available_tags_a = self.endpoint_a.available_tags.copy()
121	2	available_tags_b = self.endpoint_b.available_tags.copy()
122	2	random.shuffle(available_tags_a)
123	2	random.shuffle(available_tags_b)
124
125	2	for tag in available_tags_a:
126		# Tag does not exist in endpoint B. Try another tag.
127	2	if tag not in available_tags_b:
128		continue
129
130		# Tag already in use. Try another tag.
131	2	if not self.endpoint_a.use_tag(tag):
132		continue
133
134		# Tag already in use in B. Mark the tag as available again.
135	2	if not self.endpoint_b.use_tag(tag):
136		self.endpoint_a.make_tag_available(tag)
137		continue
138
139		# Tag used successfully by both endpoints. Returning.
140	2	return tag
141
142		raise KytosNoTagAvailableError(self)
143
144	2	def make_tag_available(self, tag):
145		"""Add a specific tag in available_tags."""
146	2	if not self.is_tag_available(tag):
147	2	self.endpoint_a.make_tag_available(tag)
148	2	self.endpoint_b.make_tag_available(tag)
149	2	return True
150	2	return False
151
152	2	def available_vlans(self):
153		"""Get all available vlans from each interface in the link."""
154	2	vlans_a = self._get_available_vlans(self.endpoint_a)
155	2	vlans_b = self._get_available_vlans(self.endpoint_b)
156	2	return [vlan for vlan in vlans_a if vlan in vlans_b]
157
158	2	@staticmethod
159		def _get_available_vlans(endpoint):
160		"""Return all vlans from endpoint."""
161	2	tags = endpoint.available_tags
162	2	return [tag for tag in tags if tag.tag_type == TAGType.VLAN]
163
164	2	def as_dict(self):
165		"""Return the Link as a dictionary."""
166		return {'id': self.id,
167		'endpoint_a': self.endpoint_a.as_dict(),
168		'endpoint_b': self.endpoint_b.as_dict(),
169		'metadata': self.get_metadata_as_dict(),
170		'active': self.is_active(),
171		'enabled': self.is_enabled()}
172
173	2	def as_json(self):
174		"""Return the Link as a JSON string."""
175		return json.dumps(self.as_dict())
176
177	2	@classmethod
178		def from_dict(cls, link_dict):
179		"""Return a Link instance from python dictionary."""
180		return cls(link_dict.get('endpoint_a'),
181		link_dict.get('endpoint_b'))
182

kytos-ng / kytos

Push — master ( f21663...e05eab )

kytos.core.link.Link.from_dict() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like