build.main.Main.provision_int() - Code Metrics - Inspection of "refactor: minimized IO, bulk updates, general enh..." - kytos-ng/telemetry_int - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#35)

by Vinicius

created 2023-08-26 20:49 UTC

build.main.Main.provision_int() C

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	53
Code Lines	26

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	110

Importance

Changes

Metric	Value
cc	10
eloc	26
nop	2
dl	0
loc	53
ccs	0
cts	22
cp	0
crap	110
rs	5.9999
c	0
b	0
f	0

How to fix Long Method Complexity

"""Main module of kytos/telemetry Network Application.

Napp to deploy In-band Network Telemetry over Ethernet Virtual Circuits

"""

from napps.kytos.telemetry_int import utils
from tenacity import RetryError

from kytos.core import KytosNApp, log, rest
from kytos.core.rest_api import HTTPException, JSONResponse, Request, aget_json_or_400

from .exceptions import (
    EVCHasINT,
    EVCHasNoINT,
    EVCNotFound,
    FlowsNotFound,
    ProxyPortNotFound,
    ProxyPortStatusNotUP,
    UnrecoverableError,
)
from .kytos_api_helper import get_evc, get_evcs
from .managers.int import INTManager

# pylint: disable=fixme


class Main(KytosNApp):
    """Main class of kytos/telemetry 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.int_manager = INTManager(self.controller)

    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.
        """

    @rest("v1/evc/enable", methods=["POST"])

    async def enable_telemetry(self, request: Request) -> JSONResponse:
        """REST to enable INT flows on EVCs.

        If a list of evc_ids is empty, it'll enable on non-INT EVCs.
        """

        try:
            content = await aget_json_or_400(request)
            evc_ids = content["evc_ids"]
            force = content.get("force", False)
            if not isinstance(force, bool):
                raise TypeError(f"'force' wrong type: {type(force)} expected bool")
        except (TypeError, KeyError):
            raise HTTPException(400, detail=f"Invalid payload: {content}")

        try:
            evcs = await get_evcs() if len(evc_ids) != 1 else await get_evc(evc_ids[0])
        except RetryError as exc:
            exc_error = str(exc.last_attempt.exception())
            log.error(exc_error)
            raise HTTPException(503, detail=exc_error)

        if evc_ids:
            evcs = {evc_id: evcs.get(evc_id, {}) for evc_id in evc_ids}
        else:
            evcs = {k: v for k, v in evcs.items() if not utils.has_int_enabled(v)}
            if not evcs:
                # There's no non-INT EVCs to get enabled.
                return JSONResponse({})

        try:
            await self.int_manager.enable_int(evcs, force)
        except (EVCNotFound, FlowsNotFound, ProxyPortNotFound) as exc:
            raise HTTPException(404, detail=str(exc))
        except (EVCHasINT, ProxyPortStatusNotUP) as exc:
            raise HTTPException(400, detail=str(exc))
        except RetryError as exc:
            exc_error = str(exc.last_attempt.exception())
            log.error(exc_error)
            raise HTTPException(503, detail=exc_error)
        except UnrecoverableError as exc:
            exc_error = str(exc)
            log.error(exc_error)
            raise HTTPException(500, detail=exc_error)

        return JSONResponse({}, status_code=201)

    @rest("v1/evc/disable", methods=["POST"])

    async def disable_telemetry(self, request: Request) -> JSONResponse:
        """REST to disable/remove INT flows for an EVC_ID

        If a list of evc_ids is empty, it'll disable on all INT EVCs.
        """
        try:
            content = await aget_json_or_400(request)
            evc_ids = content["evc_ids"]
            force = content.get("force", False)
            if not isinstance(force, bool):
                raise TypeError(f"'force' wrong type: {type(force)} expected bool")
        except (TypeError, KeyError):
            raise HTTPException(400, detail=f"Invalid payload: {content}")

        try:
            evcs = await get_evcs() if len(evc_ids) != 1 else await get_evc(evc_ids[0])
        except RetryError as exc:
            exc_error = str(exc.last_attempt.exception())
            log.error(exc_error)
            raise HTTPException(503, detail=exc_error)

        if evc_ids:
            evcs = {evc_id: evcs.get(evc_id, {}) for evc_id in evc_ids}
        else:
            evcs = {k: v for k, v in evcs.items() if utils.has_int_enabled(v)}
            if not evcs:
                # There's no INT EVCs to get disabled.
                return JSONResponse({})

        try:
            await self.int_manager.disable_int(evcs, force)
        except EVCNotFound as exc:
            raise HTTPException(404, detail=str(exc))
        except EVCHasNoINT as exc:
            raise HTTPException(400, detail=str(exc))
        except RetryError as exc:
            exc_error = str(exc.last_attempt.exception())
            log.error(exc_error)
            raise HTTPException(503, detail=exc_error)
        except UnrecoverableError as exc:
            exc_error = str(exc)
            log.error(exc_error)
            raise HTTPException(500, detail=exc_error)

        return JSONResponse({})

    @rest("v1/evc")
    def get_evcs(self, _request: Request) -> JSONResponse:
        """REST to return the list of EVCs with INT enabled"""
        return JSONResponse(utils.get_evc_with_telemetry())

    @rest("v1/sync")
    def sync_flows(self, _request: Request) -> JSONResponse:
        """Endpoint to force the telemetry napp to search for INT flows and delete them
        accordingly to the evc metadata."""

        # TODO
        # for evc_id in get_evcs_ids():
        return JSONResponse("TBD")

    @rest("v1/evc/update")
    def update_evc(self, _request: Request) -> JSONResponse:
        """If an EVC changed from unidirectional to bidirectional telemetry,
        make the change."""
        return JSONResponse({})

    # Event-driven methods: future
    def listen_for_new_evcs(self):
        """Change newly created EVC to INT-enabled EVC based on the metadata field
        (future)"""
        pass

    def listen_for_evc_change(self):
        """Change newly created EVC to INT-enabled EVC based on the
        metadata field (future)"""
        pass

    def listen_for_path_changes(self):
        """Change EVC's new path to INT-enabled EVC based on the metadata field
        when there is a path change. (future)"""
        pass

    def listen_for_evcs_removed(self):
        """Remove all INT flows belonging the just removed EVC (future)"""
        pass

    def listen_for_topology_changes(self):
        """If the topology changes, make sure it is not the loop ports.
        If so, update proxy ports"""
        # TODO:
        # self.proxy_ports = create_proxy_ports(self.proxy_ports)
        pass


