build.automate.Automate.find_circuits() - Code Metrics - Inspection of "Improve coverage" - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#30)

by Rogerio

created 2022-07-12 02:45 UTC

build.automate.Automate.find_circuits() F

↳ Parent: build.automate

Complexity

Conditions

Size

Total Lines	57
Code Lines	44

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	31
CRAP Score	17.7744

Importance

Changes

Metric	Value
cc	17
eloc	44
nop	1
dl	0
loc	57
rs	1.8
c	0
b	0
f	0
ccs	31
cts	36
cp	0.8611
crap	17.7744

How to fix Long Method Complexity

"""Automate circuit traces."""

import time

import requests
from kytos.core import KytosEvent, log
from napps.amlight.sdntrace_cp import settings
from napps.amlight.sdntrace_cp.utils import clean_circuits, format_result
from pyof.v0x01.common.phy_port import Port as Port10
from pyof.v0x04.common.port import PortNo as Port13


class Automate:
    """Find all circuits and automate trace execution."""

    def __init__(self, tracer):
        self._tracer = tracer
        self._circuits = []
        self.find_circuits()

    # pylint: disable=too-many-nested-blocks, too-many-branches
    def find_circuits(self):
        """Discover all circuits in a topology.

        Using the list of flows per switch, run control plane
        traces to find a list of circuits."""
        all_flows = {}
        circuits = []

        for switch in self._tracer.controller.switches.values():
            all_flows[switch] = []
            if switch.ofp_version == '0x01':
                controller_port = Port10.OFPP_CONTROLLER
            else:
                controller_port = Port13.OFPP_CONTROLLER

            try:
                for flow in switch.generic_flows:
                    action_ok = False
                    in_port_ok = False
                    if 'in_port' in flow.match and flow.match['in_port'] != 0:
                        in_port_ok = True
                    if in_port_ok:
                        for action in flow.actions:
                            if action.action_type == 'output' \
                                    and action.port != controller_port:
                                action_ok = True
                    if action_ok:
                        all_flows[switch].append(flow)
            except AttributeError:
                pass

        for switch, flows in all_flows.items():
            for flow in flows:
                in_port = flow.match['in_port']
                vlan = None
                if 'vlan_vid' in flow.match:
                    vlan = flow.match['vlan_vid']
                if switch.ofp_version == '0x04':
                    in_port = in_port.value
                    if vlan:
                        vlan = vlan.value
                entries = {
                    'trace': {
                        'switch': {
                            'dpid': switch.dpid,
                            'in_port': in_port
                        },
                        'eth': {
                            'dl_vlan': vlan
                        }
                    }
                }
                result = self._tracer.tracepath(entries)
                circuits.append({'circuit': format_result(result),
                                 'entries': entries})

        self._circuits = clean_circuits(circuits, self._tracer.controller)

    def run_traces(self):
        """Run traces for all circuits."""

        results = []
        for circuit in self._circuits:
            entries = circuit['entries']
            result = self._tracer.tracepath(entries)
            try:
                result = format_result(result)
                if result != circuit['circuit']:
                    results.append(circuit)
            except KeyError:
                results.append(circuit)
        log.debug('Results %s, size %s', results, len(self._circuits))
        return results

    def get_circuit(self, circuit):
        """Find the given circuit in the list of circuits."""
        for steps in self._circuits:
            endpoint_a = steps['circuit'][0]
            endpoint_z = steps['circuit'][-1]
            # pylint: disable=too-many-boolean-expressions
            if (circuit['dpid_a'] == endpoint_a['dpid'] and
                    circuit['port_a'] == endpoint_a['in_port'] and
                    circuit['vlan_a'] == endpoint_a['in_vlan'] and
                    circuit['dpid_z'] == endpoint_z['dpid'] and
                    circuit['port_z'] == endpoint_z['out_port'] and
                    circuit['vlan_z'] == endpoint_z['out_vlan']):

                return steps['circuit']
        return None

    def _check_trace(self, circuit, trace):
        steps = self.get_circuit(circuit)
        if steps:
            if len(steps) != len(trace) - 1:
                return False
            for i, step in enumerate(steps):
                if not self.check_step(step, trace[i]):
                    return False
        else:
            return False
        return True

    def run_important_traces(self):
        """Run SDNTrace in data plane for important circuits as defined
            by user."""
        event = KytosEvent(name='amlight/kytos_courier.slack_send')
        content = {
            'channel': settings.SLACK_CHANNEL,
            'source': 'amlight/sdntrace_cp'
        }

        try:
            important_circuits = settings.IMPORTANT_CIRCUITS
        except AttributeError:
            return

        for circuit in important_circuits:
            entries = {
                'trace': {
                    'switch': {
                        'dpid': circuit['dpid_a'],
                        'in_port': circuit['port_a']
                    },
                    'eth': {
                        'dl_vlan': circuit['vlan_a']
                    }
                }
            }
            result = requests.put(settings.SDNTRACE_URL, json=entries)
            trace = result.json()
            trace_id = trace['result']['trace_id']
            step_type = None
            while step_type != 'last':
                time.sleep(5)
                result = requests.get(f'{settings.SDNTRACE_URL}/{trace_id}')
                trace = result.json()
                step_type = trace['result'][-1]['type']
            check = self._check_trace(circuit, trace['result'])
            if check is False:
                content['m_body'] = 'Trace in data plane different from ' \
                                    'trace in control plane for circuit ' \
                                    f'{circuit}'
                event.content['message'] = content
                self._tracer.controller.buffers.app.put(event)

    @staticmethod
    def check_step(circuit_step, trace_step):
        """Check if a step in SDNTrace in data plane is what it should"""
        return (circuit_step['dpid'] == trace_step['dpid'] and
                circuit_step['in_port'] == trace_step['port'])


