build.main.Main.create_circuit() - Code Metrics - Inspection of "[WIP] Verify EVC enabled/disabled status when depl..." - kytos/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#116)

by Antonio

created 2019-02-21 19:27 UTC

build.main.Main.create_circuit() B

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	56
Code Lines	19

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	14
CRAP Score	5.1374

Importance

Changes

Metric	Value
cc	5
eloc	19
nop	1
dl	0
loc	56
ccs	14
cts	17
cp	0.8235
crap	5.1374
rs	8.9833
c	0
b	0
f	0

How to fix Long Method

"""Main module of kytos/mef_eline Kytos Network Application.

NApp to provision circuits from user request.
"""
from flask import jsonify, request

from kytos.core import KytosNApp, log, rest
from kytos.core.events import KytosEvent
from kytos.core.helpers import listen_to
from kytos.core.interface import TAG, UNI
from kytos.core.link import Link
from napps.kytos.mef_eline.models import EVC, DynamicPathManager
from napps.kytos.mef_eline.scheduler import CircuitSchedule, Scheduler
from napps.kytos.mef_eline.storehouse import StoreHouse


class Main(KytosNApp):
    """Main class of amlight/mef_eline NApp.

    This class is the entry point for this napp.
    """

    def setup(self):
        """Replace the '__init__' method for the KytosNApp subclass.

        The setup method is automatically called by the controller when your
        application is loaded.

        So, if you have any setup routine, insert it here.
        """
        # object used to scheduler circuit events
        self.sched = Scheduler()

        # object to save and load circuits
        self.storehouse = StoreHouse(self.controller)

        # set the controller that will manager the dynamic paths
        DynamicPathManager.set_controller(self.controller)

    def execute(self):
        """Execute once when the napp is running."""
        pass

    def shutdown(self):
        """Execute when your napp is unloaded.

        If you have some cleanup procedure, insert it here.
        """
        pass

    @rest('/v2/evc/', methods=['GET'])
    def list_circuits(self):
        """Endpoint to return all circuits stored."""
        circuits = self.storehouse.get_data()
        if not circuits:
            return jsonify({}), 200

        return jsonify(circuits), 200

    @rest('/v2/evc/<circuit_id>', methods=['GET'])
    def get_circuit(self, circuit_id):
        """Endpoint to return a circuit based on id."""
        circuits = self.storehouse.get_data()

        if circuit_id in circuits:
            result = circuits[circuit_id]
            status = 200
        else:
            result = {'response': f'circuit_id {circuit_id} not found'}
            status = 400

        return jsonify(result), status

    @rest('/v2/evc/', methods=['POST'])
    def create_circuit(self):
        """Try to create a new circuit.

        Firstly, for EVPL: E-Line NApp verifies if UNI_A's requested C-VID and
        UNI_Z's requested C-VID are available from the interfaces' pools. This
        is checked when creating the UNI object.

        Then, E-Line NApp requests a primary and a backup path to the
        Pathfinder NApp using the attributes primary_links and backup_links
        submitted via REST

        # For each link composing paths in #3:
        #  - E-Line NApp requests a S-VID available from the link VLAN pool.
        #  - Using the S-VID obtained, generate abstract flow entries to be
        #    sent to FlowManager

        Push abstract flow entries to FlowManager and FlowManager pushes
        OpenFlow entries to datapaths

        E-Line NApp generates an event to notify all Kytos NApps of a new EVC
        creation

        Finnaly, notify user of the status of its request.
        """
        # Try to create the circuit object
        data = request.get_json()

        if not data:
            return jsonify("Bad request: The request do not have a json."), 400

        try:
            evc = self.evc_from_dict(data)
        except ValueError as exception:
            return jsonify("Bad request: {}".format(exception)), 400

        # verify duplicated evc
        if self.is_duplicated_evc(evc):
            return jsonify("Not Acceptable: This evc already exists."), 409

        # save circuit
        self.storehouse.save_evc(evc)

        # Schedule the circuit deploy
        self.sched.add(evc)

        # Circuit has no schedule, deploy now
        if not evc.circuit_scheduler:
            evc.deploy()

        # Notify users
        event = KytosEvent(name='kytos.mef_eline.created',
                           content=evc.as_dict())
        self.controller.buffers.app.put(event)

        return jsonify({"circuit_id": evc.id}), 201

    @rest('/v2/evc/<circuit_id>', methods=['PATCH'])
    def update(self, circuit_id):
        """Update a circuit based on payload.

        The EVC required attributes can't be updated.
        """
        data = request.get_json()
        circuits = self.storehouse.get_data()

        if circuit_id not in circuits:
            result = {'response': f'circuit_id {circuit_id} not found'}
            return jsonify(result), 404

        try:
            evc = self.evc_from_dict(circuits.get(circuit_id))
            evc.update(**data)
            self.storehouse.save_evc(evc)
            result = {evc.id: evc.as_dict()}
            status = 200
        except ValueError as exception:
            result = "Bad request: {}".format(exception)
            status = 400

        return jsonify(result), status

    @rest('/v2/evc/<circuit_id>', methods=['DELETE'])
    def delete_circuit(self, circuit_id):
        """Remove a circuit.

        First, flows are removed from the switches, then the EVC is
        disabled.
        """
        circuits = self.storehouse.get_data()
        log.info("Removing %s" % circuit_id)
        evc = self.evc_from_dict(circuits.get(circuit_id))
        evc.remove_current_flows()
        evc.disable()
        self.storehouse.save_evc(evc)

        return jsonify("Circuit removed"), 200

    def is_duplicated_evc(self, evc):
        """Verify if the circuit given is duplicated with the stored evcs.

        Args:
            evc (EVC): circuit to be analysed.

        Returns:
            boolean: True if the circuit is duplicated, otherwise False.

        """
        for circuit_dict in self.storehouse.get_data().values():
            try:
                circuit = self.evc_from_dict(circuit_dict)
            except ValueError:
                continue

            if circuit == evc:
                return True

        return False

    @listen_to('kytos/topology.link_up')
    def handle_link_up(self, event):
        """Change circuit when link is up or end_maintenance."""
        evc = None

        for data in self.storehouse.get_data().values():
            try:
                evc = self.evc_from_dict(data)
            except ValueError as _exception:
                log.debug(f'{data.get("id")} can not be provisioning yet.')
                continue

            evc.handle_link_up(event.content['link'])

    @listen_to('kytos/topology.link_down')
    def handle_link_down(self, event):
        """Change circuit when link is down or under_mantenance."""
        evc = None

        for data in self.storehouse.get_data().values():
            try:
                evc = self.evc_from_dict(data)
            except ValueError as _exception:
                log.debug(f'{data.get("id")} can not be provisioned yet.')
                continue

            if evc.is_affected_by_link(event.content['link']):
                log.info('handling evc %s' % evc)
                evc.handle_link_down()

    def evc_from_dict(self, evc_dict):
        """Convert some dict values to instance of EVC classes.

        This method will convert: [UNI, Link]
        """
        data = evc_dict.copy()  # Do not modify the original dict

        for attribute, value in data.items():

            if 'uni' in attribute:
                try:
                    data[attribute] = self.uni_from_dict(value)
                except ValueError as exc:
                    raise ValueError(f'Error creating UNI: {exc}')

            if attribute == 'circuit_scheduler':
                data[attribute] = []
                for schedule in value:
                    data[attribute].append(CircuitSchedule.from_dict(schedule))

            if 'link' in attribute:
                if value:
                    data[attribute] = self.link_from_dict(value)

            if 'path' in attribute and attribute != 'dynamic_backup_path':
                if value:
                    data[attribute] = [self.link_from_dict(link)
                                       for link in value]

        return EVC(self.controller, **data)

    def uni_from_dict(self, uni_dict):
        """Return a UNI object from python dict."""
        if uni_dict is None:
            return False

        interface_id = uni_dict.get("interface_id")
        interface = self.controller.get_interface_by_id(interface_id)
        if interface is None:
            raise ValueError(f'Could not instantiate interface {interface_id}')

        tag_dict = uni_dict.get("tag")
        tag = TAG.from_dict(tag_dict)
        if tag is False:
            raise ValueError(f'Could not instantiate tag from dict {tag_dict}')

        uni = UNI(interface, tag)

        return uni

    def link_from_dict(self, link_dict):
        """Return a Link object from python dict."""
        id_a = link_dict.get('endpoint_a').get('id')
        id_b = link_dict.get('endpoint_b').get('id')

        endpoint_a = self.controller.get_interface_by_id(id_a)
        endpoint_b = self.controller.get_interface_by_id(id_b)

        link = Link(endpoint_a, endpoint_b)
        if 'metadata' in link_dict:
            link.extend_metadata(link_dict.get('metadata'))

        s_vlan = link.get_metadata('s_vlan')
        if s_vlan:
            tag = TAG.from_dict(s_vlan)
            if tag is False:
                error_msg = f'Could not instantiate tag from dict {s_vlan}'
                raise ValueError(error_msg)
            link.update_metadata('s_vlan', tag)
        return link


