build.main.Main.is_duplicated_evc() - Code Metrics - Inspection of "Load EVCs from storehouse on port created event" - kytos/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#155)

by Antonio

created 2019-08-14 18:35 UTC

build.main.Main.is_duplicated_evc() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	14
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	4

Importance

Changes

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

"""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 = {}

        # dictionary of EVCs by interface
        self._circuits_by_interface = {}

    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)
        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:
            response = 'Bad Request: The request is not a valid JSON.'
            result = {'response': response}
            status = 400
        except KeyError:
            result = {'response': f'circuit_id {circuit_id} not found'}
            status = 404
        else:
            evc.sync()
            result = {evc.id: evc.as_dict()}
            status = 200

        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]
        except KeyError:
            result = {'response': f'circuit_id {circuit_id} not found'}
            status = 404
        else:
            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

        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 load_circuits_by_interface(self, circuits):
        """Load circuits in storehouse for in-memory dictionary."""
        for circuit_id, circuit in circuits.items():
            intf_a = circuit['uni_a']['interface_id']
            self.add_to_circuits_by_interface(intf_a, circuit_id)
            intf_z = circuit['uni_z']['interface_id']
            self.add_to_circuits_by_interface(intf_z, circuit_id)
            for path in ('current_path', 'primary_path', 'backup_path'):
                for link in circuit[path]:
                    intf_a = link['endpoint_a']['id']
                    self.add_to_circuits_by_interface(intf_a, circuit_id)
                    intf_b = link['endpoint_b']['id']
                    self.add_to_circuits_by_interface(intf_b, circuit_id)

    def add_to_circuits_by_interface(self, intf, circuit_id):
        """Add a single item to the dictionary of circuits by interface."""
        if intf not in self._circuits_by_interface:
            self._circuits_by_interface[intf] = set()
        self._circuits_by_interface[intf].add(circuit_id)

    @listen_to('kytos/of_core.switch.port.created')
    def load_evcs(self, event):
        """Try to load the unloaded EVCs from storehouse."""
        circuits = self.storehouse.get_data()
        if not self._circuits_by_interface:
            self.load_circuits_by_interface(circuits)

        interface_id = '{}:{}'.format(event.content['switch'],
                                      event.content['port'])

        for circuit_id in self._circuits_by_interface.get(interface_id, []):
            if circuit_id in circuits and circuit_id not in self.circuits:
                try:
                    evc = self.evc_from_dict(circuits[circuit_id])
                except ValueError as exception:
                    log.info(
                        f'Could not load EVC {circuit_id} because {exception}')
                    continue
                log.info(f'Loading EVC {circuit_id}')
                if evc.archived:
                    continue
                if evc.is_enabled():
                    log.info(f'Trying to deploy EVC {circuit_id}')
                    evc.deploy()
                self.circuits[circuit_id] = evc
                self.sched.add(evc)

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

kytos / mef_eline

Pull Request — master (#155)

build.main.Main.is_duplicated_evc() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like