1		"""Automate circuit traces."""
2
3	1	import time
4
5	1	import requests
6	1	from kytos.core import KytosEvent, log
7	1	from napps.amlight.sdntrace_cp import settings
8	1	from napps.amlight.sdntrace_cp.utils import clean_circuits, format_result
9	1	from pyof.v0x01.common.phy_port import Port as Port10
10	1	from pyof.v0x04.common.port import PortNo as Port13
11
12
13	1	class Automate:
14		"""Find all circuits and automate trace execution."""
15
16	1	def __init__(self, tracer):
17	1	self._tracer = tracer
18	1	self._circuits = []
19	1	self.find_circuits()
20
21		# pylint: disable=too-many-nested-blocks, too-many-branches
22	1	def find_circuits(self):
23		"""Discover all circuits in a topology.
24
25		Using the list of flows per switch, run control plane
26		traces to find a list of circuits."""
27	1	all_flows = {}
28	1	circuits = []
29
30	1	for switch in self._tracer.controller.switches.values():
31	1	all_flows[switch] = []
32	1	if switch.ofp_version == '0x01':
33		controller_port = Port10.OFPP_CONTROLLER
34		else:
35	1	controller_port = Port13.OFPP_CONTROLLER
36
37	1	try:
38	1	for flow in switch.generic_flows:
39	1	action_ok = False
40	1	in_port_ok = False
41	1	if 'in_port' in flow.match and flow.match['in_port'] != 0:
42	1	in_port_ok = True
43	1	if in_port_ok:
44	1	for action in flow.actions:
45	1	if action.action_type == 'output' \
46		and action.port != controller_port:
47	1	action_ok = True
48	1	if action_ok:
49	1	all_flows[switch].append(flow)
50		except AttributeError:
51		pass
52
53	1	for switch, flows in all_flows.items():
54	1	for flow in flows:
55	1	in_port = flow.match['in_port']
56	1	vlan = None
57	1	if 'vlan_vid' in flow.match:
58		vlan = flow.match['vlan_vid']
59	1	if switch.ofp_version == '0x04':
60	1	in_port = in_port.value
61	1	if vlan:
62		vlan = vlan.value
63	1	entries = {
64		'trace': {
65		'switch': {
66		'dpid': switch.dpid,
67		'in_port': in_port
68		},
69		'eth': {
70		'dl_vlan': vlan
71		}
72		}
73		}
74	1	result = self._tracer.tracepath(entries)
75	1	circuits.append({'circuit': format_result(result),
76		'entries': entries})
77
78	1	self._circuits = clean_circuits(circuits, self._tracer.controller)
79
80	1	def run_traces(self):
81		"""Run traces for all circuits."""
82
83	1	results = []
84	1	for circuit in self._circuits:
85	1	entries = circuit['entries']
86	1	result = self._tracer.tracepath(entries)
87	1	try:
88	1	result = format_result(result)
89	1	if result != circuit['circuit']:
90	1	results.append(circuit)
91		except KeyError:
92		results.append(circuit)
93	1	log.debug('Results %s, size %s', results, len(self._circuits))
94	1	return results
95
96	1	def get_circuit(self, circuit):
97		"""Find the given circuit in the list of circuits."""
98	1	for steps in self._circuits:
99	1	endpoint_a = steps['circuit'][0]
100	1	endpoint_z = steps['circuit'][-1]
101		# pylint: disable=too-many-boolean-expressions
102	1	if (circuit['dpid_a'] == endpoint_a['dpid'] and
103		circuit['port_a'] == endpoint_a['in_port'] and
104		circuit['vlan_a'] == endpoint_a['in_vlan'] and
105		circuit['dpid_z'] == endpoint_z['dpid'] and
106		circuit['port_z'] == endpoint_z['out_port'] and
107		circuit['vlan_z'] == endpoint_z['out_vlan']):
108
109	1	return steps['circuit']
110	1	return None
111
112	1	def _check_trace(self, circuit, trace):
113	1	steps = self.get_circuit(circuit)
114	1	if steps:
115	1	if len(steps) != len(trace) - 1:
116	1	return False
117	1	for i, step in enumerate(steps):
118	1	if not self.check_step(step, trace[i]):
119	1	return False
120		else:
121	1	return False
122	1	return True
123
124	1	def run_important_traces(self):
125		"""Run SDNTrace in data plane for important circuits as defined
126		by user."""
127	1	event = KytosEvent(name='amlight/kytos_courier.slack_send')
128	1	content = {
129		'channel': settings.SLACK_CHANNEL,
130		'source': 'amlight/sdntrace_cp'
131		}
132
133	1	try:
134	1	important_circuits = settings.IMPORTANT_CIRCUITS
135		except AttributeError:
136		return
137
138	1	for circuit in important_circuits:
139	1	entries = {
140		'trace': {
141		'switch': {
142		'dpid': circuit['dpid_a'],
143		'in_port': circuit['port_a']
144		},
145		'eth': {
146		'dl_vlan': circuit['vlan_a']
147		}
148		}
149		}
150	1	result = requests.put(settings.SDNTRACE_URL, json=entries)
151	1	trace = result.json()
152	1	trace_id = trace['result']['trace_id']
153	1	step_type = None
154	1	while step_type != 'last':
155	1	time.sleep(5)
156	1	result = requests.get(f'{settings.SDNTRACE_URL}/{trace_id}')
157	1	trace = result.json()
158	1	step_type = trace['result'][-1]['type']
159	1	check = self._check_trace(circuit, trace['result'])
160	1	if check is False:
161	1	content['m_body'] = 'Trace in data plane different from ' \
162		'trace in control plane for circuit ' \
163		f'{circuit}'
164	1	event.content['message'] = content
165	1	self._tracer.controller.buffers.app.put(event)
166
167	1	@staticmethod
168	1	def check_step(circuit_step, trace_step):
169		"""Check if a step in SDNTrace in data plane is what it should"""
170	1	return (circuit_step['dpid'] == trace_step['dpid'] and
171		circuit_step['in_port'] == trace_step['port'])
172

kytos-ng / sdntrace_cp

Pull Request — master (#30)

build.automate.Automate.find_circuits() F

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like