build.main.Main._get_data() - Code Metrics - Inspection of "Enable/disable all interfaces from a switch" - kytos/topology - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#80)

unknown

created 2020-02-20 12:59 UTC

build.main.Main._get_data() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	5
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	5
ccs	1
cts	1
cp	1
rs	10
c	0
b	0
f	0
cc	1
nop	1
crap	1

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

Manage the network topology
"""
from flask import jsonify, request

from kytos.core import KytosEvent, KytosNApp, log, rest
from kytos.core.helpers import listen_to
from kytos.core.interface import Interface
from kytos.core.link import Link
from kytos.core.switch import Switch
# from napps.kytos.topology import settings
from napps.kytos.topology.models import Topology


class Main(KytosNApp):  # pylint: disable=too-many-public-methods
    """Main class of kytos/topology NApp.

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

    def setup(self):
        """Initialize the NApp's links list."""
        self.links = {}
        self.store_items = {}

        self.verify_storehouse('switches')
        self.verify_storehouse('interfaces')
        self.verify_storehouse('links')

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

    def shutdown(self):
        """Do nothing."""
        log.info('NApp kytos/topology shutting down.')

    def _get_link_or_create(self, endpoint_a, endpoint_b):
        new_link = Link(endpoint_a, endpoint_b)

        for link in self.links.values():
            if new_link == link:
                return link

        self.links[new_link.id] = new_link
        return new_link

    def _get_switches_dict(self):
        """Return a dictionary with the known switches."""
        return {'switches': {s.id: s.as_dict() for s in
                             self.controller.switches.values()}}

    def _get_links_dict(self):
        """Return a dictionary with the known links."""
        return {'links': {l.id: l.as_dict() for l in
                          self.links.values()}}

    def _get_topology_dict(self):
        """Return a dictionary with the known topology."""
        return {'topology': {**self._get_switches_dict(),
                             **self._get_links_dict()}}

    def _get_topology(self):
        """Return an object representing the topology."""
        return Topology(self.controller.switches, self.links)

    def _get_link_from_interface(self, interface):
        """Return the link of the interface, or None if it does not exist."""
        for link in self.links.values():
            if interface in (link.endpoint_a, link.endpoint_b):
                return link
        return None

    @staticmethod
    def _get_data(req):
        """Get request data."""
        data = req.get_json() or {}  # Valid format { "interfaces": [...] }
        return data.get('interfaces', [])

    @rest('v3/')
    def get_topology(self):
        """Return the latest known topology.

        This topology is updated when there are network events.
        """
        return jsonify(self._get_topology_dict())

    # Switch related methods
    @rest('v3/switches')
    def get_switches(self):
        """Return a json with all the switches in the topology."""
        return jsonify(self._get_switches_dict())

    @rest('v3/switches/<dpid>/enable', methods=['POST'])
    def enable_switch(self, dpid):
        """Administratively enable a switch in the topology."""
        try:
            self.controller.switches[dpid].enable()
            return jsonify("Operation successful"), 201
        except KeyError:
            return jsonify("Switch not found"), 404

    @rest('v3/switches/<dpid>/disable', methods=['POST'])
    def disable_switch(self, dpid):
        """Administratively disable a switch in the topology."""
        try:
            self.controller.switches[dpid].disable()
            return jsonify("Operation successful"), 201
        except KeyError:
            return jsonify("Switch not found"), 404

    @rest('v3/switches/<dpid>/metadata')
    def get_switch_metadata(self, dpid):
        """Get metadata from a switch."""
        try:
            return jsonify({"metadata":
                            self.controller.switches[dpid].metadata}), 200
        except KeyError:
            return jsonify("Switch not found"), 404

    @rest('v3/switches/<dpid>/metadata', methods=['POST'])
    def add_switch_metadata(self, dpid):
        """Add metadata to a switch."""
        metadata = request.get_json()
        try:
            switch = self.controller.switches[dpid]
        except KeyError:
            return jsonify("Switch not found"), 404

        switch.extend_metadata(metadata)
        self.notify_metadata_changes(switch, 'added')
        return jsonify("Operation successful"), 201

    @rest('v3/switches/<dpid>/metadata/<key>', methods=['DELETE'])
    def delete_switch_metadata(self, dpid, key):
        """Delete metadata from a switch."""
        try:
            switch = self.controller.switches[dpid]
        except KeyError:
            return jsonify("Switch not found"), 404

        switch.remove_metadata(key)
        self.notify_metadata_changes(switch, 'removed')
        return jsonify("Operation successful"), 200

    # Interface related methods
    @rest('v3/interfaces')
    def get_interfaces(self):
        """Return a json with all the interfaces in the topology."""
        interfaces = {}
        switches = self._get_switches_dict()
        for switch in switches['switches'].values():
            for interface_id, interface in switch['interfaces'].items():
                interfaces[interface_id] = interface

        return jsonify({'interfaces': interfaces})

    @rest('v3/interfaces/enable', methods=['POST'])

    @rest('v3/interfaces/<interface_enable_id>/enable', methods=['POST'])
    def enable_interface(self, interface_enable_id=None):
        """Administratively enable a list of interfaces in the topology."""
        error_list = []  # List of interfaces that were not activated.
        interface_ids = []
        msg_error = "Some interfaces couldn't be found and activated: "
        if not interface_enable_id:
            interface_ids = self._get_data(request)
        else:
            interface_ids.append(interface_enable_id)
        for interface_id in interface_ids:
            switch_id = ":".join(interface_id.split(":")[:-1])
            interface_number = int(interface_id.split(":")[-1])

            try:
                switch = self.controller.switches[switch_id]
            except KeyError as exc:
                return jsonify(f"Switch not found: {exc}"), 404

            try:
                switch.interfaces[interface_number].enable()
            except KeyError as exc:
                error_list.append(f"Switch {switch_id} Interface {exc}")

        if not error_list:
            return jsonify("Operation successful"), 201
        return jsonify({msg_error:
                        error_list}), 409

    @rest('v3/interfaces/disable', methods=['POST'])

    @rest('v3/interfaces/<interface_disable_id>/disable', methods=['POST'])
    def disable_interface(self, interface_disable_id=None):
        """Administratively disable a list of interfaces in the topology."""
        error_list = []  # List of interfaces that were not deactivated.
        interface_ids = []
        msg_error = "Some interfaces couldn't be found and deactivated: "
        if not interface_disable_id:
            interface_ids = self._get_data(request)
        else:
            interface_ids.append(interface_disable_id)
        for interface_id in interface_ids:
            switch_id = ":".join(interface_id.split(":")[:-1])
            interface_number = int(interface_id.split(":")[-1])

            try:
                switch = self.controller.switches[switch_id]
            except KeyError as exc:
                return jsonify(f"Switch not found: {exc}"), 404

            try:
                switch.interfaces[interface_number].disable()
            except KeyError as exc:
                error_list.append(f"Switch {switch_id} Interface {exc}")

        if not error_list:
            return jsonify("Operation successful"), 201

        return jsonify({msg_error:
                        error_list}), 409

    @rest('v3/interfaces/<interface_id>/metadata')
    def get_interface_metadata(self, interface_id):
        """Get metadata from an interface."""
        switch_id = ":".join(interface_id.split(":")[:-1])
        interface_number = int(interface_id.split(":")[-1])
        try:
            switch = self.controller.switches[switch_id]
        except KeyError:
            return jsonify("Switch not found"), 404

        try:
            interface = switch.interfaces[interface_number]
        except KeyError:
            return jsonify("Interface not found"), 404

        return jsonify({"metadata": interface.metadata}), 200

    @rest('v3/interfaces/<interface_id>/metadata', methods=['POST'])
    def add_interface_metadata(self, interface_id):
        """Add metadata to an interface."""
        metadata = request.get_json()

        switch_id = ":".join(interface_id.split(":")[:-1])
        interface_number = int(interface_id.split(":")[-1])
        try:
            switch = self.controller.switches[switch_id]
        except KeyError:
            return jsonify("Switch not found"), 404

        try:
            interface = switch.interfaces[interface_number]
        except KeyError:
            return jsonify("Interface not found"), 404

        interface.extend_metadata(metadata)
        self.notify_metadata_changes(interface, 'added')
        return jsonify("Operation successful"), 201

    @rest('v3/interfaces/<interface_id>/metadata/<key>', methods=['DELETE'])
    def delete_interface_metadata(self, interface_id, key):
        """Delete metadata from an interface."""
        switch_id = ":".join(interface_id.split(":")[:-1])
        interface_number = int(interface_id.split(":")[-1])

        try:
            switch = self.controller.switches[switch_id]
        except KeyError:
            return jsonify("Switch not found"), 404

        try:
            interface = switch.interfaces[interface_number]
        except KeyError:
            return jsonify("Interface not found"), 404

        if interface.remove_metadata(key) is False:
            return jsonify("Metadata not found"), 404

        self.notify_metadata_changes(interface, 'removed')
        return jsonify("Operation successful"), 200

    # Link related methods
    @rest('v3/links')
    def get_links(self):
        """Return a json with all the links in the topology.

        Links are connections between interfaces.
        """
        return jsonify(self._get_links_dict()), 200

    @rest('v3/links/<link_id>/enable', methods=['POST'])
    def enable_link(self, link_id):
        """Administratively enable a link in the topology."""
        try:
            self.links[link_id].enable()
        except KeyError:
            return jsonify("Link not found"), 404

        return jsonify("Operation successful"), 201

    @rest('v3/links/<link_id>/disable', methods=['POST'])
    def disable_link(self, link_id):
        """Administratively disable a link in the topology."""
        try:
            self.links[link_id].disable()
        except KeyError:
            return jsonify("Link not found"), 404

        return jsonify("Operation successful"), 201

    @rest('v3/links/<link_id>/metadata')
    def get_link_metadata(self, link_id):
        """Get metadata from a link."""
        try:
            return jsonify({"metadata": self.links[link_id].metadata}), 200
        except KeyError:
            return jsonify("Link not found"), 404

    @rest('v3/links/<link_id>/metadata', methods=['POST'])
    def add_link_metadata(self, link_id):
        """Add metadata to a link."""
        metadata = request.get_json()
        try:
            link = self.links[link_id]
        except KeyError:
            return jsonify("Link not found"), 404

        link.extend_metadata(metadata)
        self.notify_metadata_changes(link, 'added')
        return jsonify("Operation successful"), 201

    @rest('v3/links/<link_id>/metadata/<key>', methods=['DELETE'])
    def delete_link_metadata(self, link_id, key):
        """Delete metadata from a link."""
        try:
            link = self.links[link_id]
        except KeyError:
            return jsonify("Link not found"), 404

        if link.remove_metadata(key) is False:
            return jsonify("Metadata not found"), 404

        self.notify_metadata_changes(link, 'removed')
        return jsonify("Operation successful"), 200

    @listen_to('.*.switch.(new|reconnected)')
    def handle_new_switch(self, event):
        """Create a new Device on the Topology.

        Handle the event of a new created switch and update the topology with
        this new device.
        """
        switch = event.content['switch']
        switch.activate()
        log.debug('Switch %s added to the Topology.', switch.id)
        self.notify_topology_update()
        self.update_instance_metadata(switch)

    @listen_to('.*.connection.lost')
    def handle_connection_lost(self, event):
        """Remove a Device from the topology.

        Remove the disconnected Device and every link that has one of its
        interfaces.
        """
        switch = event.content['source'].switch
        if switch:
            switch.deactivate()
            log.debug('Switch %s removed from the Topology.', switch.id)
            self.notify_topology_update()

    def handle_interface_up(self, event):
        """Update the topology based on a Port Modify event.

        The event notifies that an interface was changed to 'up'.
        """
        interface = event.content['interface']
        interface.activate()
        self.notify_topology_update()
        self.update_instance_metadata(interface)

    @listen_to('.*.switch.interface.created')
    def handle_interface_created(self, event):
        """Update the topology based on a Port Create event."""
        self.handle_interface_up(event)

    def handle_interface_down(self, event):
        """Update the topology based on a Port Modify event.

        The event notifies that an interface was changed to 'down'.
        """
        interface = event.content['interface']
        interface.deactivate()
        self.handle_interface_link_down(event)
        self.notify_topology_update()

    @listen_to('.*.switch.interface.deleted')
    def handle_interface_deleted(self, event):
        """Update the topology based on a Port Delete event."""
        self.handle_interface_down(event)

    @listen_to('.*.switch.interface.link_up')
    def handle_interface_link_up(self, event):
        """Update the topology based on a Port Modify event.

        The event notifies that an interface's link was changed to 'up'.
        """
        interface = event.content['interface']
        link = self._get_link_from_interface(interface)
        if link and not link.is_active():
            link.activate()
            self.notify_topology_update()
            self.update_instance_metadata(interface.link)
            self.notify_link_status_change(link)

    @listen_to('.*.switch.interface.link_down')
    def handle_interface_link_down(self, event):
        """Update the topology based on a Port Modify event.

        The event notifies that an interface's link was changed to 'down'.
        """
        interface = event.content['interface']
        link = self._get_link_from_interface(interface)
        if link and link.is_active():
            link.deactivate()
            self.notify_topology_update()
            self.notify_link_status_change(link)

    @listen_to('.*.interface.is.nni')
    def add_links(self, event):
        """Update the topology with links related to the NNI interfaces."""
        interface_a = event.content['interface_a']
        interface_b = event.content['interface_b']

        link = self._get_link_or_create(interface_a, interface_b)
        interface_a.update_link(link)
        interface_b.update_link(link)

        interface_a.nni = True
        interface_b.nni = True

        self.notify_topology_update()

    # def add_host(self, event):
    #    """Update the topology with a new Host."""

    #    interface = event.content['port']
    #    mac = event.content['reachable_mac']

    #    host = Host(mac)
    #    link = self.topology.get_link(interface.id)
    #    if link is not None:
    #        return

    #    self.topology.add_link(interface.id, host.id)
    #    self.topology.add_device(host)

    #    if settings.DISPLAY_FULL_DUPLEX_LINKS:
    #        self.topology.add_link(host.id, interface.id)

    def notify_topology_update(self):
        """Send an event to notify about updates on the topology."""
        name = 'kytos/topology.updated'
        event = KytosEvent(name=name, content={'topology':
                                               self._get_topology()})
        self.controller.buffers.app.put(event)

    def notify_link_status_change(self, link):
        """Send an event to notify about a status change on a link."""
        name = 'kytos/topology.'
        if link.is_active():
            status = 'link_up'
        else:
            status = 'link_down'
        event = KytosEvent(name=name+status, content={'link': link})
        self.controller.buffers.app.put(event)

    def notify_metadata_changes(self, obj, action):
        """Send an event to notify about metadata changes."""
        if isinstance(obj, Switch):
            entity = 'switch'
            entities = 'switches'
        elif isinstance(obj, Interface):
            entity = 'interface'
            entities = 'interfaces'
        elif isinstance(obj, Link):
            entity = 'link'
            entities = 'links'

        name = f'kytos/topology.{entities}.metadata.{action}'
        event = KytosEvent(name=name, content={entity: obj,

                                               'metadata': obj.metadata})
        self.controller.buffers.app.put(event)
        log.debug(f'Metadata from {obj.id} was {action}.')

    @listen_to('.*.switch.port.created')
    def notify_port_created(self, original_event):
        """Notify when a port is created."""
        name = f'kytos/topology.port.created'
        event = KytosEvent(name=name, content=original_event.content)
        self.controller.buffers.app.put(event)

    @listen_to('kytos/topology.*.metadata.*')
    def save_metadata_on_store(self, event):
        """Send to storehouse the data updated."""
        name = 'kytos.storehouse.update'
        if 'switch' in event.content:
            store = self.store_items.get('switches')
            obj = event.content.get('switch')
            namespace = 'kytos.topology.switches.metadata'
        elif 'interface' in event.content:
            store = self.store_items.get('interfaces')
            obj = event.content.get('interface')
            namespace = 'kytos.topology.iterfaces.metadata'
        elif 'link' in event.content:
            store = self.store_items.get('links')
            obj = event.content.get('link')
            namespace = 'kytos.topology.links.metadata'

        store.data[obj.id] = obj.metadata

        content = {'namespace': namespace,

                   'box_id': store.box_id,
                   'data': store.data,
                   'callback': self.update_instance}

        event = KytosEvent(name=name, content=content)
        self.controller.buffers.app.put(event)

    @staticmethod
    def update_instance(event, _data, error):
        """Display in Kytos console if the data was updated."""
        entities = event.content.get('namespace', '').split('.')[-2]
        if error:
            log.error(f'Error trying to update storehouse {entities}.')
        else:
            log.debug(f'Storehouse update to entities: {entities}.')

    def verify_storehouse(self, entities):
        """Request a list of box saved by specific entity."""
        name = 'kytos.storehouse.list'
        content = {'namespace': f'kytos.topology.{entities}.metadata',
                   'callback': self.request_retrieve_entities}
        event = KytosEvent(name=name, content=content)
        self.controller.buffers.app.put(event)
        log.info(f'verify data in storehouse for {entities}.')

    def request_retrieve_entities(self, event, data, _error):
        """Create a box or retrieve an existent box from storehouse."""
        msg = ''
        content = {'namespace': event.content.get('namespace'),
                   'callback': self.load_from_store,
                   'data': {}}

        if not data:
            name = 'kytos.storehouse.create'
            msg = 'Create new box in storehouse'
        else:
            name = 'kytos.storehouse.retrieve'
            content['box_id'] = data[0]
            msg = 'Retrieve data from storeohouse.'

        event = KytosEvent(name=name, content=content)
        self.controller.buffers.app.put(event)
        log.debug(msg)

    def load_from_store(self, event, box, error):
        """Save the data retrived from storehouse."""
        entities = event.content.get('namespace', '').split('.')[-2]
        if error:
            log.error('Error while get a box from storehouse.')
        else:
            self.store_items[entities] = box
            log.debug('Data updated')

    def update_instance_metadata(self, obj):
        """Update object instance with saved metadata."""
        metadata = None
        if isinstance(obj, Interface):
            all_metadata = self.store_items.get('interfaces', None)
            if all_metadata:
                metadata = all_metadata.data.get(obj.id)
        elif isinstance(obj, Switch):
            all_metadata = self.store_items.get('switches', None)
            if all_metadata:
                metadata = all_metadata.data.get(obj.id)
        elif isinstance(obj, Link):
            all_metadata = self.store_items.get('links', None)
            if all_metadata:
                metadata = all_metadata.data.get(obj.id)

        if metadata:
            obj.extend_metadata(metadata)
            log.debug(f'Metadata to {obj.id} was updated')


