build.tracing.trace_entries.TraceEntries.load_entries() - Code Metrics - Inspection of "Merge pull request #52 from kytos-ng/feat/untagged..." - kytos-ng/sdntrace - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( e385f0...0b1ef2 )

by Vinicius

created 2023-06-12 19:06 UTC

TraceEntries.load_entries() F

↳ Parent: build.tracing.trace_entries

Complexity

Conditions

Size

Total Lines	76
Code Lines	45

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	28
CRAP Score	41.5697

Importance

Changes

Metric	Value
cc	20
eloc	45
nop	2
dl	0
loc	76
ccs	28
cts	45
cp	0.6222
crap	41.5697
rs	0
c	0
b	0
f	0

How to fix Long Method Complexity

"""
    Class Entries. Used to evaluate entries provided.
"""
import re


DPID_ADDR = re.compile('([0-9A-Fa-f]{2}[-:]){7}[0-9A-Fa-f]{2}$')
MAC_ADDR = re.compile('([0-9A-Fa-f]{2}[-:]){5}[0-9A-Fa-f]{2}$')
IP_ADDR = re.compile("^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\\.){3}"
                     "(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$")


class TraceEntries(object):
    """ Class Entries. Used to evaluate entries provided. """

    def __init__(self):
        self._dpid = 0
        self._in_port = 0
        self._dl_src = 0
        self._dl_dst = 'ca:fe:ca:fe:ca:fe'
        self._dl_vlan = 0
        self._dl_type = 0x800
        self._dl_vlan_pcp = 0
        self._nw_src = '1.1.1.1'
        self._nw_dst = '1.1.1.2'
        self._nw_tos = 0
        self._nw_proto = 0
        self._tp_src = 0
        self._tp_dst = 0
        self.init_entries = dict()  # User request

    @property
    def dpid(self):
        """ Property dpid """
        return self._dpid

    @dpid.setter
    def dpid(self, dpid):
        """ Validates DPID formats: '1', 'a',  '0000000000000001',
        'abcdefabcdefabcd', 'ab:cd:ef:ab:cd:ef:ab:cd'

        Args:
            dpid: entries['trace']['switch']['dpid']
        """
        if not isinstance(dpid, str):
            raise ValueError("Error: dpid has to be string")

        if 0 < len(dpid) <= 16:  # pylint: disable=len-as-condition
            # DPIDs: '1', 'a', '0000000000000001', 'abcdefabcdefabcd'
            if dpid.isalnum():
                if not re.search('[g-z]', dpid.lower()):
                    error_len = False
                else:
                    error_len = True
            else:
                error_len = True

        elif len(dpid) == 23:
            # DPIDs: 'ab:cd:ef:ab:cd:ef:ab:cd'
            # Valid chars: 0-9, :, a-f, A-Z
            if re.search(DPID_ADDR, dpid):
                error_len = False
            else:
                error_len = True
        else:
            error_len = True

        if error_len:
            msg = "Error: dpid allows [a-f], int, and :. Lengths: 1-16 and 23"
            raise ValueError(msg)

        self._dpid = dpid

    @property
    def in_port(self):
        """ IN_PORT Getter """
        return self._in_port

    @in_port.setter
    def in_port(self, in_port):
        """ IN_PORT Setter """
        if not isinstance(in_port, int):
            raise ValueError("Error: in_port has to be integer")
        else:
            if in_port <= 0 or in_port >= 2**64:
                raise ValueError("Error: in_port has to be > 0")

        self._in_port = in_port

    @property
    def dl_src(self):
        """ dl_src Getter """
        return self._dl_src

    @dl_src.setter
    def dl_src(self, dl_src):
        """ dl_src Setter: entries['trace']['eth']['dl_src'].
        Validates MAC formats: 'ab:cd:ef:ab:cd:ef'

        Args:
            dl_src: entries['trace']['eth']['dl_src']
        """

        if not isinstance(dl_src, str):
            raise ValueError("Error: dl_src has to be string")

        elif not re.search(MAC_ADDR, dl_src):
            # DPIDs: 'ab:cd:ef:ab:cd:ef'
            # Valid chars: 0-9, :, a-f, A-Z
            msg = "Error: dl_src allows char [a-f], int, and :. Lengths: 17"
            raise ValueError(msg)

        self._dl_src = dl_src

    @property
    def dl_dst(self):
        """ dl_dst Getter """
        return self._dl_dst

    @dl_dst.setter
    def dl_dst(self, dl_dst):
        """ dl_dst Setter: entries['trace']['eth']['dl_dst'].
        Validates MAC formats: 'ab:cd:ef:ab:cd:ef'

        Args:
            dl_dst: entries['trace']['eth']['dl_dst']
        """

        if not isinstance(dl_dst, str):
            raise ValueError("Error: dl_dst has to be string")

        elif not re.search(MAC_ADDR, dl_dst):
            # DPIDs: 'ab:cd:ef:ab:cd:ef'
            # Valid chars: 0-9, :, a-f, A-Z
            msg = "Error: dl_dst allows char [a-f], int, and :. Lengths: 17"
            raise ValueError(msg)

        self._dl_dst = dl_dst

    @property
    def dl_vlan(self):
        """ dl_vlan Getter """
        return self._dl_vlan

    @dl_vlan.setter
    def dl_vlan(self, dl_vlan):
        """ dl_vlan Setter """
        if not isinstance(dl_vlan, int):
            raise ValueError("Error: dl_vlan has to be integer")
        else:
            if not 0 < dl_vlan <= 4095:
                raise ValueError("Error: dl_vlan has to be between 0 and 4095")

        self._dl_vlan = dl_vlan

    @property
    def dl_type(self):
        """ dl_type Getter """
        return self._dl_type

    @dl_type.setter
    def dl_type(self, dl_type):
        """ dl_type Setter """
        if not isinstance(dl_type, int):
            raise ValueError("Error: dl_type has to be integer")
        else:
            if not 0 < dl_type <= 65535:
                raise ValueError("Error: dl_type has to be [0-65535]")

        self._dl_type = dl_type

    @property
    def dl_vlan_pcp(self):
        """ dl_vlan_pcp Getter """
        return self._dl_vlan_pcp

    @dl_vlan_pcp.setter
    def dl_vlan_pcp(self, dl_vlan_pcp):
        """ dl_vlan_pcp Setter """
        if not isinstance(dl_vlan_pcp, int):
            raise ValueError("Error: dl_vlan_pcp has to be integer")
        else:
            if not 0 < dl_vlan_pcp <= 7:
                raise ValueError("Error: dl_vlan_pcp has to be [0-7]")

        self._dl_vlan_pcp = dl_vlan_pcp

    @property
    def nw_tos(self):
        """ nw_tos Getter """
        return self._nw_tos

    @nw_tos.setter
    def nw_tos(self, nw_tos):
        """ nw_tos Setter """
        if not isinstance(nw_tos, int):
            raise ValueError("Error: nw_tos has to be integer")
        else:
            if not 0 < nw_tos <= 7:
                raise ValueError("Error: nw_tos has to be between 0 and 7")

        self._nw_tos = nw_tos

    @property
    def nw_src(self):
        """ nw_src Getter """
        return self._nw_src

    @nw_src.setter
    def nw_src(self, nw_src):
        """ nw_src Setter """

        if not isinstance(nw_src, str):
            raise ValueError("Error: nw_src has to be string")

        elif not re.search(IP_ADDR, nw_src):
            # Filters: 0.0.0.1 to 255.255.255.255
            msg = "Error: nw_src is not a proper IPv4"
            raise ValueError(msg)

        self._nw_src = nw_src

    @property
    def nw_dst(self):
        """ nw_dst Getter """
        return self._nw_dst

    @nw_dst.setter
    def nw_dst(self, nw_dst):
        """ nw_dst Setter """

        if not isinstance(nw_dst, str):
            raise ValueError("Error: nw_dst has to be string")

        elif not re.search(IP_ADDR, nw_dst):
            # Filters: 0.0.0.1 to 255.255.255.255
            msg = "Error: nw_dst is not a proper IPv4"
            raise ValueError(msg)

        self._nw_dst = nw_dst

    @property
    def nw_proto(self):
        """ nw_proto Getter """
        return self._nw_proto

    @nw_proto.setter
    def nw_proto(self, nw_proto):
        """ nw_proto Setter """
        if not isinstance(nw_proto, int):
            raise ValueError("Error: nw_proto has to be integer")
        else:
            if not 0 < nw_proto <= 65535:
                raise ValueError("Error: nw_proto has to be [0-65535]")

        self._nw_proto = nw_proto

    @property
    def tp_src(self):
        """ tp_src Getter """
        return self._tp_src

    @tp_src.setter
    def tp_src(self, tp_src):
        """ tp_src Setter """
        if not isinstance(tp_src, int):
            raise ValueError("Error: tp_src has to be integer")
        else:
            if not 0 < tp_src <= 65535:
                raise ValueError("Error: tp_src has to be between 0 and 65535")

        self._tp_src = tp_src

    @property
    def tp_dst(self):
        """ tp_dst Getter """
        return self._tp_dst

    @tp_dst.setter
    def tp_dst(self, tp_dst):
        """ tp_dst Setter """
        if not isinstance(tp_dst, int):
            raise ValueError("Error: tp_dst has to be integer")
        else:
            if not 0 < tp_dst <= 65535:
                raise ValueError("Error: tp_dst has to be between 0 and 65535")

        self._tp_dst = tp_dst

    def load_entries(self, entries):
        """ Import entries provided

        Args:
            entries: user-provided entries
        """
        # Basic entries['trace']
        if 'trace' not in entries:
            raise ValueError("Error: Trace key entry missing")
        elif not isinstance(entries['trace'], dict):
            raise ValueError("Error: Trace has to be dict")

        trace = entries['trace']

        # Basic entries['trace']['switch']
        if 'switch' not in trace:
            raise ValueError("Error: switch key not provided")
        elif not isinstance(trace['switch'], dict):
            raise ValueError("Error: switch has to be dict")

        switch = trace['switch']

        if 'dpid' not in switch:
            raise ValueError("Error: dpid not provided")
        else:
            self.dpid = switch['dpid']

        if 'in_port' not in switch:
            raise ValueError("Error: in_port not provided")
        else:
            self.in_port = switch['in_port']

        # Basic entries['trace']['switch']
        eth = trace.get("eth", {})
        if not isinstance(eth, dict):
            raise ValueError("Error: eth has to be dict")

        if 'dl_vlan' in eth:
            self.dl_vlan = eth['dl_vlan']

        if 'dl_src' in eth:
            self.dl_src = eth['dl_src']

        if 'dl_dst' in eth:
            self.dl_dst = eth['dl_dst']

        if 'dl_vlan_pcp' in eth:
            self.dl_vlan_pcp = eth['dl_vlan_pcp']

        if 'dl_type' in eth:
            self.dl_type = eth['dl_type']

        # Basic entries['trace']['ip']
        if 'ip' in trace:
            ip_ = trace['ip']

            if 'nw_src' in ip_:
                self.nw_src = ip_['nw_src']

            if 'nw_dst' in ip_:
                self.nw_dst = ip_['nw_dst']

            if 'nw_tos' in ip_:
                self.nw_tos = ip_['nw_tos']

        # Basic entries['trace']['ip']
        if 'tp' in trace:
            tp_ = trace['tp']

            if 'tp_src' in tp_:
                self.tp_src = tp_['tp_src']

            if 'tp_dst' in tp_:
                self.tp_dst = tp_['tp_dst']

        self.init_entries = entries


