build.main

Complexity

Total Complexity

24

Size/Duplication

Total Lines	139
Duplicated Lines	46.04 %

Test Coverage

Coverage

68.83%

Importance

Changes

0

8 Methods

Rating	Name	Duplication	Size	Complexity
A	Main.shortest_path()	0	18	2
A	Main.shutdown()	0	2	1
A	Main.setup()	0	4	1
A	Main.execute()	0	2	1
D	Main._filter_paths()	31	42	13
A	Main.shortest_constrained_path()	17	17	2
A	Main.shortest_constrained_path2()	16	16	2
A	Main.update_topology()	0	12	2

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

# pylint: disable=import-error
from napps.kytos.pathfinder.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('v3/', methods=['POST'])

This code seems to be duplicated in your project.

    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')
        depth = data.get('depth', 0)
        base_metrics = data.get('base_metrics', {})
        try:
            paths = self.graph.constrained_flexible_paths(source,
                                                          destination,
                                                          depth,
                                                          base=base_metrics)
            return jsonify(paths)
        except TypeError as err:
            return jsonify({"error": err})

    @ rest('v4/', methods=['POST'])

This code seems to be duplicated in your project.

    def shortest_constrained_path2(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', {})
        try:
            paths = self.graph.constrained_flexible_paths(source, destination,
                                                          base=base_metrics,
                                                          flexible=fle_metrics)
            return jsonify(paths)
        except TypeError as err:
            return jsonify({"error": err})

    @ 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 topoly updated.
        """
        if 'topology' not in event.content:
            return
        topology = event.content['topology']
        self._topology = topology
        self.graph.update_topology(topology)
        log.debug('Topology graph updated.')