build.main.Main._validate_payload() - Code Metrics - Inspection of "Updated tox.ini and removed `tests_require` and `s..." - kytos-ng/pathfinder - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#22)

by Vinicius

created 2022-06-14 14:23 UTC

build.main.Main._validate_payload() D

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	62
Code Lines	42

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	22
CRAP Score	19.3771

Importance

Changes

Metric	Value
cc	12
eloc	42
nop	2
dl	0
loc	62
rs	4.8
c	0
b	0
f	0
ccs	22
cts	35
cp	0.6286
crap	19.3771

How to fix Long Method Complexity

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

from threading import Lock

from flask import jsonify, request
from kytos.core import KytosNApp, log, rest
from kytos.core.helpers import listen_to
from napps.kytos.pathfinder.graph import KytosGraph
# pylint: disable=import-error,no-self-use
from werkzeug.exceptions import BadRequest


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

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

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

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

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

    def _filter_paths(self, paths, desired, undesired):
        """
        Apply filters to the paths list.

        Make sure that each path in the list has all the desired links and none
        of the undesired ones.
        """
        filtered_paths = []

        if desired:

            for link_id in desired:
                try:
                    endpoint_a = self._topology.links[link_id].endpoint_a.id
                    endpoint_b = self._topology.links[link_id].endpoint_b.id
                except KeyError:
                    return []

                for path in paths:
                    head = path["hops"][:-1]
                    tail = path["hops"][1:]
                    if ((endpoint_a, endpoint_b) in zip(head, tail)) or (
                        (endpoint_b, endpoint_a) in zip(head, tail)
                    ):
                        filtered_paths.append(path)
        else:
            filtered_paths = paths

        if undesired:

            for link_id in undesired:
                try:
                    endpoint_a = self._topology.links[link_id].endpoint_a.id
                    endpoint_b = self._topology.links[link_id].endpoint_b.id
                except KeyError:
                    continue

                for path in paths:
                    head = path["hops"][:-1]
                    tail = path["hops"][1:]
                    if ((endpoint_a, endpoint_b) in zip(head, tail)) or (
                        (endpoint_b, endpoint_a) in zip(head, tail)
                    ):

                        filtered_paths.remove(path)

        return filtered_paths

    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 _validate_payload(self, data):
        """Validate shortest_path v2/ POST endpoint."""
        if data.get("desired_links"):
            if not isinstance(data["desired_links"], list):
                raise BadRequest(
                    f"TypeError: desired_links is supposed to be a list."
                    f" type: {type(data['desired_links'])}"
                )

        if data.get("undesired_links"):
            if not isinstance(data["undesired_links"], list):
                raise BadRequest(
                    f"TypeError: undesired_links is supposed to be a list."
                    f" type: {type(data['undesired_links'])}"
                )

        parameter = data.get("parameter")
        spf_attr = data.get("spf_attribute")
        if not spf_attr:
            spf_attr = parameter or "hop"
        data["spf_attribute"] = spf_attr

        if spf_attr not in self.graph.spf_edge_data_cbs:
            raise BadRequest(
                "Invalid 'spf_attribute'. Valid values: "
                f"{', '.join(self.graph.spf_edge_data_cbs.keys())}"
            )

        try:
            data["spf_max_paths"] = max(int(data.get("spf_max_paths", 2)), 1)
        except (TypeError, ValueError):
            raise BadRequest(
                f"spf_max_paths {data.get('spf_max_pahts')} must be an int"
            )

        spf_max_path_cost = data.get("spf_max_path_cost")
        if spf_max_path_cost:
            try:
                spf_max_path_cost = max(int(spf_max_path_cost), 1)
                data["spf_max_path_cost"] = spf_max_path_cost
            except (TypeError, ValueError):
                raise BadRequest(
                    f"spf_max_path_cost {data.get('spf_max_path_cost')} must"
                    " be an int"
                )

        data["mandatory_metrics"] = data.get("mandatory_metrics", {})
        data["flexible_metrics"] = data.get("flexible_metrics", {})

        try:
            minimum_hits = data.get("minimum_flexible_hits")
            if minimum_hits:
                minimum_hits = min(
                    len(data["flexible_metrics"]), max(0, int(minimum_hits))
                )
            data["minimum_flexible_hits"] = minimum_hits
        except (TypeError, ValueError):
            raise BadRequest(
                f"minimum_hits {data.get('minimum_flexible_hits')} must be an int"
            )

        return data

    @rest("v2/", methods=["POST"])
    def shortest_path(self):
        """Calculate the best path between the source and destination."""
        data = request.get_json()
        data = self._validate_payload(data)

        desired = data.get("desired_links")
        undesired = data.get("undesired_links")

        spf_attr = data.get("spf_attribute")
        spf_max_paths = data.get("spf_max_paths")
        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:
                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,
                        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,
                    )

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

        paths = self._filter_paths(paths, desired, undesired)
        paths = self._filter_paths_le_cost(paths, max_cost=spf_max_path_cost)
        log.debug(f"Filtered paths: {paths}")
        return jsonify({"paths": paths})

    @listen_to("kytos.topology.updated")
    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:
            self._topology = topology
            self.graph.update_topology(topology)
        log.debug("Topology graph updated.")

    @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."""
        link = event.content["link"]
        with self._lock:
            self.graph.update_link_metadata(link)
        metadata = event.content["metadata"]
        log.debug(f"Topology graph updated link id: {link.id} metadata: {metadata}")


1			"""Main module of kytos/pathfinder Kytos Network Application."""
2
3	1		from threading import Lock
4
5	1		from flask import jsonify, request
6	1		from kytos.core import KytosNApp, log, rest
7	1		from kytos.core.helpers import listen_to
8	1		from napps.kytos.pathfinder.graph import KytosGraph
9			# pylint: disable=import-error,no-self-use
10	1		from werkzeug.exceptions import BadRequest
11
12
13	1		class Main(KytosNApp):
14			"""
15			Main class of kytos/pathfinder NApp.
16
17			This class is the entry point for this napp.
18			"""
19
20	1		def setup(self):
21			"""Create a graph to handle the nodes and edges."""
22	1		self.graph = KytosGraph()
23	1		self._topology = None
24	1		self._lock = Lock()
25
26	1		def execute(self):
27			"""Do nothing."""
28
29	1		def shutdown(self):
30			"""Shutdown the napp."""
31
32	1		def _filter_paths(self, paths, desired, undesired):
33			"""
34			Apply filters to the paths list.
35
36			Make sure that each path in the list has all the desired links and none
37			of the undesired ones.
38			"""
39	1		filtered_paths = []
40
41	1	View Code Duplication	if desired:
			0 ignored issues – show Duplication introduced 2021-11-30 20:18 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
42	1		for link_id in desired:
43	1		try:
44	1		endpoint_a = self._topology.links[link_id].endpoint_a.id
45	1		endpoint_b = self._topology.links[link_id].endpoint_b.id
46			except KeyError:
47			return []
48
49	1		for path in paths:
50	1		head = path["hops"][:-1]
51	1		tail = path["hops"][1:]
52	1		if ((endpoint_a, endpoint_b) in zip(head, tail)) or (
53			(endpoint_b, endpoint_a) in zip(head, tail)
54			):
55	1		filtered_paths.append(path)
56			else:
57	1		filtered_paths = paths
58
59	1	View Code Duplication	if undesired:
			0 ignored issues – show Duplication introduced 2021-11-30 20:18 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
60	1		for link_id in undesired:
61	1		try:
62	1		endpoint_a = self._topology.links[link_id].endpoint_a.id
63	1		endpoint_b = self._topology.links[link_id].endpoint_b.id
64			except KeyError:
65			continue
66
67	1		for path in paths:
68	1		head = path["hops"][:-1]
69	1		tail = path["hops"][1:]
70	1		if ((endpoint_a, endpoint_b) in zip(head, tail)) or (
71			(endpoint_b, endpoint_a) in zip(head, tail)
72			):
73
74	1		filtered_paths.remove(path)
75
76	1		return filtered_paths
77
78	1		def _filter_paths_le_cost(self, paths, max_cost):
79			"""Filter by paths where the cost is le <= max_cost."""
80	1		if not max_cost:
81	1		return paths
82	1		return [path for path in paths if path["cost"] <= max_cost]
83
84	1		def _validate_payload(self, data):
85			"""Validate shortest_path v2/ POST endpoint."""
86	1		if data.get("desired_links"):
87	1		if not isinstance(data["desired_links"], list):
88			raise BadRequest(
89			f"TypeError: desired_links is supposed to be a list."
90			f" type: {type(data['desired_links'])}"
91			)
92
93	1		if data.get("undesired_links"):
94			if not isinstance(data["undesired_links"], list):
95			raise BadRequest(
96			f"TypeError: undesired_links is supposed to be a list."
97			f" type: {type(data['undesired_links'])}"
98			)
99
100	1		parameter = data.get("parameter")
101	1		spf_attr = data.get("spf_attribute")
102	1		if not spf_attr:
103	1		spf_attr = parameter or "hop"
104	1		data["spf_attribute"] = spf_attr
105
106	1		if spf_attr not in self.graph.spf_edge_data_cbs:
107			raise BadRequest(
108			"Invalid 'spf_attribute'. Valid values: "
109			f"{', '.join(self.graph.spf_edge_data_cbs.keys())}"
110			)
111
112	1		try:
113	1		data["spf_max_paths"] = max(int(data.get("spf_max_paths", 2)), 1)
114			except (TypeError, ValueError):
115			raise BadRequest(
116			f"spf_max_paths {data.get('spf_max_pahts')} must be an int"
117			)
118
119	1		spf_max_path_cost = data.get("spf_max_path_cost")
120	1		if spf_max_path_cost:
121			try:
122			spf_max_path_cost = max(int(spf_max_path_cost), 1)
123			data["spf_max_path_cost"] = spf_max_path_cost
124			except (TypeError, ValueError):
125			raise BadRequest(
126			f"spf_max_path_cost {data.get('spf_max_path_cost')} must"
127			" be an int"
128			)
129
130	1		data["mandatory_metrics"] = data.get("mandatory_metrics", {})
131	1		data["flexible_metrics"] = data.get("flexible_metrics", {})
132
133	1		try:
134	1		minimum_hits = data.get("minimum_flexible_hits")
135	1		if minimum_hits:
136	1		minimum_hits = min(
137			len(data["flexible_metrics"]), max(0, int(minimum_hits))
138			)
139	1		data["minimum_flexible_hits"] = minimum_hits
140			except (TypeError, ValueError):
141			raise BadRequest(
142			f"minimum_hits {data.get('minimum_flexible_hits')} must be an int"
143			)
144
145	1		return data
146
147	1		@rest("v2/", methods=["POST"])
148	1		def shortest_path(self):
149			"""Calculate the best path between the source and destination."""
150	1		data = request.get_json()
151	1		data = self._validate_payload(data)
152
153	1		desired = data.get("desired_links")
154	1		undesired = data.get("undesired_links")
155
156	1		spf_attr = data.get("spf_attribute")
157	1		spf_max_paths = data.get("spf_max_paths")
158	1		spf_max_path_cost = data.get("spf_max_path_cost")
159	1		mandatory_metrics = data.get("mandatory_metrics")
160	1		flexible_metrics = data.get("flexible_metrics")
161	1		minimum_hits = data.get("minimum_flexible_hits")
162	1		log.debug(f"POST v2/ payload data: {data}")
163
164	1		try:
165	1		with self._lock:
166	1		if any([mandatory_metrics, flexible_metrics]):
167	1		paths = self.graph.constrained_k_shortest_paths(
168			data["source"],
169			data["destination"],
170			weight=self.graph.spf_edge_data_cbs[spf_attr],
171			k=spf_max_paths,
172			minimum_hits=minimum_hits,
173			mandatory_metrics=mandatory_metrics,
174			flexible_metrics=flexible_metrics,
175			)
176			else:
177	1		paths = self.graph.k_shortest_paths(
178			data["source"],
179			data["destination"],
180			weight=self.graph.spf_edge_data_cbs[spf_attr],
181			k=spf_max_paths,
182			)
183
184	1		paths = self.graph.path_cost_builder(
185			paths,
186			weight=spf_attr,
187			)
188	1		log.debug(f"Found paths: {paths}")
189	1		except TypeError as err:
190	1		raise BadRequest(str(err))
191
192	1		paths = self._filter_paths(paths, desired, undesired)
193	1		paths = self._filter_paths_le_cost(paths, max_cost=spf_max_path_cost)
194	1		log.debug(f"Filtered paths: {paths}")
195	1		return jsonify({"paths": paths})
196
197	1		@listen_to("kytos.topology.updated")
198	1		def on_topology_updated(self, event):
199			"""Update the graph when the network topology is updated."""
200			self.update_topology(event)
201
202	1		def update_topology(self, event):
203			"""Update the graph when the network topology is updated."""
204	1		if "topology" not in event.content:
205	1		return
206	1		topology = event.content["topology"]
207	1		with self._lock:
208	1		self._topology = topology
209	1		self.graph.update_topology(topology)
210	1		log.debug("Topology graph updated.")
211
212	1		@listen_to("kytos/topology.links.metadata.(added\|removed)")
213	1		def on_links_metadata_changed(self, event):
214			"""Update the graph when links' metadata are added or removed."""
215			link = event.content["link"]
216			with self._lock:
217			self.graph.update_link_metadata(link)
218			metadata = event.content["metadata"]
219			log.debug(f"Topology graph updated link id: {link.id} metadata: {metadata}")
220

kytos-ng / pathfinder

Pull Request — master (#22)

build.main.Main._validate_payload() D

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like