build.main.Main.create_circuit() - Code Metrics - Inspection of "Fix enabled" - kytos/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#137)

by Antonio

created 2019-06-10 15:59 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	15
CRAP Score	5.0406

Importance

Changes

Metric	Value
cc	5
eloc	19
nop	1
dl	0
loc	56
ccs	15
cts	17
cp	0.8824
crap	5.0406
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."""

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

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

    @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 (name, uni_a, uni_z) 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, the flows are removed from the switches, and 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.deactivate()
        evc.disable()
        for circuit_scheduler in evc.circuit_scheduler:
            self.sched.cancel_job(circuit_scheduler.id)
        evc.archive()
        evc.sync()

        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 not evc.archived and 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

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

kytos / mef_eline

Pull Request — master (#137)

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