build.main.Main.shutdown() - Code Metrics - Inspection of "chore: `@rest` endpoints are now run by `starlette..." - kytos-ng/maintenance - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#75)

by Vinicius

created 2023-04-26 22:04 UTC

build.main.Main.shutdown() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	6
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.125

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	6
ccs	1
cts	2
cp	0.5
crap	1.125
rs	10
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.models import MaintenanceDeployer, MaintenanceID
from napps.kytos.maintenance.models import MaintenanceWindow as MW
from napps.kytos.maintenance.models import OverlapError, Scheduler, Status
from pydantic import ValidationError

from kytos.core import KytosNApp, rest
# pylint: disable=unused-import
from kytos.core.interface import Interface
from kytos.core.link import Link
from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
                                 Response, get_json_or_400)
from kytos.core.switch import Switch

# pylint: enable=unused-import


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 = \
            MaintenanceDeployer.new_deployer(self.controller)

        # Switch.register_status_func(
        #     'maintenance_status',
        #     self.maintenance_deployer.dev_in_maintenance
        # )
        # Interface.register_status_func(
        #     'maintenance_status',
        #     self.maintenance_deployer.dev_in_maintenance
        # )
        # Link.register_status_func(
        #     'maintenance_status',
        #     self.maintenance_deployer.dev_in_maintenance
        # )
        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)
        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'
            )
        try:
            if data.get('id') == '':
                del data['id']
            maintenance = MW.parse_obj(data)
            force = data.get('force', False)
        except ValidationError as err:
            raise HTTPException(400,
                                detail=f'{err.errors()[0]["msg"]}') from err
        try:
            self.scheduler.add(maintenance, force=force)
        except OverlapError 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)
        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:
            detail = f'{err.errors()[0]["msg"]}'
            raise HTTPException(400, detail=detail) 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)
        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'})


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

kytos-ng / maintenance

Pull Request — master (#75)

build.main.Main.shutdown() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like