1			"""Main module of kytos/topology Kytos Network Application.
2
3			Manage the network topology
4			"""
5	1		from flask import jsonify, request
6	1
7	1		from kytos.core import KytosEvent, KytosNApp, log, rest
8	1		from kytos.core.helpers import listen_to
9	1		from kytos.core.interface import Interface
10	1		from kytos.core.link import Link
11			from kytos.core.switch import Switch
12			# from napps.kytos.topology import settings
13	1		from napps.kytos.topology.models import Topology
14
15
16	1		class Main(KytosNApp): # pylint: disable=too-many-public-methods
17			"""Main class of kytos/topology NApp.
18
19			This class is the entry point for this napp.
20			"""
21
22	1		def setup(self):
23			"""Initialize the NApp's links list."""
24	1		self.links = {}
25	1		self.store_items = {}
26
27	1		self.verify_storehouse('switches')
28	1		self.verify_storehouse('interfaces')
29	1		self.verify_storehouse('links')
30
31	1		def execute(self):
32			"""Do nothing."""
33
34	1		def shutdown(self):
35			"""Do nothing."""
36			log.info('NApp kytos/topology shutting down.')
37
38	1		def _get_link_or_create(self, endpoint_a, endpoint_b):
39			new_link = Link(endpoint_a, endpoint_b)
40
41			for link in self.links.values():
42			if new_link == link:
43			return link
44
45			self.links[new_link.id] = new_link
46			return new_link
47
48	1		def _get_switches_dict(self):
49			"""Return a dictionary with the known switches."""
50			return {'switches': {s.id: s.as_dict() for s in
51			self.controller.switches.values()}}
52
53	1		def _get_links_dict(self):
54			"""Return a dictionary with the known links."""
55			return {'links': {l.id: l.as_dict() for l in
56			self.links.values()}}
57
58	1		def _get_topology_dict(self):
59			"""Return a dictionary with the known topology."""
60			return {'topology': {**self._get_switches_dict(),
61			**self._get_links_dict()}}
62
63	1		def _get_topology(self):
64			"""Return an object representing the topology."""
65			return Topology(self.controller.switches, self.links)
66
67	1		def _get_link_from_interface(self, interface):
68			"""Return the link of the interface, or None if it does not exist."""
69			for link in self.links.values():
70			if interface in (link.endpoint_a, link.endpoint_b):
71			return link
72			return None
73
74	1		@staticmethod
75			def _get_data(req):
76			"""Get request data."""
77			data = req.get_json() or {} # Valid format { "interfaces": [...] }
78			return data.get('interfaces', [])
79
80			@rest('v3/')
81			def get_topology(self):
82			"""Return the latest known topology.
83	1
84			This topology is updated when there are network events.
85			"""
86			return jsonify(self._get_topology_dict())
87
88	1		# Switch related methods
89			@rest('v3/switches')
90			def get_switches(self):
91			"""Return a json with all the switches in the topology."""
92			return jsonify(self._get_switches_dict())
93
94			@rest('v3/switches/<dpid>/enable', methods=['POST'])
95			def enable_switch(self, dpid):
96			"""Administratively enable a switch in the topology."""
97	1		try:
98			self.controller.switches[dpid].enable()
99			return jsonify("Operation successful"), 201
100			except KeyError:
101			return jsonify("Switch not found"), 404
102
103			@rest('v3/switches/<dpid>/disable', methods=['POST'])
104			def disable_switch(self, dpid):
105			"""Administratively disable a switch in the topology."""
106	1		try:
107			self.controller.switches[dpid].disable()
108			return jsonify("Operation successful"), 201
109			except KeyError:
110			return jsonify("Switch not found"), 404
111
112			@rest('v3/switches/<dpid>/metadata')
113			def get_switch_metadata(self, dpid):
114			"""Get metadata from a switch."""
115	1		try:
116			return jsonify({"metadata":
117			self.controller.switches[dpid].metadata}), 200
118			except KeyError:
119			return jsonify("Switch not found"), 404
120
121			@rest('v3/switches/<dpid>/metadata', methods=['POST'])
122			def add_switch_metadata(self, dpid):
123			"""Add metadata to a switch."""
124			metadata = request.get_json()
125			try:
126			switch = self.controller.switches[dpid]
127			except KeyError:
128	1		return jsonify("Switch not found"), 404
129
130			switch.extend_metadata(metadata)
131			self.notify_metadata_changes(switch, 'added')
132			return jsonify("Operation successful"), 201
133
134			@rest('v3/switches/<dpid>/metadata/<key>', methods=['DELETE'])
135			def delete_switch_metadata(self, dpid, key):
136			"""Delete metadata from a switch."""
137			try:
138			switch = self.controller.switches[dpid]
139			except KeyError:
140			return jsonify("Switch not found"), 404
141	1
142			switch.remove_metadata(key)
143			self.notify_metadata_changes(switch, 'removed')
144			return jsonify("Operation successful"), 200
145
146			# Interface related methods
147			@rest('v3/interfaces')
148			def get_interfaces(self):
149			"""Return a json with all the interfaces in the topology."""
150			interfaces = {}
151			switches = self._get_switches_dict()
152	1		for switch in switches['switches'].values():
153			for interface_id, interface in switch['interfaces'].items():
154			interfaces[interface_id] = interface
155
156			return jsonify({'interfaces': interfaces})
157
158		View Code Duplication	@rest('v3/interfaces/enable', methods=['POST'])
			0 ignored issues – show Duplication introduced 2020-01-13 15:12 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
