build.main.Main._validate_payload() - Code Metrics - Inspection of "fix: removed support for `desired_links` and fixed..." - kytos-ng/pathfinder - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#47)

by Vinicius

created 2023-05-10 23:22 UTC

build.main.Main._validate_payload() F

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	78
Code Lines	56

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	24
CRAP Score	30.5635

Importance

Changes

Metric	Value
cc	16
eloc	56
nop	2
dl	0
loc	78
ccs	24
cts	39
cp	0.6153
crap	30.5635
rs	2.4
c	0
b	0
f	0

How to fix Long Method Complexity

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

import pathlib
from threading import Lock
from typing import Generator

from kytos.core import KytosEvent, KytosNApp, log, rest
from kytos.core.helpers import listen_to, load_spec, validate_openapi
from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
                                 get_json_or_400)
from napps.kytos.pathfinder.graph import KytosGraph


class Main(KytosNApp):
    """
    Main class of kytos/pathfinder NApp.

    This class is the entry point for this napp.
    """

    spec = load_spec(pathlib.Path(__file__).parent / "openapi.yml")

    def setup(self):
        """Create a graph to handle the nodes and edges."""
        self.graph = KytosGraph()
        self._topology = None
        self._lock = Lock()
        self._topology_updated_at = None
        self._links_updated_at = {}

    def execute(self):
        """Do nothing."""

    def shutdown(self):
        """Shutdown the napp."""

    def _filter_paths_le_cost(self, paths, max_cost):
        """Filter by paths where the cost is le <= max_cost."""
        if not max_cost:
            return paths
        return [path for path in paths if path["cost"] <= max_cost]

    def _non_excluded_edges(self, links: list[str]) -> list[tuple[str, str]]:
        """Exlude undesired links. It'll return the remaning edges."""

        endpoints: list[tuple[str, str]] = []
        if not self._topology:
            return endpoints
        endpoint_ids = self._map_endpoints_from_link_ids(links)
        for edge in self.graph.graph.edges:
            if edge not in endpoint_ids:
                endpoints.append(edge)
        return endpoints

    def _map_endpoints_from_link_ids(self, link_ids: list[str]) -> dict:
        """Map endpoints from link ids."""
        endpoints = {}
        for link_id in link_ids:
            try:
                link = self._topology.links[link_id]
                endpoint_a, endpoint_b = link.endpoint_a, link.endpoint_b
                endpoints[(endpoint_a.id, endpoint_b.id)] = link
            except KeyError:
                pass
        return endpoints

    def _find_all_link_ids(
        self, paths: list[dict], link_ids: list[str]
    ) -> Generator[int, None, None]:
        """Find indexes of the paths that contain all link ids."""
        endpoints_links = self._map_endpoints_from_link_ids(link_ids)
        if not endpoints_links:
            return None
        endpoint_keys = set(endpoints_links.keys())
        for idx, path in enumerate(paths):
            head, tail, found_endpoints = path["hops"][:-1], path["hops"][1:], set()
            for endpoint_a, endpoint_b in zip(head, tail):
                if (endpoint_a, endpoint_b) in endpoints_links:
                    found_endpoints.add((endpoint_a, endpoint_b))
                if (endpoint_b, endpoint_a) in endpoints_links:
                    found_endpoints.add((endpoint_b, endpoint_a))
            if found_endpoints == endpoint_keys:
                yield idx
        return None

    @rest("v3/", methods=["POST"])
    @validate_openapi(spec)
    def shortest_path(self, request: Request) -> JSONResponse:
        """Calculate the best path between the source and destination."""
        data = get_json_or_400(request, self.controller.loop)
        if not isinstance(data, dict):
            raise HTTPException(400, detail=f"Invalid body value: {data}")

        undesired = data.get("undesired_links", [])
        spf_attr = data.get("spf_attribute", "hop")
        spf_max_paths = data.get("spf_max_paths", 2)
        spf_max_path_cost = data.get("spf_max_path_cost")
        mandatory_metrics = data.get("mandatory_metrics", {})
        flexible_metrics = data.get("flexible_metrics", {})
        minimum_hits = data.get("minimum_flexible_hits")
        log.debug(f"POST v2/ payload data: {data}")

        try:
            with self._lock:
                graph = self.graph.graph
                if undesired:
                    non_excluded_edges = self._non_excluded_edges(undesired)
                    graph = graph.edge_subgraph(non_excluded_edges)

                if any([mandatory_metrics, flexible_metrics]):
                    paths = self.graph.constrained_k_shortest_paths(
                        data["source"],
                        data["destination"],
                        weight=self.graph.spf_edge_data_cbs[spf_attr],
                        k=spf_max_paths,
                        graph=graph,
                        minimum_hits=minimum_hits,
                        mandatory_metrics=mandatory_metrics,
                        flexible_metrics=flexible_metrics,
                    )
                else:
                    paths = self.graph.k_shortest_paths(
                        data["source"],
                        data["destination"],
                        weight=self.graph.spf_edge_data_cbs[spf_attr],
                        k=spf_max_paths,
                        graph=graph,
                    )

                paths = self.graph.path_cost_builder(
                    paths,
                    weight=spf_attr,
                )
            log.debug(f"Found paths: {paths}")
        except TypeError as err:
            raise HTTPException(400, str(err))

        paths = self._filter_paths_le_cost(paths, max_cost=spf_max_path_cost)
        log.debug(f"Filtered paths: {paths}")
        return JSONResponse({"paths": paths})

    @listen_to(
        "kytos.topology.updated",
        "kytos/topology.current",
        "kytos/topology.topology_loaded",
    )
    def on_topology_updated(self, event):
        """Update the graph when the network topology is updated."""
        self.update_topology(event)

    def update_topology(self, event):
        """Update the graph when the network topology is updated."""
        if "topology" not in event.content:
            return
        topology = event.content["topology"]
        with self._lock:
            if (
                self._topology_updated_at
                and self._topology_updated_at > event.timestamp
            ):
                return
            self._topology = topology
            self._topology_updated_at = event.timestamp
            self.graph.update_topology(topology)
        switches = list(topology.switches.keys())
        links = list(topology.links.keys())
        log.debug(f"Topology graph updated with switches: {switches}, links: {links}.")

    def update_links_metadata_changed(self, event) -> None:
        """Update the graph when links' metadata are added or removed."""
        link = event.content["link"]
        try:
            with self._lock:
                if (
                    link.id in self._links_updated_at
                    and self._links_updated_at[link.id] > event.timestamp
                ):
                    return
                self.graph.update_link_metadata(link)
                self._links_updated_at[link.id] = event.timestamp
            metadata = event.content["metadata"]
            log.debug(f"Topology graph updated link id: {link.id} metadata: {metadata}")
        except KeyError as exc:
            log.warning(
                f"Unexpected KeyError {str(exc)} on event {event}."
                " pathfinder will reconciliate the topology"
            )
            self.controller.buffers.app.put(KytosEvent(name="kytos/topology.get"))

    @listen_to("kytos/topology.links.metadata.(added|removed)")
    def on_links_metadata_changed(self, event):
        """Update the graph when links' metadata are added or removed."""
        self.update_links_metadata_changed(event)


