build.main

Complexity

Total Complexity

27

Size/Duplication

Total Lines	157
Duplicated Lines	19.75 %

Test Coverage

Coverage

92.73%

Importance

Changes

0

8 Methods

Rating	Name	Duplication	Size	Complexity
A	Main.shortest_path()	0	18	2
A	Main.shortest_constrained_path()	0	22	3
A	Main.shutdown()	0	2	1
A	Main.setup()	0	4	1
A	Main.execute()	0	2	1
D	Main._filter_paths()	31	43	13
A	Main.path_exact_delay()	0	21	2
A	Main.update_topology()	0	18	4

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

from flask import jsonify, request
from kytos.core import KytosNApp, log, rest
from kytos.core.helpers import listen_to
import traceback
# pylint: disable=import-error
# from napps.kytos.pathfinder.graph import KytosGraph
from graph import KytosGraph
# pylint: enable=import-error


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

    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:

This code seems to be duplicated in your project.

            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:

This code seems to be duplicated in your project.

            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

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

        desired = data.get('desired_links')
        undesired = data.get('undesired_links')
        parameter = data.get('parameter')

        paths = []
        for path in self.graph.shortest_paths(data['source'],
                                              data['destination'],
                                              parameter):

            paths.append({'hops': path})

        paths = self._filter_paths(paths, desired, undesired)
        return jsonify({'paths': paths})

    @rest('v2/path-exact-delay', methods=['POST'])
    def path_exact_delay(self):
        """
        Calculate the path with the exact delay
        between the source and destination.
        """
        data = request.get_json()

        source = data.get('source')
        destination = data.get('destination')
        delay = data.get('delay')

        graph_data = {}
        try:
            result = self.graph.exact_path(delay, source, destination)
        except Exception:
            return jsonify({"exception": str(traceback.format_exc())})
            
        graph_data["Exact Path Result"] = result

        return jsonify(graph_data)

    @rest('v2/best-constrained-paths', methods=['POST'])
    def shortest_constrained_path(self):
        """
        Get the set of shortest paths between the source and destination.
        """
        data = request.get_json()

        source = data.get('source')
        destination = data.get('destination')
        base_metrics = data.get('base_metrics', {})
        fle_metrics = data.get('flexible_metrics', {})
        minimum_hits = data.get('minimum_flexible_hits')
        try:
            paths = self.graph.constrained_flexible_paths(source, destination,
                                                          minimum_hits,
                                                          base=base_metrics,
                                                          flexible=fle_metrics)
            return jsonify(paths)
        except TypeError as err:
            return jsonify({"error": str(err)}), 400
        except Exception as err:
            return jsonify({"error": str(err)}), 500

    @listen_to('kytos.topology.updated')
    def update_topology(self, event):
        """
        Update the graph when the network topology was updated.

        Clear the current graph and create a new with the most topology updated.
        """
        if 'topology' not in event.content:
            return
        try:
            topology = event.content['topology']
            self._topology = topology
            self.graph.update_topology(topology)
            log.debug('Topology graph updated.')
        except TypeError as err:
            log.debug(err)
        except Exception as err:
            log.debug(err)