build.main.Main.create_mw() - Code Metrics - Inspection of "Merge pull request #78 from kytos-ng/feat/service_..." - kytos-ng/maintenance - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 0d5b5f...2f9122 )

unknown

created 2023-06-26 20:30 UTC

build.main.Main.create_mw() C

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	35
Code Lines	28

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	17
CRAP Score	11.0104

Importance

Changes

Metric	Value
cc	9
eloc	28
nop	2
dl	0
loc	35
ccs	17
cts	24
cp	0.7083
crap	11.0104
rs	6.6666
c	0
b	0
f	0

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

This NApp creates maintenance windows, allowing the maintenance of network
devices (switch, link, and interface) without receiving alerts.
"""
from datetime import timedelta

from napps.kytos.maintenance.managers import MaintenanceDeployer as Deployer
from napps.kytos.maintenance.managers import MaintenanceScheduler as Scheduler
from napps.kytos.maintenance.models import MaintenanceID
from napps.kytos.maintenance.models import MaintenanceWindow as MW
from napps.kytos.maintenance.models import OverlapError, Status
from pydantic import ValidationError
from pymongo.errors import DuplicateKeyError

from kytos.core import KytosNApp, rest
from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
                                 Response, error_msg, get_json_or_400)


class Main(KytosNApp):
    """Main class of kytos/maintenance 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.
        """
        self.maintenance_deployer = Deployer.new_deployer(self.controller)
        self.scheduler = Scheduler.new_scheduler(self.maintenance_deployer)
        self.scheduler.start()

    def execute(self):
        """Run after the setup method execution.

        You can also use this method in loop mode if you add to the above setup
        method a line like the following example:

            self.execute_as_loop(30)  # 30-second interval.
        """

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

        If you have some cleanup procedure, insert it here.
        """
        self.scheduler.shutdown()

    @rest('/v1', methods=['GET'])
    def get_all_mw(self, _request: Request) -> Response:
        """Return all maintenance windows."""
        maintenances = self.scheduler.list_maintenances()
        return Response(
            f"{maintenances.json()}\n",
            status_code=200,
            media_type="application/json",
        )

    @rest('/v1/{mw_id}', methods=['GET'])
    def get_mw(self, request: Request) -> Response:
        """Return one maintenance window."""
        mw_id: MaintenanceID = request.path_params["mw_id"]
        window = self.scheduler.get_maintenance(mw_id)
        if window:
            return Response(
                f"{window.json()}\n",
                status_code=200,
                media_type="application/json",
            )
        raise HTTPException(404, f'Maintenance with id {mw_id} not found')

    @rest('/v1', methods=['POST'])
    def create_mw(self, request: Response) -> JSONResponse:
        """Create a new maintenance window."""
        data = get_json_or_400(request, self.controller.loop)
        if not isinstance(data, dict) or not data:
            raise HTTPException(400, detail=f"Invalid json body value: {data}")

        if 'status' in data:
            raise HTTPException(
                400, detail='Setting a maintenance status is not allowed'
            )
        # if 'id' in data:
        #     raise HTTPException(
        #         400, detail='Setting a maintenance id is not allowed'
        #     )
        try:
            maintenance = MW.parse_obj(data)
            force = data.get('force', False)
            ignore_no_exists = data.get('ignore_no_exists')
            if not ignore_no_exists:
                self.validate_item_existence(maintenance)
            self.scheduler.add(maintenance, force=force)
        except ValidationError as err:
            msg = error_msg(err.errors())
            raise HTTPException(400, detail=msg) from err
        except DuplicateKeyError as err:
            raise HTTPException(
                409,
                detail=f'Window with id: {maintenance.id} already exists'
            ) from err
        except OverlapError as err:
            raise HTTPException(400, detail=f'{err}') from err
        except ValueError as err:
            raise HTTPException(400, detail=f'{err}') from err
        return JSONResponse({'mw_id': maintenance.id}, status_code=201)

    @rest('/v1/{mw_id}', methods=['PATCH'])
    def update_mw(self, request: Request) -> JSONResponse:
        """Update a maintenance window."""
        data = get_json_or_400(request, self.controller.loop)
        if not isinstance(data, dict) or not data:
            raise HTTPException(400, detail=f"Invalid json body value: {data}")

        mw_id: MaintenanceID = request.path_params["mw_id"]
        old_maintenance = self.scheduler.get_maintenance(mw_id)
        if old_maintenance is None:
            raise HTTPException(
                404, detail=f'Maintenance with id {mw_id} not found'
            )
        if old_maintenance.status == Status.RUNNING:
            raise HTTPException(
                400, detail='Updating a running maintenance is not allowed'
            )
        if 'status' in data:
            raise HTTPException(
                400, detail='Updating a maintenance status is not allowed'
            )
        try:
            new_maintenance = MW.parse_obj({**old_maintenance.dict(), **data})
        except ValidationError as err:
            msg = error_msg(err.errors())
            raise HTTPException(400, detail=msg) from err
        if new_maintenance.id != old_maintenance.id:
            raise HTTPException(400, detail='Updated id must match old id')
        self.scheduler.update(new_maintenance)
        return JSONResponse({'response': f'Maintenance {mw_id} updated'})

    @rest('/v1/{mw_id}', methods=['DELETE'])
    def remove_mw(self, request: Request) -> JSONResponse:
        """Delete a maintenance window."""
        mw_id: MaintenanceID = request.path_params["mw_id"]
        maintenance = self.scheduler.get_maintenance(mw_id)
        if maintenance is None:
            raise HTTPException(
                404, detail=f'Maintenance with id {mw_id} not found'
            )
        if maintenance.status == Status.RUNNING:
            raise HTTPException(
                400, detail='Deleting a running maintenance is not allowed'
            )
        self.scheduler.remove(mw_id)
        return JSONResponse({'response': f'Maintenance with id {mw_id} '
                                         f'successfully removed'})

    @rest('/v1/{mw_id}/end', methods=['PATCH'])
    def end_mw(self, request: Request) -> JSONResponse:
        """Finish a maintenance window right now."""
        mw_id: MaintenanceID = request.path_params["mw_id"]
        maintenance = self.scheduler.get_maintenance(mw_id)
        if maintenance is None:
            raise HTTPException(
                404, detail=f'Maintenance with id {mw_id} not found'
            )
        if maintenance.status == Status.PENDING:
            raise HTTPException(
                400, detail=f'Maintenance window {mw_id} has not yet started'
            )
        if maintenance.status == Status.FINISHED:
            raise HTTPException(
                400, detail=f'Maintenance window {mw_id} has already finished'
            )
        self.scheduler.end_maintenance_early(mw_id)
        return JSONResponse({'response': f'Maintenance window {mw_id} '
                                         f'finished'})

    @rest('/v1/{mw_id}/extend', methods=['PATCH'])
    def extend_mw(self, request: Request) -> JSONResponse:
        """Extend a running maintenance window."""
        mw_id: MaintenanceID = request.path_params["mw_id"]
        data = get_json_or_400(request, self.controller.loop)
        if not isinstance(data, dict):
            raise HTTPException(400, detail=f"Invalid json body value: {data}")

        maintenance = self.scheduler.get_maintenance(mw_id)
        if maintenance is None:
            raise HTTPException(
                404, detail=f'Maintenance with id {mw_id} not found'
            )
        if 'minutes' not in data:
            raise HTTPException(
                400,
                detail='Minutes of extension must be sent'
            )
        if maintenance.status == Status.PENDING:
            raise HTTPException(
                400,
                detail=f'Maintenance window {mw_id} has not yet started'
            )
        if maintenance.status == Status.FINISHED:
            raise HTTPException(
                400,
                detail=f'Maintenance window {mw_id} has already finished'
            )
        try:
            maintenance_end = maintenance.end + \
                timedelta(minutes=data['minutes'])
            new_maintenance = maintenance.copy(
                update={'end': maintenance_end}
            )
        except TypeError as exc:
            raise HTTPException(
                400,
                detail='Minutes of extension must be integer') from exc

        self.scheduler.update(new_maintenance)
        return JSONResponse({'response': f'Maintenance {mw_id} extended'})

    def validate_item_existence(self, window: MW):
        """Validate that all items in a maintenance window exist."""
        non_existant_switches = list(
            filter(
                lambda switch_id:
                    self.controller.switches.get(switch_id)
                    is None,
                window.switches
            )
        )
        non_existant_interfaces = list(
            filter(
                lambda interface_id:
                    self.controller.get_interface_by_id(interface_id)
                    is None,
                window.interfaces
            )
        )
        non_existant_links = list(
            filter(
                lambda interface_id:
                    self.controller.napps[('kytos', 'topology')]
                    .links.get(interface_id)
                    is None,
                window.interfaces
            )
        )

        if (
            non_existant_switches
            or non_existant_interfaces
            or non_existant_links
        ):
            items = {
                'switches': non_existant_switches,
                'interfaces': non_existant_interfaces,
                'links': non_existant_links,
            }
            raise HTTPException(
                400,
                f"Window contains non-existant items: {items}")