1		"""Main module of kytos/pathfinder Kytos Network Application."""
2
3	1	import pathlib
4	1	from threading import Lock
5	1	from typing import Generator
6
7	1	from kytos.core import KytosEvent, KytosNApp, log, rest
8	1	from kytos.core.helpers import listen_to, load_spec, validate_openapi
9	1	from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
10		get_json_or_400)
11	1	from napps.kytos.pathfinder.graph import KytosGraph
12
13
14	1	class Main(KytosNApp):
15		"""
16		Main class of kytos/pathfinder NApp.
17
18		This class is the entry point for this napp.
19		"""
20
21	1	spec = load_spec(pathlib.Path(__file__).parent / "openapi.yml")
22
23	1	def setup(self):
24		"""Create a graph to handle the nodes and edges."""
25	1	self.graph = KytosGraph()
26	1	self._topology = None
27	1	self._lock = Lock()
28	1	self._topology_updated_at = None
29	1	self._links_updated_at = {}
30
31	1	def execute(self):
32		"""Do nothing."""
33
34	1	def shutdown(self):
35		"""Shutdown the napp."""
36
37	1	def _filter_paths_le_cost(self, paths, max_cost):
38		"""Filter by paths where the cost is le <= max_cost."""
39	1	if not max_cost:
40	1	return paths
41	1	return [path for path in paths if path["cost"] <= max_cost]
42
43	1	def _non_excluded_edges(self, links: list[str]) -> list[tuple[str, str]]:
44		"""Exlude undesired links. It'll return the remaning edges."""
45
46	1	endpoints: list[tuple[str, str]] = []
47	1	if not self._topology:
48	1	return endpoints
49	1	endpoint_ids = self._map_endpoints_from_link_ids(links)
50	1	for edge in self.graph.graph.edges:
51	1	if edge not in endpoint_ids:
52	1	endpoints.append(edge)
53	1	return endpoints
54
55	1	def _map_endpoints_from_link_ids(self, link_ids: list[str]) -> dict:
56		"""Map endpoints from link ids."""
57	1	endpoints = {}
58	1	for link_id in link_ids:
59	1	try:
60	1	link = self._topology.links[link_id]
61	1	endpoint_a, endpoint_b = link.endpoint_a, link.endpoint_b
62	1	endpoints[(endpoint_a.id, endpoint_b.id)] = link
63		except KeyError:
64		pass
65	1	return endpoints
66
67	1	def _find_all_link_ids(
68		self, paths: list[dict], link_ids: list[str]
69		) -> Generator[int, None, None]:
70		"""Find indexes of the paths that contain all link ids."""
71		endpoints_links = self._map_endpoints_from_link_ids(link_ids)
72		if not endpoints_links:
73		return None
74		endpoint_keys = set(endpoints_links.keys())
75		for idx, path in enumerate(paths):
76		head, tail, found_endpoints = path["hops"][:-1], path["hops"][1:], set()
77		for endpoint_a, endpoint_b in zip(head, tail):
78		if (endpoint_a, endpoint_b) in endpoints_links:
79		found_endpoints.add((endpoint_a, endpoint_b))
80		if (endpoint_b, endpoint_a) in endpoints_links:
81		found_endpoints.add((endpoint_b, endpoint_a))
82		if found_endpoints == endpoint_keys:
83		yield idx
84		return None
85
86	1	@rest("v3/", methods=["POST"])
87	1	@validate_openapi(spec)
88	1	def shortest_path(self, request: Request) -> JSONResponse:
89		"""Calculate the best path between the source and destination."""
90	1	data = get_json_or_400(request, self.controller.loop)
91	1	if not isinstance(data, dict):
92		raise HTTPException(400, detail=f"Invalid body value: {data}")
93
94	1	undesired = data.get("undesired_links", [])
95	1	spf_attr = data.get("spf_attribute", "hop")
96	1	spf_max_paths = data.get("spf_max_paths", 2)
97	1	spf_max_path_cost = data.get("spf_max_path_cost")
98	1	mandatory_metrics = data.get("mandatory_metrics", {})
99	1	flexible_metrics = data.get("flexible_metrics", {})
100	1	minimum_hits = data.get("minimum_flexible_hits")
101	1	log.debug(f"POST v2/ payload data: {data}")
102
103	1	try:
104	1	with self._lock:
105	1	graph = self.graph.graph
106	1	if undesired:
107		non_excluded_edges = self._non_excluded_edges(undesired)
108		graph = graph.edge_subgraph(non_excluded_edges)
109
110	1	if any([mandatory_metrics, flexible_metrics]):
111	1	paths = self.graph.constrained_k_shortest_paths(
112		data["source"],
113		data["destination"],
114		weight=self.graph.spf_edge_data_cbs[spf_attr],
115		k=spf_max_paths,
116		graph=graph,
117		minimum_hits=minimum_hits,
118		mandatory_metrics=mandatory_metrics,
119		flexible_metrics=flexible_metrics,
120		)
121		else:
122	1	paths = self.graph.k_shortest_paths(
123		data["source"],
124		data["destination"],
125		weight=self.graph.spf_edge_data_cbs[spf_attr],
126		k=spf_max_paths,
127		graph=graph,
128		)
129
130	1	paths = self.graph.path_cost_builder(
131		paths,
132		weight=spf_attr,
133		)
134	1	log.debug(f"Found paths: {paths}")
135	1	except TypeError as err:
136	1	raise HTTPException(400, str(err))
137
138	1	paths = self._filter_paths_le_cost(paths, max_cost=spf_max_path_cost)
139	1	log.debug(f"Filtered paths: {paths}")
140	1	return JSONResponse({"paths": paths})
141
142	1	@listen_to(
143		"kytos.topology.updated",
144		"kytos/topology.current",
145		"kytos/topology.topology_loaded",
146		)
147	1	def on_topology_updated(self, event):
148		"""Update the graph when the network topology is updated."""
149		self.update_topology(event)
150
151	1	def update_topology(self, event):
152		"""Update the graph when the network topology is updated."""
153	1	if "topology" not in event.content:
154	1	return
155	1	topology = event.content["topology"]
156	1	with self._lock:
157	1	if (
158		self._topology_updated_at
159		and self._topology_updated_at > event.timestamp
160		):
161	1	return
162	1	self._topology = topology
163	1	self._topology_updated_at = event.timestamp
164	1	self.graph.update_topology(topology)
165	1	switches = list(topology.switches.keys())
166	1	links = list(topology.links.keys())
167	1	log.debug(f"Topology graph updated with switches: {switches}, links: {links}.")
168
169	1	def update_links_metadata_changed(self, event) -> None:
170		"""Update the graph when links' metadata are added or removed."""
171	1	link = event.content["link"]
172	1	try:
173	1	with self._lock:
174	1	if (
175		link.id in self._links_updated_at
176		and self._links_updated_at[link.id] > event.timestamp
177		):
178	1	return
179	1	self.graph.update_link_metadata(link)
180	1	self._links_updated_at[link.id] = event.timestamp
181	1	metadata = event.content["metadata"]
182	1	log.debug(f"Topology graph updated link id: {link.id} metadata: {metadata}")
183	1	except KeyError as exc:
184	1	log.warning(
185		f"Unexpected KeyError {str(exc)} on event {event}."
186		" pathfinder will reconciliate the topology"
187		)
188	1	self.controller.buffers.app.put(KytosEvent(name="kytos/topology.get"))
189
190	1	@listen_to("kytos/topology.links.metadata.(added\|removed)")
191	1	def on_links_metadata_changed(self, event):
192		"""Update the graph when links' metadata are added or removed."""
193		self.update_links_metadata_changed(event)
194

kytos-ng / pathfinder

Pull Request — master (#47)

build.main.Main._validate_payload() F

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like