1		"""Main module of kytos/mef_eline Kytos Network Application.
2
3		NApp to provision circuits from user request.
4		"""
5	1	from flask import jsonify, request
6
7	1	from kytos.core import KytosNApp, log, rest
8	1	from kytos.core.events import KytosEvent
9	1	from kytos.core.helpers import listen_to
10	1	from kytos.core.interface import TAG, UNI
11	1	from kytos.core.link import Link
12	1	from napps.kytos.mef_eline.models import EVC, DynamicPathManager
13	1	from napps.kytos.mef_eline.scheduler import CircuitSchedule, Scheduler
14	1	from napps.kytos.mef_eline.storehouse import StoreHouse
15
16
17	1	class Main(KytosNApp):
18		"""Main class of amlight/mef_eline NApp.
19
20		This class is the entry point for this napp.
21		"""
22
23	1	def setup(self):
24		"""Replace the '__init__' method for the KytosNApp subclass.
25
26		The setup method is automatically called by the controller when your
27		application is loaded.
28
29		So, if you have any setup routine, insert it here.
30		"""
31		# object used to scheduler circuit events
32	1	self.sched = Scheduler()
33
34		# object to save and load circuits
35	1	self.storehouse = StoreHouse(self.controller)
36
37		# set the controller that will manager the dynamic paths
38	1	DynamicPathManager.set_controller(self.controller)
39
40	1	def execute(self):
41		"""Execute once when the napp is running."""
42		pass
43
44	1	def shutdown(self):
45		"""Execute when your napp is unloaded.
46
47		If you have some cleanup procedure, insert it here.
48		"""
49		pass
50
51	1	@rest('/v2/evc/', methods=['GET'])
52		def list_circuits(self):
53		"""Endpoint to return all circuits stored."""
54	1	circuits = self.storehouse.get_data()
55	1	if not circuits:
56	1	return jsonify({}), 200
57
58	1	return jsonify(circuits), 200
59
60	1	@rest('/v2/evc/<circuit_id>', methods=['GET'])
61		def get_circuit(self, circuit_id):
62		"""Endpoint to return a circuit based on id."""
63	1	circuits = self.storehouse.get_data()
64
65	1	if circuit_id in circuits:
66	1	result = circuits[circuit_id]
67	1	status = 200
68		else:
69	1	result = {'response': f'circuit_id {circuit_id} not found'}
70	1	status = 400
71
72	1	return jsonify(result), status
73
74	1	@rest('/v2/evc/', methods=['POST'])
75		def create_circuit(self):
76		"""Try to create a new circuit.
77
78		Firstly, for EVPL: E-Line NApp verifies if UNI_A's requested C-VID and
79		UNI_Z's requested C-VID are available from the interfaces' pools. This
80		is checked when creating the UNI object.
81
82		Then, E-Line NApp requests a primary and a backup path to the
83		Pathfinder NApp using the attributes primary_links and backup_links
84		submitted via REST
85
86		# For each link composing paths in #3:
87		# - E-Line NApp requests a S-VID available from the link VLAN pool.
88		# - Using the S-VID obtained, generate abstract flow entries to be
89		# sent to FlowManager
90
91		Push abstract flow entries to FlowManager and FlowManager pushes
92		OpenFlow entries to datapaths
93
94		E-Line NApp generates an event to notify all Kytos NApps of a new EVC
95		creation
96
97		Finnaly, notify user of the status of its request.
98		"""
99		# Try to create the circuit object
100	1	data = request.get_json()
101
102	1	if not data:
103	1	return jsonify("Bad request: The request do not have a json."), 400
104
105	1	try:
106	1	evc = self.evc_from_dict(data)
107		except ValueError as exception:
108		return jsonify("Bad request: {}".format(exception)), 400
109
110		# verify duplicated evc
111	1	if self.is_duplicated_evc(evc):
112		return jsonify("Not Acceptable: This evc already exists."), 409
113
114		# save circuit
115	1	self.storehouse.save_evc(evc)
116
117		# Schedule the circuit deploy
118	1	self.sched.add(evc)
119	1
120		# Circuit has no schedule, deploy now
121		if not evc.circuit_scheduler:
122	1	evc.deploy()
123
124	1	# Notify users
125		event = KytosEvent(name='kytos.mef_eline.created',
126	1	content=evc.as_dict())
127		self.controller.buffers.app.put(event)
128	1
129		return jsonify({"circuit_id": evc.id}), 201
130
131		@rest('/v2/evc/<circuit_id>', methods=['PATCH'])
132		def update(self, circuit_id):
133		"""Update a circuit based on payload.
134
135		The EVC required attributes can't be updated.
136		"""
137		data = request.get_json()
138		circuits = self.storehouse.get_data()
139
140		if circuit_id not in circuits:
141		result = {'response': f'circuit_id {circuit_id} not found'}
142		return jsonify(result), 404
143
144		try:
145		evc = self.evc_from_dict(circuits.get(circuit_id))
146		evc.update(**data)
147		self.storehouse.save_evc(evc)
148		result = {evc.id: evc.as_dict()}
149		status = 200
150		except ValueError as exception:
151		result = "Bad request: {}".format(exception)
152		status = 400
153	1
154		return jsonify(result), status
155
156		@rest('/v2/evc/<circuit_id>', methods=['DELETE'])
157		def delete_circuit(self, circuit_id):
158		"""Remove a circuit.
159
160		First, flows are removed from the switches, then the EVC is
161		disabled.
162		"""
163		circuits = self.storehouse.get_data()
164		log.info("Removing %s" % circuit_id)
165		evc = self.evc_from_dict(circuits.get(circuit_id))
166		evc.remove_current_flows()
167		evc.disable()
168		self.storehouse.save_evc(evc)
169	1
170		return jsonify("Circuit removed"), 200
171
172		def is_duplicated_evc(self, evc):
173		"""Verify if the circuit given is duplicated with the stored evcs.
174
175		Args:
176		evc (EVC): circuit to be analysed.
177
178		Returns:
179	1	boolean: True if the circuit is duplicated, otherwise False.
180
181		"""
182		for circuit_dict in self.storehouse.get_data().values():
183		try:
184		circuit = self.evc_from_dict(circuit_dict)
185		except ValueError:
186		continue
187
188	1	if circuit == evc:
189		return True
190	1
191		return False
192
193		@listen_to('kytos/topology.link_up')
194		def handle_link_up(self, event):
195		"""Change circuit when link is up or end_maintenance."""
196		evc = None
197
198		for data in self.storehouse.get_data().values():
199		try:
200		evc = self.evc_from_dict(data)
201		except ValueError as _exception:
202		log.debug(f'{data.get("id")} can not be provisioning yet.')
203	1	continue
204
205		evc.handle_link_up(event.content['link'])
206
207		@listen_to('kytos/topology.link_down')
208		def handle_link_down(self, event):
209		"""Change circuit when link is down or under_mantenance."""
210		evc = None
211
212		for data in self.storehouse.get_data().values():
213		try:
214		evc = self.evc_from_dict(data)
215		except ValueError as _exception:
216		log.debug(f'{data.get("id")} can not be provisioned yet.')
217	1	continue
218
219		if evc.is_affected_by_link(event.content['link']):
220		log.info('handling evc %s' % evc)
221		evc.handle_link_down()
222
223		def evc_from_dict(self, evc_dict):
224		"""Convert some dict values to instance of EVC classes.
225
226		This method will convert: [UNI, Link]
227		"""
228		data = evc_dict.copy() # Do not modify the original dict
229
230		for attribute, value in data.items():
231
232		if 'uni' in attribute:
233	1	try:
234		data[attribute] = self.uni_from_dict(value)
235		except ValueError as exc:
236		raise ValueError(f'Error creating UNI: {exc}')
237
238	1	if attribute == 'circuit_scheduler':
239		data[attribute] = []
240	1	for schedule in value:
241		data[attribute].append(CircuitSchedule.from_dict(schedule))
242	1
243	1	if 'link' in attribute:
244	1	if value:
245		data[attribute] = self.link_from_dict(value)
246
247		if 'path' in attribute and attribute != 'dynamic_backup_path':
248	1	if value:
249		data[attribute] = [self.link_from_dict(link)
250		for link in value]
251
252		return EVC(self.controller, **data)
253	1
254		def uni_from_dict(self, uni_dict):
255		"""Return a UNI object from python dict."""
256		if uni_dict is None:
257	1	return False
258
259		interface_id = uni_dict.get("interface_id")
260		interface = self.controller.get_interface_by_id(interface_id)
261		if interface is None:
262	1	raise ValueError(f'Could not instantiate interface {interface_id}')
263
264	1	tag_dict = uni_dict.get("tag")
265		tag = TAG.from_dict(tag_dict)
266		if tag is False:
267		raise ValueError(f'Could not instantiate tag from dict {tag_dict}')
268
269		uni = UNI(interface, tag)
270
271		return uni
272
273		def link_from_dict(self, link_dict):
274		"""Return a Link object from python dict."""
275		id_a = link_dict.get('endpoint_a').get('id')
276		id_b = link_dict.get('endpoint_b').get('id')
277
278		endpoint_a = self.controller.get_interface_by_id(id_a)
279		endpoint_b = self.controller.get_interface_by_id(id_b)
280
281		link = Link(endpoint_a, endpoint_b)
282		if 'metadata' in link_dict:
283	1	link.extend_metadata(link_dict.get('metadata'))
284
285		s_vlan = link.get_metadata('s_vlan')
286		if s_vlan:
287		tag = TAG.from_dict(s_vlan)
288		if tag is False:
289		error_msg = f'Could not instantiate tag from dict {s_vlan}'
290		raise ValueError(error_msg)
291		link.update_metadata('s_vlan', tag)
292		return link
293

kytos / mef_eline

Pull Request — master (#116)

build.main.Main.create_circuit() B

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like