159			@rest('v3/interfaces/<interface_enable_id>/enable', methods=['POST'])
160			def enable_interface(self, interface_enable_id=None):
161			"""Administratively enable a list of interfaces in the topology."""
162			error_list = [] # List of interfaces that were not activated.
163			interface_ids = []
164			msg_error = "Some interfaces couldn't be found and activated: "
165			if not interface_enable_id:
166			interface_ids = self._get_data(request)
167			else:
168			interface_ids.append(interface_enable_id)
169			for interface_id in interface_ids:
170	1		switch_id = ":".join(interface_id.split(":")[:-1])
171			interface_number = int(interface_id.split(":")[-1])
172
173			try:
174			switch = self.controller.switches[switch_id]
175			except KeyError as exc:
176			return jsonify(f"Switch not found: {exc}"), 404
177
178			try:
179			switch.interfaces[interface_number].enable()
180			except KeyError as exc:
181			error_list.append(f"Switch {switch_id} Interface {exc}")
182
183			if not error_list:
184			return jsonify("Operation successful"), 201
185			return jsonify({msg_error:
186			error_list}), 409
187
188	1	View Code Duplication	@rest('v3/interfaces/disable', methods=['POST'])
			0 ignored issues – show Duplication introduced 2020-01-10 20:50 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