1		"""
2		Class Entries. Used to evaluate entries provided.
3		"""
4	1	import re
5
6
7	1	DPID_ADDR = re.compile('([0-9A-Fa-f]{2}[-:]){7}[0-9A-Fa-f]{2}$')
8	1	MAC_ADDR = re.compile('([0-9A-Fa-f]{2}[-:]){5}[0-9A-Fa-f]{2}$')
9	1	IP_ADDR = re.compile("^((25[0-5]\|2[0-4][0-9]\|[01]?[0-9][0-9]?)\\.){3}"
10		"(25[0-5]\|2[0-4][0-9]\|[01]?[0-9][0-9]?)$")
11
12
13	1	class TraceEntries(object):
14		""" Class Entries. Used to evaluate entries provided. """
15
16	1	def __init__(self):
17	1	self._dpid = 0
18	1	self._in_port = 0
19	1	self._dl_src = 0
20	1	self._dl_dst = 'ca:fe:ca:fe:ca:fe'
21	1	self._dl_vlan = 0
22	1	self._dl_type = 0x800
23	1	self._dl_vlan_pcp = 0
24	1	self._nw_src = '1.1.1.1'
25	1	self._nw_dst = '1.1.1.2'
26	1	self._nw_tos = 0
27	1	self._nw_proto = 0
28	1	self._tp_src = 0
29	1	self._tp_dst = 0
30	1	self.init_entries = dict() # User request
31
32	1	@property
33	1	def dpid(self):
34		""" Property dpid """
35	1	return self._dpid
36
37	1	@dpid.setter
38	1	def dpid(self, dpid):
39		""" Validates DPID formats: '1', 'a', '0000000000000001',
40		'abcdefabcdefabcd', 'ab:cd:ef:ab:cd:ef:ab:cd'
41
42		Args:
43		dpid: entries['trace']['switch']['dpid']
44		"""
45	1	if not isinstance(dpid, str):
46	1	raise ValueError("Error: dpid has to be string")
47
48	1	if 0 < len(dpid) <= 16: # pylint: disable=len-as-condition
49		# DPIDs: '1', 'a', '0000000000000001', 'abcdefabcdefabcd'
50	1	if dpid.isalnum():
51	1	if not re.search('[g-z]', dpid.lower()):
52	1	error_len = False
53		else:
54		error_len = True
55		else:
56	1	error_len = True
57
58	1	elif len(dpid) == 23:
59		# DPIDs: 'ab:cd:ef:ab:cd:ef:ab:cd'
60		# Valid chars: 0-9, :, a-f, A-Z
61	1	if re.search(DPID_ADDR, dpid):
62	1	error_len = False
63		else:
64		error_len = True
65		else:
66	1	error_len = True
67
68	1	if error_len:
69	1	msg = "Error: dpid allows [a-f], int, and :. Lengths: 1-16 and 23"
70	1	raise ValueError(msg)
71
72	1	self._dpid = dpid
73
74	1	@property
75	1	def in_port(self):
76		""" IN_PORT Getter """
77	1	return self._in_port
78
79	1	@in_port.setter
80	1	def in_port(self, in_port):
81		""" IN_PORT Setter """
82	1	if not isinstance(in_port, int):
83	1	raise ValueError("Error: in_port has to be integer")
84		else:
85	1	if in_port <= 0 or in_port >= 2**64:
86	1	raise ValueError("Error: in_port has to be > 0")
87
88	1	self._in_port = in_port
89
90	1	@property
91	1	def dl_src(self):
92		""" dl_src Getter """
93	1	return self._dl_src
94
95	1	@dl_src.setter
96	1	def dl_src(self, dl_src):
97		""" dl_src Setter: entries['trace']['eth']['dl_src'].
98		Validates MAC formats: 'ab:cd:ef:ab:cd:ef'
99
100		Args:
101		dl_src: entries['trace']['eth']['dl_src']
102		"""
103
104	1	if not isinstance(dl_src, str):
105	1	raise ValueError("Error: dl_src has to be string")
106
107	1	elif not re.search(MAC_ADDR, dl_src):
108		# DPIDs: 'ab:cd:ef:ab:cd:ef'
109		# Valid chars: 0-9, :, a-f, A-Z
110	1	msg = "Error: dl_src allows char [a-f], int, and :. Lengths: 17"
111	1	raise ValueError(msg)
112
113	1	self._dl_src = dl_src
114
115	1	@property
116	1	def dl_dst(self):
117		""" dl_dst Getter """
118	1	return self._dl_dst
119
120	1	@dl_dst.setter
121	1	def dl_dst(self, dl_dst):
122		""" dl_dst Setter: entries['trace']['eth']['dl_dst'].
123		Validates MAC formats: 'ab:cd:ef:ab:cd:ef'
124
125		Args:
126		dl_dst: entries['trace']['eth']['dl_dst']
127		"""
128
129	1	if not isinstance(dl_dst, str):
130	1	raise ValueError("Error: dl_dst has to be string")
131
132	1	elif not re.search(MAC_ADDR, dl_dst):
133		# DPIDs: 'ab:cd:ef:ab:cd:ef'
134		# Valid chars: 0-9, :, a-f, A-Z
135	1	msg = "Error: dl_dst allows char [a-f], int, and :. Lengths: 17"
136	1	raise ValueError(msg)
137
138	1	self._dl_dst = dl_dst
139
140	1	@property
141	1	def dl_vlan(self):
142		""" dl_vlan Getter """
143	1	return self._dl_vlan
144
145	1	@dl_vlan.setter
146	1	def dl_vlan(self, dl_vlan):
147		""" dl_vlan Setter """
148	1	if not isinstance(dl_vlan, int):
149	1	raise ValueError("Error: dl_vlan has to be integer")
150		else:
151	1	if not 0 < dl_vlan <= 4095:
152	1	raise ValueError("Error: dl_vlan has to be between 0 and 4095")
153
154	1	self._dl_vlan = dl_vlan
155
156	1	@property
157	1	def dl_type(self):
158		""" dl_type Getter """
159	1	return self._dl_type
160
161	1	@dl_type.setter
162	1	def dl_type(self, dl_type):
163		""" dl_type Setter """
164	1	if not isinstance(dl_type, int):
165	1	raise ValueError("Error: dl_type has to be integer")
166		else:
167	1	if not 0 < dl_type <= 65535:
168	1	raise ValueError("Error: dl_type has to be [0-65535]")
169
170	1	self._dl_type = dl_type
171
172	1	@property
173	1	def dl_vlan_pcp(self):
174		""" dl_vlan_pcp Getter """
175	1	return self._dl_vlan_pcp
176
177	1	@dl_vlan_pcp.setter
178	1	def dl_vlan_pcp(self, dl_vlan_pcp):
179		""" dl_vlan_pcp Setter """
180	1	if not isinstance(dl_vlan_pcp, int):
181	1	raise ValueError("Error: dl_vlan_pcp has to be integer")
182		else:
183	1	if not 0 < dl_vlan_pcp <= 7:
184	1	raise ValueError("Error: dl_vlan_pcp has to be [0-7]")
185
186	1	self._dl_vlan_pcp = dl_vlan_pcp
187
188	1	@property
189	1	def nw_tos(self):
190		""" nw_tos Getter """
191	1	return self._nw_tos
192
193	1	@nw_tos.setter
194	1	def nw_tos(self, nw_tos):
195		""" nw_tos Setter """
196	1	if not isinstance(nw_tos, int):
197	1	raise ValueError("Error: nw_tos has to be integer")
198		else:
199	1	if not 0 < nw_tos <= 7:
200	1	raise ValueError("Error: nw_tos has to be between 0 and 7")
201
202	1	self._nw_tos = nw_tos
203
204	1	@property
205	1	def nw_src(self):
206		""" nw_src Getter """
207	1	return self._nw_src
208
209	1	@nw_src.setter
210	1	def nw_src(self, nw_src):
211		""" nw_src Setter """
212
213	1	if not isinstance(nw_src, str):
214	1	raise ValueError("Error: nw_src has to be string")
215
216	1	elif not re.search(IP_ADDR, nw_src):
217		# Filters: 0.0.0.1 to 255.255.255.255
218	1	msg = "Error: nw_src is not a proper IPv4"
219	1	raise ValueError(msg)
220
221	1	self._nw_src = nw_src
222
223	1	@property
224	1	def nw_dst(self):
225		""" nw_dst Getter """
226	1	return self._nw_dst
227
228	1	@nw_dst.setter
229	1	def nw_dst(self, nw_dst):
230		""" nw_dst Setter """
231
232	1	if not isinstance(nw_dst, str):
233	1	raise ValueError("Error: nw_dst has to be string")
234
235	1	elif not re.search(IP_ADDR, nw_dst):
236		# Filters: 0.0.0.1 to 255.255.255.255
237	1	msg = "Error: nw_dst is not a proper IPv4"
238	1	raise ValueError(msg)
239
240	1	self._nw_dst = nw_dst
241
242	1	@property
243	1	def nw_proto(self):
244		""" nw_proto Getter """
245	1	return self._nw_proto
246
247	1	@nw_proto.setter
248	1	def nw_proto(self, nw_proto):
249		""" nw_proto Setter """
250		if not isinstance(nw_proto, int):
251		raise ValueError("Error: nw_proto has to be integer")
252		else:
253		if not 0 < nw_proto <= 65535:
254		raise ValueError("Error: nw_proto has to be [0-65535]")
255
256		self._nw_proto = nw_proto
257
258	1	@property
259	1	def tp_src(self):
260		""" tp_src Getter """
261	1	return self._tp_src
262
263	1	@tp_src.setter
264	1	def tp_src(self, tp_src):
265		""" tp_src Setter """
266	1	if not isinstance(tp_src, int):
267	1	raise ValueError("Error: tp_src has to be integer")
268		else:
269	1	if not 0 < tp_src <= 65535:
270	1	raise ValueError("Error: tp_src has to be between 0 and 65535")
271
272	1	self._tp_src = tp_src
273
274	1	@property
275	1	def tp_dst(self):
276		""" tp_dst Getter """
277	1	return self._tp_dst
278
279	1	@tp_dst.setter
280	1	def tp_dst(self, tp_dst):
281		""" tp_dst Setter """
282	1	if not isinstance(tp_dst, int):
283	1	raise ValueError("Error: tp_dst has to be integer")
284		else:
285	1	if not 0 < tp_dst <= 65535:
286	1	raise ValueError("Error: tp_dst has to be between 0 and 65535")
287
288	1	self._tp_dst = tp_dst
289
290	1	def load_entries(self, entries):
291		""" Import entries provided
292
293		Args:
294		entries: user-provided entries
295		"""
296		# Basic entries['trace']
297	1	if 'trace' not in entries:
298	1	raise ValueError("Error: Trace key entry missing")
299	1	elif not isinstance(entries['trace'], dict):
300		raise ValueError("Error: Trace has to be dict")
301
302	1	trace = entries['trace']
303
304		# Basic entries['trace']['switch']
305	1	if 'switch' not in trace:
306	1	raise ValueError("Error: switch key not provided")
307	1	elif not isinstance(trace['switch'], dict):
308		raise ValueError("Error: switch has to be dict")
309
310	1	switch = trace['switch']
311
312	1	if 'dpid' not in switch:
313	1	raise ValueError("Error: dpid not provided")
314		else:
315	1	self.dpid = switch['dpid']
316
317	1	if 'in_port' not in switch:
318	1	raise ValueError("Error: in_port not provided")
319		else:
320	1	self.in_port = switch['in_port']
321
322		# Basic entries['trace']['switch']
323	1	eth = trace.get("eth", {})
324	1	if not isinstance(eth, dict):
325	1	raise ValueError("Error: eth has to be dict")
326
327	1	if 'dl_vlan' in eth:
328	1	self.dl_vlan = eth['dl_vlan']
329
330	1	if 'dl_src' in eth:
331		self.dl_src = eth['dl_src']
332
333	1	if 'dl_dst' in eth:
334		self.dl_dst = eth['dl_dst']
335
336	1	if 'dl_vlan_pcp' in eth:
337		self.dl_vlan_pcp = eth['dl_vlan_pcp']
338
339	1	if 'dl_type' in eth:
340	1	self.dl_type = eth['dl_type']
341
342		# Basic entries['trace']['ip']
343	1	if 'ip' in trace:
344		ip_ = trace['ip']
345
346		if 'nw_src' in ip_:
347		self.nw_src = ip_['nw_src']
348
349		if 'nw_dst' in ip_:
350		self.nw_dst = ip_['nw_dst']
351
352		if 'nw_tos' in ip_:
353		self.nw_tos = ip_['nw_tos']
354
355		# Basic entries['trace']['ip']
356	1	if 'tp' in trace:
357		tp_ = trace['tp']
358
359		if 'tp_src' in tp_:
360		self.tp_src = tp_['tp_src']
361
362		if 'tp_dst' in tp_:
363		self.tp_dst = tp_['tp_dst']
364
365		self.init_entries = entries
366

kytos-ng / sdntrace

Push — master ( e385f0...0b1ef2 )

TraceEntries.load_entries() F

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like