1		"""Main module of kytos/telemetry Network Application.
2
3		Napp to deploy In-band Network Telemetry over Ethernet Virtual Circuits
4
5		"""
6
7		from napps.kytos.telemetry_int import utils
8		from tenacity import RetryError
9
10		from kytos.core import KytosNApp, log, rest
11		from kytos.core.rest_api import HTTPException, JSONResponse, Request, aget_json_or_400
12
13		from .exceptions import (
14		EVCHasINT,
15		EVCHasNoINT,
16		EVCNotFound,
17		FlowsNotFound,
18		ProxyPortNotFound,
19		ProxyPortStatusNotUP,
20		UnrecoverableError,
21		)
22		from .kytos_api_helper import get_evc, get_evcs
23		from .managers.int import INTManager
24
25		# pylint: disable=fixme
26
27
28		class Main(KytosNApp):
29		"""Main class of kytos/telemetry NApp.
30
31		This class is the entry point for this NApp.
32		"""
33
34		def setup(self):
35		"""Replace the '__init__' method for the KytosNApp subclass.
36
37		The setup method is automatically called by the controller when your
38		application is loaded.
39
40		So, if you have any setup routine, insert it here.
41		"""
42
43		self.int_manager = INTManager(self.controller)
44
45		def execute(self):
46		"""Run after the setup method execution.
47
48		You can also use this method in loop mode if you add to the above setup
49		method a line like the following example:
50
51		self.execute_as_loop(30) # 30-second interval.
52		"""
53
54		def shutdown(self):
55		"""Run when your NApp is unloaded.
56
57		If you have some cleanup procedure, insert it here.
58		"""
59
60	View Code Duplication	@rest("v1/evc/enable", methods=["POST"])
		0 ignored issues – show Duplication introduced 2023-08-02 20:05 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