1		"""Main module of kytos/maintenance Kytos Network Application.
2
3		This NApp creates maintenance windows, allowing the maintenance of network
4		devices (switch, link, and interface) without receiving alerts.
5		"""
6	1	from datetime import timedelta
7
8	1	from napps.kytos.maintenance.managers import MaintenanceDeployer as Deployer
9	1	from napps.kytos.maintenance.managers import MaintenanceScheduler as Scheduler
10	1	from napps.kytos.maintenance.models import MaintenanceID
11	1	from napps.kytos.maintenance.models import MaintenanceWindow as MW
12	1	from napps.kytos.maintenance.models import OverlapError, Status
13	1	from pydantic import ValidationError
14	1	from pymongo.errors import DuplicateKeyError
15
16	1	from kytos.core import KytosNApp, rest
17	1	from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
18		Response, error_msg, get_json_or_400)
19
20
21	1	class Main(KytosNApp):
22		"""Main class of kytos/maintenance NApp.
23
24		This class is the entry point for this napp.
25		"""
26
27	1	def setup(self):
28		"""Replace the '__init__' method for the KytosNApp subclass.
29
30		The setup method is automatically called by the controller when your
31		application is loaded.
32
33		So, if you have any setup routine, insert it here.
34		"""
35	1	self.maintenance_deployer = Deployer.new_deployer(self.controller)
36	1	self.scheduler = Scheduler.new_scheduler(self.maintenance_deployer)
37	1	self.scheduler.start()
38
39	1	def execute(self):
40		"""Run after the setup method execution.
41
42		You can also use this method in loop mode if you add to the above setup
43		method a line like the following example:
44
45		self.execute_as_loop(30) # 30-second interval.
46		"""
47
48	1	def shutdown(self):
49		"""Run when your napp is unloaded.
50
51		If you have some cleanup procedure, insert it here.
52		"""
53		self.scheduler.shutdown()
54
55	1	@rest('/v1', methods=['GET'])
56	1	def get_all_mw(self, _request: Request) -> Response:
57		"""Return all maintenance windows."""
58	1	maintenances = self.scheduler.list_maintenances()
59	1	return Response(
60		f"{maintenances.json()}\n",
61		status_code=200,
62		media_type="application/json",
63		)
64
65	1	@rest('/v1/{mw_id}', methods=['GET'])
66	1	def get_mw(self, request: Request) -> Response:
67		"""Return one maintenance window."""
68	1	mw_id: MaintenanceID = request.path_params["mw_id"]
69	1	window = self.scheduler.get_maintenance(mw_id)
70	1	if window:
71	1	return Response(
72		f"{window.json()}\n",
73		status_code=200,
74		media_type="application/json",
75		)
76	1	raise HTTPException(404, f'Maintenance with id {mw_id} not found')
77
78	1	@rest('/v1', methods=['POST'])
79	1	def create_mw(self, request: Response) -> JSONResponse:
80		"""Create a new maintenance window."""
81	1	data = get_json_or_400(request, self.controller.loop)
82	1	if not isinstance(data, dict) or not data:
83		raise HTTPException(400, detail=f"Invalid json body value: {data}")
84
85	1	if 'status' in data:
86	1	raise HTTPException(
87		400, detail='Setting a maintenance status is not allowed'
88		)
89		# if 'id' in data:
90		# raise HTTPException(
91		# 400, detail='Setting a maintenance id is not allowed'
92		# )
93	1	try:
94	1	maintenance = MW.parse_obj(data)
95	1	force = data.get('force', False)
96	1	ignore_no_exists = data.get('ignore_no_exists')
97	1	if not ignore_no_exists:
98	1	self.validate_item_existence(maintenance)
99	1	self.scheduler.add(maintenance, force=force)
100	1	except ValidationError as err:
101	1	msg = error_msg(err.errors())
102	1	raise HTTPException(400, detail=msg) from err
103		except DuplicateKeyError as err:
104		raise HTTPException(
105		409,
106		detail=f'Window with id: {maintenance.id} already exists'
107		) from err
108		except OverlapError as err:
109		raise HTTPException(400, detail=f'{err}') from err
110		except ValueError as err:
111		raise HTTPException(400, detail=f'{err}') from err
112	1	return JSONResponse({'mw_id': maintenance.id}, status_code=201)
113
114	1	@rest('/v1/{mw_id}', methods=['PATCH'])
115	1	def update_mw(self, request: Request) -> JSONResponse:
116		"""Update a maintenance window."""
117	1	data = get_json_or_400(request, self.controller.loop)
118	1	if not isinstance(data, dict) or not data:
119		raise HTTPException(400, detail=f"Invalid json body value: {data}")
120
121	1	mw_id: MaintenanceID = request.path_params["mw_id"]
122	1	old_maintenance = self.scheduler.get_maintenance(mw_id)
123	1	if old_maintenance is None:
124	1	raise HTTPException(
125		404, detail=f'Maintenance with id {mw_id} not found'
126		)
127	1	if old_maintenance.status == Status.RUNNING:
128		raise HTTPException(
129		400, detail='Updating a running maintenance is not allowed'
130		)
131	1	if 'status' in data:
132	1	raise HTTPException(
133		400, detail='Updating a maintenance status is not allowed'
134		)
135	1	try:
136	1	new_maintenance = MW.parse_obj({old_maintenance.dict(), data})
137	1	except ValidationError as err:
138	1	msg = error_msg(err.errors())
139	1	raise HTTPException(400, detail=msg) from err
140	1	if new_maintenance.id != old_maintenance.id:
141		raise HTTPException(400, detail='Updated id must match old id')
142	1	self.scheduler.update(new_maintenance)
143	1	return JSONResponse({'response': f'Maintenance {mw_id} updated'})
144
145	1	@rest('/v1/{mw_id}', methods=['DELETE'])
146	1	def remove_mw(self, request: Request) -> JSONResponse:
147		"""Delete a maintenance window."""
148	1	mw_id: MaintenanceID = request.path_params["mw_id"]
149	1	maintenance = self.scheduler.get_maintenance(mw_id)
150	1	if maintenance is None:
151	1	raise HTTPException(
152		404, detail=f'Maintenance with id {mw_id} not found'
153		)
154	1	if maintenance.status == Status.RUNNING:
155	1	raise HTTPException(
156		400, detail='Deleting a running maintenance is not allowed'
157		)
158	1	self.scheduler.remove(mw_id)
159	1	return JSONResponse({'response': f'Maintenance with id {mw_id} '
160		f'successfully removed'})
161
162	1	@rest('/v1/{mw_id}/end', methods=['PATCH'])
163	1	def end_mw(self, request: Request) -> JSONResponse:
164		"""Finish a maintenance window right now."""
165	1	mw_id: MaintenanceID = request.path_params["mw_id"]
166	1	maintenance = self.scheduler.get_maintenance(mw_id)
167	1	if maintenance is None:
168	1	raise HTTPException(
169		404, detail=f'Maintenance with id {mw_id} not found'
170		)
171	1	if maintenance.status == Status.PENDING:
172	1	raise HTTPException(
173		400, detail=f'Maintenance window {mw_id} has not yet started'
174		)
175	1	if maintenance.status == Status.FINISHED:
176	1	raise HTTPException(
177		400, detail=f'Maintenance window {mw_id} has already finished'
178		)
179	1	self.scheduler.end_maintenance_early(mw_id)
180	1	return JSONResponse({'response': f'Maintenance window {mw_id} '
181		f'finished'})
182
183	1	@rest('/v1/{mw_id}/extend', methods=['PATCH'])
184	1	def extend_mw(self, request: Request) -> JSONResponse:
185		"""Extend a running maintenance window."""
186	1	mw_id: MaintenanceID = request.path_params["mw_id"]
187	1	data = get_json_or_400(request, self.controller.loop)
188	1	if not isinstance(data, dict):
189		raise HTTPException(400, detail=f"Invalid json body value: {data}")
190
191	1	maintenance = self.scheduler.get_maintenance(mw_id)
192	1	if maintenance is None:
193	1	raise HTTPException(
194		404, detail=f'Maintenance with id {mw_id} not found'
195		)
196	1	if 'minutes' not in data:
197	1	raise HTTPException(
198		400,
199		detail='Minutes of extension must be sent'
200		)
201	1	if maintenance.status == Status.PENDING:
202	1	raise HTTPException(
203		400,
204		detail=f'Maintenance window {mw_id} has not yet started'
205		)
206	1	if maintenance.status == Status.FINISHED:
207	1	raise HTTPException(
208		400,
209		detail=f'Maintenance window {mw_id} has already finished'
210		)
211	1	try:
212	1	maintenance_end = maintenance.end + \
213		timedelta(minutes=data['minutes'])
214	1	new_maintenance = maintenance.copy(
215		update={'end': maintenance_end}
216		)
217	1	except TypeError as exc:
218	1	raise HTTPException(
219		400,
220		detail='Minutes of extension must be integer') from exc
221
222	1	self.scheduler.update(new_maintenance)
223	1	return JSONResponse({'response': f'Maintenance {mw_id} extended'})
224
225	1	def validate_item_existence(self, window: MW):
226		"""Validate that all items in a maintenance window exist."""
227	1	non_existant_switches = list(
228		filter(
229		lambda switch_id:
230		self.controller.switches.get(switch_id)
231		is None,
232		window.switches
233		)
234		)
235	1	non_existant_interfaces = list(
236		filter(
237		lambda interface_id:
238		self.controller.get_interface_by_id(interface_id)
239		is None,
240		window.interfaces
241		)
242		)
243	1	non_existant_links = list(
244		filter(
245		lambda interface_id:
246		self.controller.napps[('kytos', 'topology')]
247		.links.get(interface_id)
248		is None,
249		window.interfaces
250		)
251		)
252
253	1	if (
254		non_existant_switches
255		or non_existant_interfaces
256		or non_existant_links
257		):
258		items = {
259		'switches': non_existant_switches,
260		'interfaces': non_existant_interfaces,
261		'links': non_existant_links,
262		}
263		raise HTTPException(
264		400,
265		f"Window contains non-existant items: {items}")
266

kytos-ng / maintenance

Push — master ( 0d5b5f...2f9122 )

build.main.Main.create_mw() C

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like