build.main.Main.list_schedules() - Code Metrics - Inspection of "New REST API's to EVC Schedules" - kytos/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#140)

by Rogerio

created 2019-09-03 22:04 UTC

build.main.Main.list_schedules() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	14
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	4

Importance

Changes

Metric	Value
cc	4
eloc	11
nop	1
dl	0
loc	14
ccs	5
cts	5
cp	1
crap	4
rs	9.85
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. It acts as a circuit buffer.
        # Every create/update/delete must be synced to storehouse.
        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:
            log.error(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:
            log.error(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

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

        result = []
        for circuit in circuits:
            if "circuit_scheduler" in circuit:
                schedule = {circuit["id"]: circuit["circuit_scheduler"]}
                result.append(schedule)

        return jsonify(result), 200

    @rest('/v2/evc/schedule/', methods=['POST'])
    def create_schedule(self):
        """
        Create a new schedule for a given circuit.

        This service do no check if there are conflicts with another schedule.
        Payload example:
            {
              "circuit_id":"aa:bb:cc",
              "schedule": {
                "date": "2019-08-07T14:52:10.967Z",
                "interval": "string",
                "frequency": "1 * * * *",
                "action": "create"
              }
            }
        """
        try:
            # Try to create the circuit object
            json_data = request.get_json()
            result = ""
            status = 200
            if not json_data:
                result = "Bad request: The request does not have a json."
                status = 400
            elif "circuit_id" not in json_data:
                result = result = "Bad request: Missing circuit_id."
                status = 400
            elif "schedule" not in json_data:
                result = "Bad request: Missing schedule data."
                status = 400
            else:
                circuit_id = json_data["circuit_id"]
                schedule_data = json_data["schedule"]

                # Get EVC from circuits buffer
                circuits = self.get_circuits_buffer()

                # get the circuit
                evc = circuits.get(circuit_id, None)

                # get the circuit
                if not evc:
                    result = {'response': f'circuit_id {circuit_id} not found'}
                    status = 404
                else:
                    # Can not modify circuits deleted and archived
                    if evc.archived:
                        result = {'response': f'Circuit is archived.'
                                              f'Update is forbidden.'}
                        status = 403
                    else:
                        # new schedule from dict
                        new_schedule = CircuitSchedule.from_dict(schedule_data)

                        # If there is no schedule, create the list
                        if not evc.circuit_scheduler:
                            evc.circuit_scheduler = []

                        # Add the new schedule
                        evc.circuit_scheduler.append(new_schedule)

                        # Add schedule job
                        self.sched.add_circuit_job(evc, new_schedule)

                        # save circuit to storehouse
                        evc.sync()

                        result = new_schedule.as_dict()
                        status = 201

        except ValueError as exception:
            log.error(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

        return jsonify(result), status

    @rest('/v2/evc/schedule/<schedule_id>', methods=['PATCH'])
    def update_schedule(self, schedule_id):
        """Update a schedule.

        Change all attributes from the given schedule from a EVC circuit.
        The schedule ID is preserved as default.
        Payload example:
            {
              "date": "2019-08-07T14:52:10.967Z",
              "interval": "string",
              "frequency": "1 * * *",
              "action": "create"
            }
        """
        try:
            # Try to find a circuit schedule
            evc, found_schedule = self.find_evc_by_schedule_id(schedule_id)

            # Can not modify circuits deleted and archived
            if not found_schedule:
                result = {'response':
                          f'schedule_id {schedule_id} not found'}
                status = 404
            else:
                if evc.archived:
                    result = {'response': f'Circuit is archived.'
                                          f'Update is forbidden.'}
                    status = 403
                else:
                    data = request.get_json()

                    new_schedule = CircuitSchedule.from_dict(data)
                    new_schedule.id = found_schedule.id
                    # Remove the old schedule
                    evc.circuit_scheduler.remove(found_schedule)
                    # Append the modified schedule
                    evc.circuit_scheduler.append(new_schedule)

                    # Cancel all schedule jobs
                    self.sched.cancel_job(found_schedule.id)
                    # Add the new circuit schedule
                    self.sched.add_circuit_job(evc, new_schedule)
                    # Save EVC to the storehouse
                    evc.sync()

                    result = new_schedule.as_dict()
                    status = 200

        except ValueError as exception:
            log.error(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 a valid JSON.'}
            status = 400

        return jsonify(result), status

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

        Remove the Schedule from EVC.
        Remove the Schedule from cron job.
        Save the EVC to the Storehouse.
        """
        evc, found_schedule = self.find_evc_by_schedule_id(schedule_id)

        # Can not modify circuits deleted and archived
        if not found_schedule:
            result = {'response': f'schedule_id {schedule_id} not found'}
            status = 404
        else:
            if evc.archived:
                result = {'response': f'Circuit is archived.'
                                      f'Update is forbidden.'}
                status = 403
            else:
                # Remove the old schedule
                evc.circuit_scheduler.remove(found_schedule)

                # Cancel all schedule jobs
                self.sched.cancel_job(found_schedule.id)
                # Save EVC to the storehouse
                evc.sync()

                result = "Schedule 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_dict_of_sets(intf_a, circuit_id)
            intf_z = circuit['uni_z']['interface_id']
            self.add_to_dict_of_sets(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_dict_of_sets(intf_a, circuit_id)
                    intf_b = link['endpoint_b']['id']
                    self.add_to_dict_of_sets(intf_b, circuit_id)

    def add_to_dict_of_sets(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/topology.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():
            # Get multiple attributes.
            # Ex: uni_a, uni_z
            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))

            # Get multiple attributes.
            # Ex: primary_links,
            #     backup_links,
            #     current_links_cache,
            #     primary_links_cache,
            #     backup_links_cache
            if 'links' in attribute:
                data[attribute] = [self.link_from_dict(link)
                                   for link in value]

            # Get multiple attributes.
            # Ex: current_path,
            #     primary_path,
            #     backup_path
            if 'path' in attribute and attribute != 'dynamic_backup_path':
                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

    def find_evc_by_schedule_id(self, schedule_id):
        """
        Find an EVC and CircuitSchedule based on schedule_id.

        :param schedule_id: Schedule ID
        :return: EVC and Schedule
        """
        circuits = self.get_circuits_buffer()
        found_schedule = None
        evc = None

        # pylint: disable=unused-variable
        for c_id, circuit in circuits.items():
            for schedule in circuit.circuit_scheduler:
                if schedule.id == schedule_id:
                    found_schedule = schedule
                    evc = circuit
                    break
            if found_schedule:
                break
        return evc, found_schedule

    def get_circuits_buffer(self):
        """
        Return the circuit buffer.

        If the buffer is empty, try to load data from storehouse.
        """
        if not self.circuits:
            # Load storehouse circuits to buffer
            circuits = self.storehouse.get_data()
            for c_id, circuit in circuits.items():
                evc = self.evc_from_dict(circuit)
                self.circuits[c_id] = evc
        return self.circuits


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. It acts as a circuit buffer.
42	2	# Every create/update/delete must be synced to storehouse.
43		self.circuits = {}
44
45	2	# dictionary of EVCs by interface
46		self._circuits_by_interface = {}
47	2
48		def execute(self):
49		"""Execute once when the napp is running."""
50	2
51		def shutdown(self):
52		"""Execute when your napp is unloaded.
53
54		If you have some cleanup procedure, insert it here.
55		"""
56	2
57		@rest('/v2/evc/', methods=['GET'])
58		def list_circuits(self):
59	2	"""Endpoint to return all circuits stored."""
60	2	circuits = self.storehouse.get_data()
61	2	if not circuits:
62		return jsonify({}), 200
63	2
64		return jsonify(circuits), 200
65	2
66		@rest('/v2/evc/<circuit_id>', methods=['GET'])
67		def get_circuit(self, circuit_id):
68	2	"""Endpoint to return a circuit based on id."""
69	2	circuits = self.storehouse.get_data()
70	2
71	2	try:
72	2	result = circuits[circuit_id]
73	2	status = 200
74	2	except KeyError:
75		result = {'response': f'circuit_id {circuit_id} not found'}
76	2	status = 404
77
78	2	return jsonify(result), status
79
80		@rest('/v2/evc/', methods=['POST'])
81		def create_circuit(self):
82		"""Try to create a new circuit.
83
84		Firstly, for EVPL: E-Line NApp verifies if UNI_A's requested C-VID and
85		UNI_Z's requested C-VID are available from the interfaces' pools. This
86		is checked when creating the UNI object.
87
88		Then, E-Line NApp requests a primary and a backup path to the
89		Pathfinder NApp using the attributes primary_links and backup_links
90		submitted via REST
91
92		# For each link composing paths in #3:
93		# - E-Line NApp requests a S-VID available from the link VLAN pool.
94		# - Using the S-VID obtained, generate abstract flow entries to be
95		# sent to FlowManager
96
97		Push abstract flow entries to FlowManager and FlowManager pushes
98		OpenFlow entries to datapaths
99
100		E-Line NApp generates an event to notify all Kytos NApps of a new EVC
101		creation
102
103		Finnaly, notify user of the status of its request.
104	2	"""
105		# Try to create the circuit object
106	2	data = request.get_json()
107	2
108		if not data:
109	2	return jsonify("Bad request: The request do not have a json."), 400
110	2
111		try:
112		evc = self.evc_from_dict(data)
113		except ValueError as exception:
114		log.error(exception)
115	2	return jsonify("Bad request: {}".format(exception)), 400
116	2
117		# verify duplicated evc
118		if self.is_duplicated_evc(evc):
119	2	return jsonify("Not Acceptable: This evc already exists."), 409
120
121		# store circuit in dictionary
122	2	self.circuits[evc.id] = evc
123
124		# save circuit
125	2	self.storehouse.save_evc(evc)
126
127		# Schedule the circuit deploy
128	2	self.sched.add(evc)
129	2
130		# Circuit has no schedule, deploy now
131		if not evc.circuit_scheduler:
132	2	evc.deploy()
133
134	2	# Notify users
135		event = KytosEvent(name='kytos.mef_eline.created',
136	2	content=evc.as_dict())
137		self.controller.buffers.app.put(event)
138	2
139		return jsonify({"circuit_id": evc.id}), 201
140
141		@rest('/v2/evc/<circuit_id>', methods=['PATCH'])
142		def update(self, circuit_id):
143		"""Update a circuit based on payload.
144
145		The EVC required attributes (name, uni_a, uni_z) can't be updated.
146		"""
147		try:
148		evc = self.circuits[circuit_id]
149		data = request.get_json()
150		evc.update(**data)
151		except ValueError as exception:
152		log.error(exception)
153		result = {'response': 'Bad Request: {}'.format(exception)}
154		status = 400
155		except TypeError:
156		result = {'response': 'Content-Type must be application/json'}
157		status = 415
158		except BadRequest:
159		response = 'Bad Request: The request is not a valid JSON.'
160		result = {'response': response}
161		status = 400
162		except KeyError:
163		result = {'response': f'circuit_id {circuit_id} not found'}
164		status = 404
165		else:
166		evc.sync()
167		result = {evc.id: evc.as_dict()}
168	2	status = 200
169
170		return jsonify(result), status
171
172		@rest('/v2/evc/<circuit_id>', methods=['DELETE'])
173		def delete_circuit(self, circuit_id):
174		"""Remove a circuit.
175
176		First, the flows are removed from the switches, and then the EVC is
177		disabled.
178		"""
179		try:
180		evc = self.circuits[circuit_id]
181		except KeyError:
182		result = {'response': f'circuit_id {circuit_id} not found'}
183		status = 404
184		else:
185		log.info(f'Removing {circuit_id}')
186		if evc.archived:
187		result = {'response': f'Circuit {circuit_id} already removed'}
188		status = 404
189		else:
190		evc.remove_current_flows()
191		evc.deactivate()
192		evc.disable()
193		self.sched.remove(evc)
194		evc.archive()
195		evc.sync()
196		result = {'response': f'Circuit {circuit_id} removed'}
197	2	status = 200
198
199		return jsonify(result), status
200
201		@rest('/v2/evc/schedule', methods=['GET'])
202		def list_schedules(self):
203		"""Endpoint to return all circuits stored."""
204		circuits = self.storehouse.get_data().values()
205		if not circuits:
206		return jsonify({}), 200
207	2
208	2	result = []
209	2	for circuit in circuits:
210	2	if "circuit_scheduler" in circuit:
211		schedule = {circuit["id"]: circuit["circuit_scheduler"]}
212	2	result.append(schedule)
213
214		return jsonify(result), 200
215
216		@rest('/v2/evc/schedule/', methods=['POST'])
217		def create_schedule(self):
218		"""
219	2	Create a new schedule for a given circuit.
220
221		This service do no check if there are conflicts with another schedule.
222		Payload example:
223		{
224		"circuit_id":"aa:bb:cc",
225		"schedule": {
226		"date": "2019-08-07T14:52:10.967Z",
227	2	"interval": "string",
228		"frequency": "1 * * * *",
229	2	"action": "create"
230	2	}
231	2	}
232	2	"""
233	2	try:
234	2	# Try to create the circuit object
235	2	json_data = request.get_json()
236	2	result = ""
237	2	status = 200
238	2	if not json_data:
239	2	result = "Bad request: The request does not have a json."
240		status = 400
241	2	elif "circuit_id" not in json_data:
242		result = result = "Bad request: Missing circuit_id."
243	2	status = 400
244	2	elif "schedule" not in json_data:
245	2	result = "Bad request: Missing schedule data."
246		status = 400
247	2	else:
248		circuit_id = json_data["circuit_id"]
249		schedule_data = json_data["schedule"]
250
251		# Get EVC from circuits buffer
252		circuits = self.get_circuits_buffer()
253
254		# get the circuit
255		evc = circuits.get(circuit_id, None)
256
257		# get the circuit
258		if not evc:
259		result = {'response': f'circuit_id {circuit_id} not found'}
260		status = 404
261		else:
262		# Can not modify circuits deleted and archived
263		if evc.archived:
264		result = {'response': f'Circuit is archived.'
265		f'Update is forbidden.'}
266		status = 403
267		else:
268		# new schedule from dict
269		new_schedule = CircuitSchedule.from_dict(schedule_data)
270
271		# If there is no schedule, create the list
272		if not evc.circuit_scheduler:
273		evc.circuit_scheduler = []
274	2
275		# Add the new schedule
276		evc.circuit_scheduler.append(new_schedule)
277
278		# Add schedule job
279	2	self.sched.add_circuit_job(evc, new_schedule)
280
281	2	# save circuit to storehouse
282		evc.sync()
283	2
284	2	result = new_schedule.as_dict()
285	2	status = 201
286
287		except ValueError as exception:
288		log.error(exception)
289	2	result = {'response': 'Bad Request: {}'.format(exception)}
290		status = 400
291		except TypeError:
292		result = {'response': 'Content-Type must be application/json'}
293		status = 415
294	2	except BadRequest:
295		response = 'Bad Request: The request is not a valid JSON.'
296		result = {'response': response}
297		status = 400
298	2
299		return jsonify(result), status
300
301		@rest('/v2/evc/schedule/<schedule_id>', methods=['PATCH'])
302		def update_schedule(self, schedule_id):
303	2	"""Update a schedule.
304
305	2	Change all attributes from the given schedule from a EVC circuit.
306		The schedule ID is preserved as default.
307		Payload example:
308		{
309		"date": "2019-08-07T14:52:10.967Z",
310		"interval": "string",
311		"frequency": "1 * * *",
312		"action": "create"
313		}
314		"""
315		try:
316		# Try to find a circuit schedule
317		evc, found_schedule = self.find_evc_by_schedule_id(schedule_id)
318
319		# Can not modify circuits deleted and archived
320		if not found_schedule:
321		result = {'response':
322		f'schedule_id {schedule_id} not found'}
323		status = 404
324	2	else:
325		if evc.archived:
326		result = {'response': f'Circuit is archived.'
327		f'Update is forbidden.'}
328		status = 403
329		else:
330		data = request.get_json()
331
332		new_schedule = CircuitSchedule.from_dict(data)
333		new_schedule.id = found_schedule.id
334		# Remove the old schedule
335		evc.circuit_scheduler.remove(found_schedule)
336		# Append the modified schedule
337		evc.circuit_scheduler.append(new_schedule)
338
339		# Cancel all schedule jobs
340		self.sched.cancel_job(found_schedule.id)
341		# Add the new circuit schedule
342		self.sched.add_circuit_job(evc, new_schedule)
343		# Save EVC to the storehouse
344		evc.sync()
345
346		result = new_schedule.as_dict()
347		status = 200
348
349		except ValueError as exception:
350		log.error(exception)
351		result = {'response': 'Bad Request: {}'.format(exception)}
352		status = 400
353		except TypeError:
354		result = {'response': 'Content-Type must be application/json'}
355		status = 415
356		except BadRequest:
357		result = {'response':
358		'Bad Request: The request is not a valid JSON.'}
359		status = 400
360
361		return jsonify(result), status
362
363		@rest('/v2/evc/schedule/<schedule_id>', methods=['DELETE'])
364		def delete_schedule(self, schedule_id):
365		"""Remove a circuit schedule.
366
367		Remove the Schedule from EVC.
368		Remove the Schedule from cron job.
369		Save the EVC to the Storehouse.
370		"""
371		evc, found_schedule = self.find_evc_by_schedule_id(schedule_id)
372
373		# Can not modify circuits deleted and archived
374		if not found_schedule:
375		result = {'response': f'schedule_id {schedule_id} not found'}
376		status = 404
377		else:
378		if evc.archived:
379		result = {'response': f'Circuit is archived.'
380		f'Update is forbidden.'}
381		status = 403
382		else:
383		# Remove the old schedule
384		evc.circuit_scheduler.remove(found_schedule)
385
386		# Cancel all schedule jobs
387		self.sched.cancel_job(found_schedule.id)
388		# Save EVC to the storehouse
389		evc.sync()
390
391		result = "Schedule removed"
392		status = 200
393
394		return jsonify(result), status
395
396		def is_duplicated_evc(self, evc):
397		"""Verify if the circuit given is duplicated with the stored evcs.
398
399		Args:
400		evc (EVC): circuit to be analysed.
401
402		Returns:
403		boolean: True if the circuit is duplicated, otherwise False.
404
405		"""
406		for circuit in self.circuits.values():
407		if not circuit.archived and circuit == evc:
408		return True
409		return False
410
411		@listen_to('kytos/topology.link_up')
412		def handle_link_up(self, event):
413		"""Change circuit when link is up or end_maintenance."""
414		for evc in self.circuits.values():
415		if evc.is_enabled() and not evc.archived:
416		evc.handle_link_up(event.content['link'])
417
418		@listen_to('kytos/topology.link_down')
419		def handle_link_down(self, event):
420		"""Change circuit when link is down or under_mantenance."""
421		for evc in self.circuits.values():
422		if evc.is_affected_by_link(event.content['link']):
423		log.info('handling evc %s' % evc)
424		evc.handle_link_down()
425
426		def load_circuits_by_interface(self, circuits):
427		"""Load circuits in storehouse for in-memory dictionary."""
428		for circuit_id, circuit in circuits.items():
429		intf_a = circuit['uni_a']['interface_id']
430		self.add_to_dict_of_sets(intf_a, circuit_id)
431		intf_z = circuit['uni_z']['interface_id']
432		self.add_to_dict_of_sets(intf_z, circuit_id)
433		for path in ('current_path', 'primary_path', 'backup_path'):
434		for link in circuit[path]:
435		intf_a = link['endpoint_a']['id']
436		self.add_to_dict_of_sets(intf_a, circuit_id)
437		intf_b = link['endpoint_b']['id']
438		self.add_to_dict_of_sets(intf_b, circuit_id)
439
440		def add_to_dict_of_sets(self, intf, circuit_id):
441		"""Add a single item to the dictionary of circuits by interface."""
442		if intf not in self._circuits_by_interface:
443		self._circuits_by_interface[intf] = set()
444		self._circuits_by_interface[intf].add(circuit_id)
445
446		@listen_to('kytos/topology.port.created')
447		def load_evcs(self, event):
448		"""Try to load the unloaded EVCs from storehouse."""
449		circuits = self.storehouse.get_data()
450		if not self._circuits_by_interface:
451		self.load_circuits_by_interface(circuits)
452
453		interface_id = '{}:{}'.format(event.content['switch'],
454		event.content['port'])
455
456		for circuit_id in self._circuits_by_interface.get(interface_id, []):
457		if circuit_id in circuits and circuit_id not in self.circuits:
458		try:
459		evc = self.evc_from_dict(circuits[circuit_id])
460		except ValueError as exception:
461		log.info(
462		f'Could not load EVC {circuit_id} because {exception}')
463		continue
464		log.info(f'Loading EVC {circuit_id}')
465		if evc.archived:
466		continue
467		if evc.is_enabled():
468		log.info(f'Trying to deploy EVC {circuit_id}')
469		evc.deploy()
470		self.circuits[circuit_id] = evc
471		self.sched.add(evc)
472
473		def evc_from_dict(self, evc_dict):
474		"""Convert some dict values to instance of EVC classes.
475
476		This method will convert: [UNI, Link]
477		"""
478		data = evc_dict.copy() # Do not modify the original dict
479
480		for attribute, value in data.items():
481		# Get multiple attributes.
482		# Ex: uni_a, uni_z
483		if 'uni' in attribute:
484		try:
485		data[attribute] = self.uni_from_dict(value)
486		except ValueError as exc:
487		raise ValueError(f'Error creating UNI: {exc}')
488
489		if attribute == 'circuit_scheduler':
490		data[attribute] = []
491		for schedule in value:
492		data[attribute].append(CircuitSchedule.from_dict(schedule))
493
494		# Get multiple attributes.
495		# Ex: primary_links,
496		# backup_links,
497		# current_links_cache,
498		# primary_links_cache,
499		# backup_links_cache
500		if 'links' in attribute:
501		data[attribute] = [self.link_from_dict(link)
502		for link in value]
503
504		# Get multiple attributes.
505		# Ex: current_path,
506		# primary_path,
507		# backup_path
508		if 'path' in attribute and attribute != 'dynamic_backup_path':
509		data[attribute] = [self.link_from_dict(link)
510		for link in value]
511
512		return EVC(self.controller, **data)
513
514		def uni_from_dict(self, uni_dict):
515		"""Return a UNI object from python dict."""
516		if uni_dict is None:
517		return False
518
519		interface_id = uni_dict.get("interface_id")
520		interface = self.controller.get_interface_by_id(interface_id)
521		if interface is None:
522		raise ValueError(f'Could not instantiate interface {interface_id}')
523
524		tag_dict = uni_dict.get("tag")
525		tag = TAG.from_dict(tag_dict)
526		if tag is False:
527		raise ValueError(f'Could not instantiate tag from dict {tag_dict}')
528
529		uni = UNI(interface, tag)
530
531		return uni
532
533		def link_from_dict(self, link_dict):
534		"""Return a Link object from python dict."""
535		id_a = link_dict.get('endpoint_a').get('id')
536		id_b = link_dict.get('endpoint_b').get('id')
537
538		endpoint_a = self.controller.get_interface_by_id(id_a)
539		endpoint_b = self.controller.get_interface_by_id(id_b)
540
541		link = Link(endpoint_a, endpoint_b)
542		if 'metadata' in link_dict:
543		link.extend_metadata(link_dict.get('metadata'))
544
545		s_vlan = link.get_metadata('s_vlan')
546		if s_vlan:
547		tag = TAG.from_dict(s_vlan)
548		if tag is False:
549		error_msg = f'Could not instantiate tag from dict {s_vlan}'
550		raise ValueError(error_msg)
551		link.update_metadata('s_vlan', tag)
552		return link
553
554		def find_evc_by_schedule_id(self, schedule_id):
555		"""
556		Find an EVC and CircuitSchedule based on schedule_id.
557
558		:param schedule_id: Schedule ID
559		:return: EVC and Schedule
560		"""
561		circuits = self.get_circuits_buffer()
562		found_schedule = None
563		evc = None
564
565		# pylint: disable=unused-variable
566		for c_id, circuit in circuits.items():
567		for schedule in circuit.circuit_scheduler:
568		if schedule.id == schedule_id:
569		found_schedule = schedule
570		evc = circuit
571		break
572		if found_schedule:
573		break
574		return evc, found_schedule
575
576		def get_circuits_buffer(self):
577		"""
578		Return the circuit buffer.
579
580		If the buffer is empty, try to load data from storehouse.
581		"""
582		if not self.circuits:
583		# Load storehouse circuits to buffer
584		circuits = self.storehouse.get_data()
585		for c_id, circuit in circuits.items():
586		evc = self.evc_from_dict(circuit)
587		self.circuits[c_id] = evc
588		return self.circuits
589

kytos / mef_eline

Pull Request — master (#140)

build.main.Main.list_schedules() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like