189			@rest('v3/interfaces/<interface_disable_id>/disable', methods=['POST'])
190			def disable_interface(self, interface_disable_id=None):
191			"""Administratively disable a list of interfaces in the topology."""
192			error_list = [] # List of interfaces that were not deactivated.
193			interface_ids = []
194			msg_error = "Some interfaces couldn't be found and deactivated: "
195			if not interface_disable_id:
196			interface_ids = self._get_data(request)
197			else:
198			interface_ids.append(interface_disable_id)
199			for interface_id in interface_ids:
200			switch_id = ":".join(interface_id.split(":")[:-1])
201			interface_number = int(interface_id.split(":")[-1])
202
203			try:
204			switch = self.controller.switches[switch_id]
205	1		except KeyError as exc:
206			return jsonify(f"Switch not found: {exc}"), 404
207
208			try:
209			switch.interfaces[interface_number].disable()
210			except KeyError as exc:
211			error_list.append(f"Switch {switch_id} Interface {exc}")
212
213			if not error_list:
214			return jsonify("Operation successful"), 201
215
216			return jsonify({msg_error:
217			error_list}), 409
218
219			@rest('v3/interfaces/<interface_id>/metadata')
220			def get_interface_metadata(self, interface_id):
221			"""Get metadata from an interface."""
222			switch_id = ":".join(interface_id.split(":")[:-1])
223			interface_number = int(interface_id.split(":")[-1])
224			try:
225			switch = self.controller.switches[switch_id]
226	1		except KeyError:
227			return jsonify("Switch not found"), 404
228
229			try:
230			interface = switch.interfaces[interface_number]
231			except KeyError:
232			return jsonify("Interface not found"), 404
233
234			return jsonify({"metadata": interface.metadata}), 200
235
236			@rest('v3/interfaces/<interface_id>/metadata', methods=['POST'])
237			def add_interface_metadata(self, interface_id):
238			"""Add metadata to an interface."""
239			metadata = request.get_json()
240
241			switch_id = ":".join(interface_id.split(":")[:-1])
242			interface_number = int(interface_id.split(":")[-1])
243			try:
244			switch = self.controller.switches[switch_id]
245			except KeyError:
246			return jsonify("Switch not found"), 404
247
248			try:
249	1		interface = switch.interfaces[interface_number]
250			except KeyError:
251			return jsonify("Interface not found"), 404
252
253			interface.extend_metadata(metadata)
254			self.notify_metadata_changes(interface, 'added')
255			return jsonify("Operation successful"), 201
256
257	1		@rest('v3/interfaces/<interface_id>/metadata/<key>', methods=['DELETE'])
258			def delete_interface_metadata(self, interface_id, key):
259			"""Delete metadata from an interface."""
260			switch_id = ":".join(interface_id.split(":")[:-1])
261			interface_number = int(interface_id.split(":")[-1])
262
263			try:
264			switch = self.controller.switches[switch_id]
265			except KeyError:
266			return jsonify("Switch not found"), 404
267	1
268			try:
269			interface = switch.interfaces[interface_number]
270			except KeyError:
271			return jsonify("Interface not found"), 404
272
273			if interface.remove_metadata(key) is False:
274			return jsonify("Metadata not found"), 404
275
276			self.notify_metadata_changes(interface, 'removed')
277	1		return jsonify("Operation successful"), 200
278
279			# Link related methods
280			@rest('v3/links')
281			def get_links(self):
282			"""Return a json with all the links in the topology.
283
284			Links are connections between interfaces.
285	1		"""
286			return jsonify(self._get_links_dict()), 200
287
288			@rest('v3/links/<link_id>/enable', methods=['POST'])
289			def enable_link(self, link_id):
290			"""Administratively enable a link in the topology."""
291			try:
292			self.links[link_id].enable()
293			except KeyError:
294			return jsonify("Link not found"), 404
295
296			return jsonify("Operation successful"), 201
297
298	1		@rest('v3/links/<link_id>/disable', methods=['POST'])
299			def disable_link(self, link_id):
300			"""Administratively disable a link in the topology."""
301			try:
302			self.links[link_id].disable()
303			except KeyError:
304			return jsonify("Link not found"), 404
305
306			return jsonify("Operation successful"), 201
307
308			@rest('v3/links/<link_id>/metadata')
309			def get_link_metadata(self, link_id):
310			"""Get metadata from a link."""
311			try:
312	1		return jsonify({"metadata": self.links[link_id].metadata}), 200
313			except KeyError:
314			return jsonify("Link not found"), 404
315
316			@rest('v3/links/<link_id>/metadata', methods=['POST'])
317			def add_link_metadata(self, link_id):
318			"""Add metadata to a link."""
319			metadata = request.get_json()
320			try:
321			link = self.links[link_id]
322			except KeyError:
323			return jsonify("Link not found"), 404
324
325	1		link.extend_metadata(metadata)
326			self.notify_metadata_changes(link, 'added')
327			return jsonify("Operation successful"), 201
328
329			@rest('v3/links/<link_id>/metadata/<key>', methods=['DELETE'])
330			def delete_link_metadata(self, link_id, key):
331			"""Delete metadata from a link."""
332			try:
333			link = self.links[link_id]
334			except KeyError:
335			return jsonify("Link not found"), 404
336
337			if link.remove_metadata(key) is False:
338	1		return jsonify("Metadata not found"), 404
339
340			self.notify_metadata_changes(link, 'removed')
341			return jsonify("Operation successful"), 200
342
343			@listen_to('.*.switch.(new\|reconnected)')
344			def handle_new_switch(self, event):
345			"""Create a new Device on the Topology.
346
347			Handle the event of a new created switch and update the topology with
348	1		this new device.
349			"""
350			switch = event.content['switch']
351			switch.activate()
352			log.debug('Switch %s added to the Topology.', switch.id)
353	1		self.notify_topology_update()
354			self.update_instance_metadata(switch)
355
356			@listen_to('.*.connection.lost')
357			def handle_connection_lost(self, event):
358			"""Remove a Device from the topology.
359
360			Remove the disconnected Device and every link that has one of its
361			interfaces.
362			"""
363	1		switch = event.content['source'].switch
364			if switch:
365			switch.deactivate()
366			log.debug('Switch %s removed from the Topology.', switch.id)
367			self.notify_topology_update()
368	1
369			def handle_interface_up(self, event):
370			"""Update the topology based on a Port Modify event.
371
372			The event notifies that an interface was changed to 'up'.
373			"""
374			interface = event.content['interface']
375			interface.activate()
376			self.notify_topology_update()
377			self.update_instance_metadata(interface)
378
379			@listen_to('.*.switch.interface.created')
380			def handle_interface_created(self, event):
381			"""Update the topology based on a Port Create event."""
382	1		self.handle_interface_up(event)
383
384			def handle_interface_down(self, event):
385			"""Update the topology based on a Port Modify event.
386
387			The event notifies that an interface was changed to 'down'.
388			"""
389			interface = event.content['interface']
390			interface.deactivate()
391			self.handle_interface_link_down(event)
392			self.notify_topology_update()
393
394			@listen_to('.*.switch.interface.deleted')
395	1		def handle_interface_deleted(self, event):
396			"""Update the topology based on a Port Delete event."""
397			self.handle_interface_down(event)
398
399			@listen_to('.*.switch.interface.link_up')
400			def handle_interface_link_up(self, event):
401			"""Update the topology based on a Port Modify event.
402
403			The event notifies that an interface's link was changed to 'up'.
404			"""
405			interface = event.content['interface']
406			link = self._get_link_from_interface(interface)
407			if link and not link.is_active():
408			link.activate()
409			self.notify_topology_update()
410			self.update_instance_metadata(interface.link)
411			self.notify_link_status_change(link)
412
413			@listen_to('.*.switch.interface.link_down')
414			def handle_interface_link_down(self, event):
415			"""Update the topology based on a Port Modify event.
416
417			The event notifies that an interface's link was changed to 'down'.
418			"""
419			interface = event.content['interface']
420			link = self._get_link_from_interface(interface)
421			if link and link.is_active():
422			link.deactivate()
423			self.notify_topology_update()
424			self.notify_link_status_change(link)
425
426			@listen_to('.*.interface.is.nni')
427	1		def add_links(self, event):
428			"""Update the topology with links related to the NNI interfaces."""
429			interface_a = event.content['interface_a']
430			interface_b = event.content['interface_b']
431
432			link = self._get_link_or_create(interface_a, interface_b)
433			interface_a.update_link(link)
434	1		interface_b.update_link(link)
435
436			interface_a.nni = True
437			interface_b.nni = True
438
439			self.notify_topology_update()
440
441			# def add_host(self, event):
442			# """Update the topology with a new Host."""
443
444	1		# interface = event.content['port']
445			# mac = event.content['reachable_mac']
446
447			# host = Host(mac)
448			# link = self.topology.get_link(interface.id)
449			# if link is not None:
450			# return
451
452			# self.topology.add_link(interface.id, host.id)
453			# self.topology.add_device(host)
454
455			# if settings.DISPLAY_FULL_DUPLEX_LINKS:
456			# self.topology.add_link(host.id, interface.id)
457
458			def notify_topology_update(self):
459			"""Send an event to notify about updates on the topology."""
460			name = 'kytos/topology.updated'
461			event = KytosEvent(name=name, content={'topology':
462	1		self._get_topology()})
463			self.controller.buffers.app.put(event)
464
465			def notify_link_status_change(self, link):
466			"""Send an event to notify about a status change on a link."""
467			name = 'kytos/topology.'
468			if link.is_active():
469	1		status = 'link_up'
470			else:
471			status = 'link_down'
472			event = KytosEvent(name=name+status, content={'link': link})
473			self.controller.buffers.app.put(event)
474
475			def notify_metadata_changes(self, obj, action):
476			"""Send an event to notify about metadata changes."""
477			if isinstance(obj, Switch):
478			entity = 'switch'
479			entities = 'switches'
480			elif isinstance(obj, Interface):
481			entity = 'interface'
482			entities = 'interfaces'
483			elif isinstance(obj, Link):
484			entity = 'link'
485			entities = 'links'
486
487			name = f'kytos/topology.{entities}.metadata.{action}'
488			event = KytosEvent(name=name, content={entity: obj,
			0 ignored issues – show introduced 2019-02-27 14:03 UTC by Report Bug Copy Issue Report The variable `entity` does not seem to be defined for all execution paths. Loading history...
489			'metadata': obj.metadata})
490			self.controller.buffers.app.put(event)
491			log.debug(f'Metadata from {obj.id} was {action}.')
492
493			@listen_to('.*.switch.port.created')
494			def notify_port_created(self, original_event):
495	1		"""Notify when a port is created."""
496			name = f'kytos/topology.port.created'
497			event = KytosEvent(name=name, content=original_event.content)
498			self.controller.buffers.app.put(event)
499
500			@listen_to('kytos/topology..metadata.')
501			def save_metadata_on_store(self, event):
502			"""Send to storehouse the data updated."""
503			name = 'kytos.storehouse.update'
504	1		if 'switch' in event.content:
505			store = self.store_items.get('switches')
506	1		obj = event.content.get('switch')
507	1		namespace = 'kytos.topology.switches.metadata'
508			elif 'interface' in event.content:
509	1		store = self.store_items.get('interfaces')
510	1		obj = event.content.get('interface')
511	1		namespace = 'kytos.topology.iterfaces.metadata'
512			elif 'link' in event.content:
513	1		store = self.store_items.get('links')
514			obj = event.content.get('link')
515			namespace = 'kytos.topology.links.metadata'
516
517			store.data[obj.id] = obj.metadata
			0 ignored issues – show introduced 2019-02-27 14:03 UTC by Report Bug Copy Issue Report The variable `obj` does not seem to be defined for all execution paths. Loading history... introduced 2019-02-27 14:03 UTC by Report Bug Copy Issue Report The variable `store` does not seem to be defined for all execution paths. Loading history...