61		async def enable_telemetry(self, request: Request) -> JSONResponse:
62		"""REST to enable INT flows on EVCs.
63
64		If a list of evc_ids is empty, it'll enable on non-INT EVCs.
65		"""
66
67		try:
68		content = await aget_json_or_400(request)
69		evc_ids = content["evc_ids"]
70		force = content.get("force", False)
71		if not isinstance(force, bool):
72		raise TypeError(f"'force' wrong type: {type(force)} expected bool")
73		except (TypeError, KeyError):
74		raise HTTPException(400, detail=f"Invalid payload: {content}")
75
76		try:
77		evcs = await get_evcs() if len(evc_ids) != 1 else await get_evc(evc_ids[0])
78		except RetryError as exc:
79		exc_error = str(exc.last_attempt.exception())
80		log.error(exc_error)
81		raise HTTPException(503, detail=exc_error)
82
83		if evc_ids:
84		evcs = {evc_id: evcs.get(evc_id, {}) for evc_id in evc_ids}
85		else:
86		evcs = {k: v for k, v in evcs.items() if not utils.has_int_enabled(v)}
87		if not evcs:
88		# There's no non-INT EVCs to get enabled.
89		return JSONResponse({})
90
91		try:
92		await self.int_manager.enable_int(evcs, force)
93		except (EVCNotFound, FlowsNotFound, ProxyPortNotFound) as exc:
94		raise HTTPException(404, detail=str(exc))
95		except (EVCHasINT, ProxyPortStatusNotUP) as exc:
96		raise HTTPException(400, detail=str(exc))
97		except RetryError as exc:
98		exc_error = str(exc.last_attempt.exception())
99		log.error(exc_error)
100		raise HTTPException(503, detail=exc_error)
101		except UnrecoverableError as exc:
102		exc_error = str(exc)
103		log.error(exc_error)
104		raise HTTPException(500, detail=exc_error)
105
106		return JSONResponse({}, status_code=201)
107
108	View Code Duplication	@rest("v1/evc/disable", methods=["POST"])
		0 ignored issues – show Duplication introduced 2023-08-02 20:05 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
109		async def disable_telemetry(self, request: Request) -> JSONResponse:
110		"""REST to disable/remove INT flows for an EVC_ID
111
112		If a list of evc_ids is empty, it'll disable on all INT EVCs.
113		"""
114		try:
115		content = await aget_json_or_400(request)
116		evc_ids = content["evc_ids"]
117		force = content.get("force", False)
118		if not isinstance(force, bool):
119		raise TypeError(f"'force' wrong type: {type(force)} expected bool")
120		except (TypeError, KeyError):
121		raise HTTPException(400, detail=f"Invalid payload: {content}")
122
123		try:
124		evcs = await get_evcs() if len(evc_ids) != 1 else await get_evc(evc_ids[0])
125		except RetryError as exc:
126		exc_error = str(exc.last_attempt.exception())
127		log.error(exc_error)
128		raise HTTPException(503, detail=exc_error)
129
130		if evc_ids:
131		evcs = {evc_id: evcs.get(evc_id, {}) for evc_id in evc_ids}
132		else:
133		evcs = {k: v for k, v in evcs.items() if utils.has_int_enabled(v)}
134		if not evcs:
135		# There's no INT EVCs to get disabled.
136		return JSONResponse({})
137
138		try:
139		await self.int_manager.disable_int(evcs, force)
140		except EVCNotFound as exc:
141		raise HTTPException(404, detail=str(exc))
142		except EVCHasNoINT as exc:
143		raise HTTPException(400, detail=str(exc))
144		except RetryError as exc:
145		exc_error = str(exc.last_attempt.exception())
146		log.error(exc_error)
147		raise HTTPException(503, detail=exc_error)
148		except UnrecoverableError as exc:
149		exc_error = str(exc)
150		log.error(exc_error)
151		raise HTTPException(500, detail=exc_error)
152
153		return JSONResponse({})
154
155		@rest("v1/evc")
156		def get_evcs(self, _request: Request) -> JSONResponse:
157		"""REST to return the list of EVCs with INT enabled"""
158		return JSONResponse(utils.get_evc_with_telemetry())
159
160		@rest("v1/sync")
161		def sync_flows(self, _request: Request) -> JSONResponse:
162		"""Endpoint to force the telemetry napp to search for INT flows and delete them
163		accordingly to the evc metadata."""
164
165		# TODO
166		# for evc_id in get_evcs_ids():
167		return JSONResponse("TBD")
168
169		@rest("v1/evc/update")
170		def update_evc(self, _request: Request) -> JSONResponse:
171		"""If an EVC changed from unidirectional to bidirectional telemetry,
172		make the change."""
173		return JSONResponse({})
174
175		# Event-driven methods: future
176		def listen_for_new_evcs(self):
177		"""Change newly created EVC to INT-enabled EVC based on the metadata field
178		(future)"""
179		pass
180
181		def listen_for_evc_change(self):
182		"""Change newly created EVC to INT-enabled EVC based on the
183		metadata field (future)"""
184		pass
185
186		def listen_for_path_changes(self):
187		"""Change EVC's new path to INT-enabled EVC based on the metadata field
188		when there is a path change. (future)"""
189		pass
190
191		def listen_for_evcs_removed(self):
192		"""Remove all INT flows belonging the just removed EVC (future)"""
193		pass
194
195		def listen_for_topology_changes(self):
196		"""If the topology changes, make sure it is not the loop ports.
197		If so, update proxy ports"""
198		# TODO:
199		# self.proxy_ports = create_proxy_ports(self.proxy_ports)
200		pass
201

kytos-ng / telemetry_int

Pull Request — master (#35)

build.main.Main.provision_int() C

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like