build.main.Main.setup() - Code Metrics - Inspection of "Fix return messages and codes on REST API" - kytos/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#152)

by Antonio

created 2019-06-27 13:18 UTC

build.main.Main.setup() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	19
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	1

Importance

Changes

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

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

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

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)

        # dictionary of EVCs created
        self.circuits = {}

    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()
        try:
            result = circuits[circuit_id]
            status = 200
        except KeyError:
            result = {'response': f'circuit_id {circuit_id} not found'}
            status = 404

        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

        # store circuit in dictionary
        self.circuits[evc.id] = evc

        # 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.
        """
        try:
            evc = self.circuits[circuit_id]
            data = request.get_json()
            evc.update(**data)
            evc.sync()
            result = {evc.id: evc.as_dict()}
            status = 200
        except ValueError as exception:
            result = {'response': 'Bad Request: {}'.format(exception)}
            status = 400
        except TypeError:
            result = {'response': 'Content-Type must be application/json'}
            status = 415
        except BadRequest:
            result = {'response': 'Bad Request: The request is not in JSON format.'}
            status = 400
        except KeyError:
            result = {'response': f'circuit_id {circuit_id} not found'}
            status = 404

        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.
        """
        try:
            evc = self.circuits[circuit_id]
            log.info(f'Removing {circuit_id}')
            if evc.archived:
                result = {'response': f'Circuit {circuit_id} already removed'}
                status = 404
            else:
                evc.remove_current_flows()
                evc.deactivate()
                evc.disable()
                self.sched.remove(evc)
                evc.archive()
                evc.sync()
                result = {'response': f'Circuit {circuit_id} removed'}
                status = 200
        except KeyError:
            result = {'response': f'circuit_id {circuit_id} not found'}
            status = 404

        return jsonify(result), status

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

kytos / mef_eline

Pull Request — master (#152)

build.main.Main.setup() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like