518			content = {'namespace': namespace,
			0 ignored issues – show introduced 2019-02-27 14:03 UTC by Report Bug Copy Issue Report The variable `namespace` does not seem to be defined for all execution paths. Loading history...
519			'box_id': store.box_id,
520			'data': store.data,
521			'callback': self.update_instance}
522
523			event = KytosEvent(name=name, content=content)
524			self.controller.buffers.app.put(event)
525
526			@staticmethod
527			def update_instance(event, _data, error):
528			"""Display in Kytos console if the data was updated."""
529			entities = event.content.get('namespace', '').split('.')[-2]
530			if error:
531			log.error(f'Error trying to update storehouse {entities}.')
532	1		else:
533			log.debug(f'Storehouse update to entities: {entities}.')
534
535			def verify_storehouse(self, entities):
536			"""Request a list of box saved by specific entity."""
537			name = 'kytos.storehouse.list'
538			content = {'namespace': f'kytos.topology.{entities}.metadata',
539			'callback': self.request_retrieve_entities}
540			event = KytosEvent(name=name, content=content)
541	1		self.controller.buffers.app.put(event)
542			log.info(f'verify data in storehouse for {entities}.')
543
544			def request_retrieve_entities(self, event, data, _error):
545			"""Create a box or retrieve an existent box from storehouse."""
546			msg = ''
547			content = {'namespace': event.content.get('namespace'),
548			'callback': self.load_from_store,
549			'data': {}}
550
551			if not data:
552			name = 'kytos.storehouse.create'
553			msg = 'Create new box in storehouse'
554			else:
555			name = 'kytos.storehouse.retrieve'
556			content['box_id'] = data[0]
557			msg = 'Retrieve data from storeohouse.'
558
559			event = KytosEvent(name=name, content=content)
560			self.controller.buffers.app.put(event)
561			log.debug(msg)
562
563			def load_from_store(self, event, box, error):
564			"""Save the data retrived from storehouse."""
565			entities = event.content.get('namespace', '').split('.')[-2]
566			if error:
567			log.error('Error while get a box from storehouse.')
568			else:
569			self.store_items[entities] = box
570			log.debug('Data updated')
571
572			def update_instance_metadata(self, obj):
573			"""Update object instance with saved metadata."""
574			metadata = None
575			if isinstance(obj, Interface):
576			all_metadata = self.store_items.get('interfaces', None)
577			if all_metadata:
578			metadata = all_metadata.data.get(obj.id)
579			elif isinstance(obj, Switch):
580			all_metadata = self.store_items.get('switches', None)
581			if all_metadata:
582			metadata = all_metadata.data.get(obj.id)
583			elif isinstance(obj, Link):
584			all_metadata = self.store_items.get('links', None)
585			if all_metadata:
586			metadata = all_metadata.data.get(obj.id)
587
588			if metadata:
589			obj.extend_metadata(metadata)
590			log.debug(f'Metadata to {obj.id} was updated')
591

kytos / topology

Pull Request — master (#80)

build.main.Main._get_data() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like