build.main.Main.enable_lldp() - Code Metrics - Inspection of "Now the user chooses the interface to send LLDP #3..." - kytos/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#32)

unknown

created 2019-08-01 01:25 UTC

build.main.Main.enable_lldp() B

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	30
Code Lines	24

Duplication

Lines	30
Ratio	100 %

Code Coverage

Tests	0
CRAP Score	56

Importance

Changes

Metric	Value
cc	7
eloc	24
nop	1
dl	30
loc	30
ccs	0
cts	22
cp	0
crap	56
rs	7.904
c	0
b	0
f	0

"""NApp responsible to discover new switches and hosts."""
import struct

from flask import jsonify, request
from pyof.foundation.basic_types import DPID, UBInt16, UBInt32
from pyof.foundation.network_types import LLDP, VLAN, Ethernet, EtherType
from pyof.v0x01.common.action import ActionOutput as AO10
from pyof.v0x01.common.phy_port import Port as Port10
from pyof.v0x01.controller2switch.flow_mod import FlowMod as FM10
from pyof.v0x01.controller2switch.flow_mod import FlowModCommand as FMC
from pyof.v0x01.controller2switch.packet_out import PacketOut as PO10
from pyof.v0x04.common.action import ActionOutput as AO13
from pyof.v0x04.common.flow_instructions import InstructionApplyAction
from pyof.v0x04.common.flow_match import OxmOfbMatchField, OxmTLV, VlanId
from pyof.v0x04.common.port import PortNo as Port13
from pyof.v0x04.controller2switch.flow_mod import FlowMod as FM13
from pyof.v0x04.controller2switch.packet_out import PacketOut as PO13

from kytos.core import KytosEvent, KytosNApp, log, rest
from kytos.core.helpers import listen_to
from napps.kytos.of_lldp import constants, settings


class Main(KytosNApp):
    """Main OF_LLDP NApp Class."""

    def setup(self):
        """Make this NApp run in a loop."""
        self.vlan_id = None
        if hasattr(settings, "FLOW_VLAN_VID"):
            self.vlan_id = settings.FLOW_VLAN_VID
        self.execute_as_loop(settings.POLLING_TIME)

    def execute(self):
        """Send LLDP Packets every 'POLLING_TIME' seconds to all switches."""
        switches = list(self.controller.switches.values())
        for switch in switches:
            try:
                of_version = switch.connection.protocol.version
            except AttributeError:
                of_version = None

            if not switch.is_connected():
                continue

            if of_version == 0x01:
                port_type = UBInt16
                local_port = Port10.OFPP_LOCAL
            elif of_version == 0x04:
                port_type = UBInt32
                local_port = Port13.OFPP_LOCAL
            else:
                # skip the current switch with unsupported OF version
                continue

            interfaces = list(switch.interfaces.values())
            for interface in interfaces:
                # Interface marked to receive lldp package
                if not interface.lldp:
                    # print(f"Disabled LLDP package {interface.id}")
                    continue
                # Avoid the interface that connects to the controller.
                if interface.port_number == local_port:
                    continue

                lldp = LLDP()
                lldp.chassis_id.sub_value = DPID(switch.dpid)
                lldp.port_id.sub_value = port_type(interface.port_number)

                ethernet = Ethernet()
                ethernet.ether_type = EtherType.LLDP
                ethernet.source = interface.address
                ethernet.destination = constants.LLDP_MULTICAST_MAC
                ethernet.data = lldp.pack()
                # self.vlan_id == None will result in a pack with no VLAN.
                ethernet.vlans.append(VLAN(vid=self.vlan_id))

                packet_out = self._build_lldp_packet_out(
                                    of_version,
                                    interface.port_number, ethernet.pack())

                if packet_out is None:
                    continue

                event_out = KytosEvent(
                    name='kytos/of_lldp.messages.out.ofpt_packet_out',
                    content={
                            'destination': switch.connection,
                            'message': packet_out})
                self.controller.buffers.msg_out.put(event_out)
                # print(f"Sending package to interface {interface.id}")
                log.debug(
                    "Sending a LLDP PacketOut to the switch %s",
                    switch.dpid)

                msg = '\n'
                msg += 'Switch: %s (%s)\n'
                msg += ' Interfaces: %s\n'
                msg += ' -- LLDP PacketOut --\n'
                msg += ' Ethernet: eth_type (%s) | src (%s) | dst (%s)'
                msg += '\n'
                msg += ' LLDP: Switch (%s) | port (%s)'

                log.debug(
                    msg,
                    switch.connection.address, switch.dpid,
                    switch.interfaces, ethernet.ether_type,
                    ethernet.source, ethernet.destination,
                    switch.dpid, interface.port_number)

    @listen_to('kytos/of_core.handshake.completed')
    def install_lldp_flow(self, event):
        """Install a flow to send LLDP packets to the controller.

        The proactive flow is installed whenever a switch connects.

        Args:
            event (:class:`~kytos.core.events.KytosEvent`):
                Event with new switch information.

        """
        try:
            of_version = event.content['switch'].connection.protocol.version
        except AttributeError:
            of_version = None

        flow_mod = self._build_lldp_flow_mod(of_version)

        if flow_mod:
            name = 'kytos/of_lldp.messages.out.ofpt_flow_mod'
            content = {'destination': event.content['switch'].connection,
                       'message': flow_mod}

            event_out = KytosEvent(name=name, content=content)
            self.controller.buffers.msg_out.put(event_out)

    @listen_to('kytos/of_core.v0x0[14].messages.in.ofpt_packet_in')
    def notify_uplink_detected(self, event):
        """Dispatch two KytosEvents to notify identified NNI interfaces.

        Args:
            event (:class:`~kytos.core.events.KytosEvent`):
                Event with an LLDP packet as data.

        """
        ethernet = self._unpack_non_empty(Ethernet, event.message.data)
        if ethernet.ether_type == EtherType.LLDP:
            try:
                lldp = self._unpack_non_empty(LLDP, ethernet.data)
                dpid = self._unpack_non_empty(DPID, lldp.chassis_id.sub_value)
            except struct.error:
                #: If we have a LLDP packet but we cannot unpack it, or the
                #: unpacked packet does not contain the dpid attribute, then
                #: we are dealing with a LLDP generated by someone else. Thus
                #: this packet is not useful for us and we may just ignore it.
                return

            switch_a = event.source.switch
            port_a = event.message.in_port
            switch_b = None
            port_b = None

            # in_port is currently a UBInt16 in v0x01 and an Int in v0x04.
            if isinstance(port_a, int):
                port_a = UBInt32(port_a)

            try:
                switch_b = self.controller.get_switch_by_dpid(dpid.value)
                of_version = switch_b.connection.protocol.version
                port_type = UBInt16 if of_version == 0x01 else UBInt32
                port_b = self._unpack_non_empty(port_type,
                                                lldp.port_id.sub_value)
            except AttributeError:
                log.debug("Couldn't find datapath %s.", dpid.value)

            # Return if any of the needed information are not available
            if not (switch_a and port_a and switch_b and port_b):
                return

            interface_a = switch_a.get_interface_by_port_no(port_a.value)
            interface_b = switch_b.get_interface_by_port_no(port_b.value)

            event_out = KytosEvent(name='kytos/of_lldp.interface.is.nni',
                                   content={'interface_a': interface_a,
                                            'interface_b': interface_b})
            self.controller.buffers.app.put(event_out)

    def shutdown(self):
        """End of the application."""
        log.debug('Shutting down...')

    @staticmethod
    def _build_lldp_packet_out(version, port_number, data):
        """Build a LLDP PacketOut message.

        Args:
            version (int): OpenFlow version
            port_number (int): Switch port number where the packet must be
                forwarded to.
            data (bytes): Binary data to be sent through the port.

        Returns:
            PacketOut message for the specific given OpenFlow version, if it
                is supported.
            None if the OpenFlow version is not supported.

        """
        if version == 0x01:
            action_output_class = AO10
            packet_out_class = PO10
        elif version == 0x04:
            action_output_class = AO13
            packet_out_class = PO13
        else:
            log.info('Openflow version %s is not yet supported.', version)
            return None

        output_action = action_output_class()
        output_action.port = port_number

        packet_out = packet_out_class()
        packet_out.data = data
        packet_out.actions.append(output_action)

        return packet_out

    def _build_lldp_flow_mod(self, version):
        """Build a FlodMod message to send LLDP to the controller.

        Args:
            version (int): OpenFlow version.

        Returns:
            FlowMod message for the specific given OpenFlow version, if it is
                supported.
            None if the OpenFlow version is not supported.

        """
        if version == 0x01:
            flow_mod = FM10()
            flow_mod.command = FMC.OFPFC_ADD
            flow_mod.priority = settings.FLOW_PRIORITY
            flow_mod.match.dl_type = EtherType.LLDP
            if self.vlan_id:
                flow_mod.match.dl_vlan = self.vlan_id
            flow_mod.actions.append(AO10(port=Port10.OFPP_CONTROLLER))

        elif version == 0x04:
            flow_mod = FM13()
            flow_mod.command = FMC.OFPFC_ADD
            flow_mod.priority = settings.FLOW_PRIORITY

            match_lldp = OxmTLV()
            match_lldp.oxm_field = OxmOfbMatchField.OFPXMT_OFB_ETH_TYPE
            match_lldp.oxm_value = EtherType.LLDP.to_bytes(2, 'big')
            flow_mod.match.oxm_match_fields.append(match_lldp)

            if self.vlan_id:
                match_vlan = OxmTLV()
                match_vlan.oxm_field = OxmOfbMatchField.OFPXMT_OFB_VLAN_VID
                vlan_value = self.vlan_id | VlanId.OFPVID_PRESENT
                match_vlan.oxm_value = vlan_value.to_bytes(2, 'big')
                flow_mod.match.oxm_match_fields.append(match_vlan)

            instruction = InstructionApplyAction()
            instruction.actions.append(AO13(port=Port13.OFPP_CONTROLLER))
            flow_mod.instructions.append(instruction)

        else:
            flow_mod = None

        return flow_mod

    @staticmethod
    def _unpack_non_empty(desired_class, data):
        """Unpack data using an instance of desired_class.

        Args:
            desired_class (class): The class to be used to unpack data.
            data (bytes): bytes to be unpacked.

        Return:
            An instance of desired_class class with data unpacked into it.

        Raises:
            UnpackException if the unpack could not be performed.

        """
        obj = desired_class()

        if hasattr(data, 'value'):
            data = data.value

        obj.unpack(data)

        return obj

    @staticmethod
    def _get_data(req):
        """Get request data."""
        data = req.get_json()
        interface_list = []
        if data:
            interface_list = data.get('interfaces')
        return interface_list

    def _get_interfaces(self):
        """Get all interfaces."""
        interfaces = []
        switches = list(self.controller.switches.values())
        for switch in switches:
            interface_list = list(switch.interfaces.values())
            interfaces = interfaces + interface_list
        return interfaces

    @staticmethod
    def _get_interfaces_dict(interfaces):
        """Return a dict of interfaces."""
        interfaces_dic = {}
        interfaces_dic = {interface.id: interface for interface in interfaces}
        return interfaces_dic

    def _get_lldp_interfaces(self):
        """Get interfaces enabled to receive the LLDP package."""
        enabled_lldp = []
        interfaces = self._get_interfaces()
        enabled_lldp = [inter.id for inter in interfaces if inter.lldp]
        return enabled_lldp

    @rest('v1/interfaces', methods=['GET'])
    def get_lldp_interfaces(self):
        """Return of the interfaces available to receive the LLDP package."""
        result = {"interfaces": self._get_lldp_interfaces()}
        return jsonify(result), 200

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

    def disable_lldp(self):
        """Disables an interface to receive an LLDP package."""
        interfaces_id = self._get_data(request)
        error_list = []
        if interfaces_id is None:
            return jsonify("Bad format."), 401
        interfaces = self._get_interfaces()

        if not interfaces:
            return jsonify("No interfaces were found"), 404

        interfaces = self._get_interfaces_dict(interfaces)
        for interface_id in interfaces_id:
            if not interface_id:
                continue
            if interfaces.get(interface_id):
                interface = interfaces[interface_id]
                interface.lldp = False
            else:
                # List of interfaces that were not activated.
                error_list.append(interface_id)
        if not error_list:
            return jsonify("All interface passed has been disabled."), 200

        # Return a list of interfaces has not disabled.
        # Other interfaces passed has enabled.
        msg_partial_enabled = ""
        msg_partial_enabled += "Some interfaces have not been disabled. "
        msg_partial_enabled += "Interface list not disabled:"
        return jsonify({msg_partial_enabled:
                        error_list}), 400

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

    def enable_lldp(self):
        """Enable an interface to receive an LLDP package."""
        interfaces_id = self._get_data(request)
        error_list = []
        if interfaces_id is None:
            return jsonify("Bad format."), 401
        interfaces = self._get_interfaces()
        if not interfaces:
            return jsonify("No interfaces were found"), 404
        interfaces = self._get_interfaces_dict(interfaces)
        for interface_id in interfaces_id:
            if not interface_id:
                continue
            if interfaces.get(interface_id):
                interface = interfaces[interface_id]
                interface.lldp = True
            else:
                # List of interfaces that were not activated.
                error_list.append(interface_id)
        if not error_list:
            return jsonify("All interface passed has been enabled."), 200

        # Return a list of interfaces has not enabled.
        # The other interfaces have been activated.
        msg_partial_enabled = ""
        msg_partial_enabled += "Some interfaces have not been enabled. "
        msg_partial_enabled += "Interface list not enabled:"
        return jsonify({msg_partial_enabled:
                        error_list}), 400

    @rest('v1/interfaces/disable/all', methods=['POST'])
    def disable_lldp_all(self):
        """Disable LLDP on all interfaces."""
        interfaces = self._get_interfaces()
        if not interfaces:
            return jsonify("No interfaces were found"), 404
        for interface in interfaces:
            interface.lldp = False

        return jsonify("All interfaces has been disabled"), 200

    @rest('v1/interfaces/enable/all', methods=['POST'])
    def enable_all_lldp(self):
        """Enable LLLDP on all interfaces."""
        interfaces = self._get_interfaces()
        if not interfaces:
            return jsonify("No interfaces were found."), 404
        for interface in interfaces:
            interface.lldp = True

        return jsonify("All interfaces has been enabled."), 200


1		"""NApp responsible to discover new switches and hosts."""
2		import struct
3
4		from flask import jsonify, request
5		from pyof.foundation.basic_types import DPID, UBInt16, UBInt32
6		from pyof.foundation.network_types import LLDP, VLAN, Ethernet, EtherType
7		from pyof.v0x01.common.action import ActionOutput as AO10
8		from pyof.v0x01.common.phy_port import Port as Port10
9		from pyof.v0x01.controller2switch.flow_mod import FlowMod as FM10
10		from pyof.v0x01.controller2switch.flow_mod import FlowModCommand as FMC
11		from pyof.v0x01.controller2switch.packet_out import PacketOut as PO10
12		from pyof.v0x04.common.action import ActionOutput as AO13
13		from pyof.v0x04.common.flow_instructions import InstructionApplyAction
14		from pyof.v0x04.common.flow_match import OxmOfbMatchField, OxmTLV, VlanId
15		from pyof.v0x04.common.port import PortNo as Port13
16		from pyof.v0x04.controller2switch.flow_mod import FlowMod as FM13
17		from pyof.v0x04.controller2switch.packet_out import PacketOut as PO13
18
19		from kytos.core import KytosEvent, KytosNApp, log, rest
20		from kytos.core.helpers import listen_to
21		from napps.kytos.of_lldp import constants, settings
22
23
24		class Main(KytosNApp):
25		"""Main OF_LLDP NApp Class."""
26
27		def setup(self):
28		"""Make this NApp run in a loop."""
29		self.vlan_id = None
30		if hasattr(settings, "FLOW_VLAN_VID"):
31		self.vlan_id = settings.FLOW_VLAN_VID
32		self.execute_as_loop(settings.POLLING_TIME)
33
34		def execute(self):
35		"""Send LLDP Packets every 'POLLING_TIME' seconds to all switches."""
36		switches = list(self.controller.switches.values())
37		for switch in switches:
38		try:
39		of_version = switch.connection.protocol.version
40		except AttributeError:
41		of_version = None
42
43		if not switch.is_connected():
44		continue
45
46		if of_version == 0x01:
47		port_type = UBInt16
48		local_port = Port10.OFPP_LOCAL
49		elif of_version == 0x04:
50		port_type = UBInt32
51		local_port = Port13.OFPP_LOCAL
52		else:
53		# skip the current switch with unsupported OF version
54		continue
55
56		interfaces = list(switch.interfaces.values())
57		for interface in interfaces:
58		# Interface marked to receive lldp package
59		if not interface.lldp:
60		# print(f"Disabled LLDP package {interface.id}")
61		continue
62		# Avoid the interface that connects to the controller.
63		if interface.port_number == local_port:
64		continue
65
66		lldp = LLDP()
67		lldp.chassis_id.sub_value = DPID(switch.dpid)
68		lldp.port_id.sub_value = port_type(interface.port_number)
69
70		ethernet = Ethernet()
71		ethernet.ether_type = EtherType.LLDP
72		ethernet.source = interface.address
73		ethernet.destination = constants.LLDP_MULTICAST_MAC
74		ethernet.data = lldp.pack()
75		# self.vlan_id == None will result in a pack with no VLAN.
76		ethernet.vlans.append(VLAN(vid=self.vlan_id))
77
78		packet_out = self._build_lldp_packet_out(
79		of_version,
80		interface.port_number, ethernet.pack())
81
82		if packet_out is None:
83		continue
84
85		event_out = KytosEvent(
86		name='kytos/of_lldp.messages.out.ofpt_packet_out',
87		content={
88		'destination': switch.connection,
89		'message': packet_out})
90		self.controller.buffers.msg_out.put(event_out)
91		# print(f"Sending package to interface {interface.id}")
92		log.debug(
93		"Sending a LLDP PacketOut to the switch %s",
94		switch.dpid)
95
96		msg = '\n'
97		msg += 'Switch: %s (%s)\n'
98		msg += ' Interfaces: %s\n'
99		msg += ' -- LLDP PacketOut --\n'
100		msg += ' Ethernet: eth_type (%s) \| src (%s) \| dst (%s)'
101		msg += '\n'
102		msg += ' LLDP: Switch (%s) \| port (%s)'
103
104		log.debug(
105		msg,
106		switch.connection.address, switch.dpid,
107		switch.interfaces, ethernet.ether_type,
108		ethernet.source, ethernet.destination,
109		switch.dpid, interface.port_number)
110
111		@listen_to('kytos/of_core.handshake.completed')
112		def install_lldp_flow(self, event):
113		"""Install a flow to send LLDP packets to the controller.
114
115		The proactive flow is installed whenever a switch connects.
116
117		Args:
118		event (:class:`~kytos.core.events.KytosEvent`):
119		Event with new switch information.
120
121		"""
122		try:
123		of_version = event.content['switch'].connection.protocol.version
124		except AttributeError:
125		of_version = None
126
127		flow_mod = self._build_lldp_flow_mod(of_version)
128
129		if flow_mod:
130		name = 'kytos/of_lldp.messages.out.ofpt_flow_mod'
131		content = {'destination': event.content['switch'].connection,
132		'message': flow_mod}
133
134		event_out = KytosEvent(name=name, content=content)
135		self.controller.buffers.msg_out.put(event_out)
136
137		@listen_to('kytos/of_core.v0x0[14].messages.in.ofpt_packet_in')
138		def notify_uplink_detected(self, event):
139		"""Dispatch two KytosEvents to notify identified NNI interfaces.
140
141		Args:
142		event (:class:`~kytos.core.events.KytosEvent`):
143		Event with an LLDP packet as data.
144
145		"""
146		ethernet = self._unpack_non_empty(Ethernet, event.message.data)
147		if ethernet.ether_type == EtherType.LLDP:
148		try:
149		lldp = self._unpack_non_empty(LLDP, ethernet.data)
150		dpid = self._unpack_non_empty(DPID, lldp.chassis_id.sub_value)
151		except struct.error:
152		#: If we have a LLDP packet but we cannot unpack it, or the
153		#: unpacked packet does not contain the dpid attribute, then
154		#: we are dealing with a LLDP generated by someone else. Thus
155		#: this packet is not useful for us and we may just ignore it.
156		return
157
158		switch_a = event.source.switch
159		port_a = event.message.in_port
160		switch_b = None
161		port_b = None
162
163		# in_port is currently a UBInt16 in v0x01 and an Int in v0x04.
164		if isinstance(port_a, int):
165		port_a = UBInt32(port_a)
166
167		try:
168		switch_b = self.controller.get_switch_by_dpid(dpid.value)
169		of_version = switch_b.connection.protocol.version
170		port_type = UBInt16 if of_version == 0x01 else UBInt32
171		port_b = self._unpack_non_empty(port_type,
172		lldp.port_id.sub_value)
173		except AttributeError:
174		log.debug("Couldn't find datapath %s.", dpid.value)
175
176		# Return if any of the needed information are not available
177		if not (switch_a and port_a and switch_b and port_b):
178		return
179
180		interface_a = switch_a.get_interface_by_port_no(port_a.value)
181		interface_b = switch_b.get_interface_by_port_no(port_b.value)
182
183		event_out = KytosEvent(name='kytos/of_lldp.interface.is.nni',
184		content={'interface_a': interface_a,
185		'interface_b': interface_b})
186		self.controller.buffers.app.put(event_out)
187
188		def shutdown(self):
189		"""End of the application."""
190		log.debug('Shutting down...')
191
192		@staticmethod
193		def _build_lldp_packet_out(version, port_number, data):
194		"""Build a LLDP PacketOut message.
195
196		Args:
197		version (int): OpenFlow version
198		port_number (int): Switch port number where the packet must be
199		forwarded to.
200		data (bytes): Binary data to be sent through the port.
201
202		Returns:
203		PacketOut message for the specific given OpenFlow version, if it
204		is supported.
205		None if the OpenFlow version is not supported.
206
207		"""
208		if version == 0x01:
209		action_output_class = AO10
210		packet_out_class = PO10
211		elif version == 0x04:
212		action_output_class = AO13
213		packet_out_class = PO13
214		else:
215		log.info('Openflow version %s is not yet supported.', version)
216		return None
217
218		output_action = action_output_class()
219		output_action.port = port_number
220
221		packet_out = packet_out_class()
222		packet_out.data = data
223		packet_out.actions.append(output_action)
224
225		return packet_out
226
227		def _build_lldp_flow_mod(self, version):
228		"""Build a FlodMod message to send LLDP to the controller.
229
230		Args:
231		version (int): OpenFlow version.
232
233		Returns:
234		FlowMod message for the specific given OpenFlow version, if it is
235		supported.
236		None if the OpenFlow version is not supported.
237
238		"""
239		if version == 0x01:
240		flow_mod = FM10()
241		flow_mod.command = FMC.OFPFC_ADD
242		flow_mod.priority = settings.FLOW_PRIORITY
243		flow_mod.match.dl_type = EtherType.LLDP
244		if self.vlan_id:
245		flow_mod.match.dl_vlan = self.vlan_id
246		flow_mod.actions.append(AO10(port=Port10.OFPP_CONTROLLER))
247
248		elif version == 0x04:
249		flow_mod = FM13()
250		flow_mod.command = FMC.OFPFC_ADD
251		flow_mod.priority = settings.FLOW_PRIORITY
252
253		match_lldp = OxmTLV()
254		match_lldp.oxm_field = OxmOfbMatchField.OFPXMT_OFB_ETH_TYPE
255		match_lldp.oxm_value = EtherType.LLDP.to_bytes(2, 'big')
256		flow_mod.match.oxm_match_fields.append(match_lldp)
257
258		if self.vlan_id:
259		match_vlan = OxmTLV()
260		match_vlan.oxm_field = OxmOfbMatchField.OFPXMT_OFB_VLAN_VID
261		vlan_value = self.vlan_id \| VlanId.OFPVID_PRESENT
262		match_vlan.oxm_value = vlan_value.to_bytes(2, 'big')
263		flow_mod.match.oxm_match_fields.append(match_vlan)
264
265		instruction = InstructionApplyAction()
266		instruction.actions.append(AO13(port=Port13.OFPP_CONTROLLER))
267		flow_mod.instructions.append(instruction)
268
269		else:
270		flow_mod = None
271
272		return flow_mod
273
274		@staticmethod
275		def _unpack_non_empty(desired_class, data):
276		"""Unpack data using an instance of desired_class.
277
278		Args:
279		desired_class (class): The class to be used to unpack data.
280		data (bytes): bytes to be unpacked.
281
282		Return:
283		An instance of desired_class class with data unpacked into it.
284
285		Raises:
286		UnpackException if the unpack could not be performed.
287
288		"""
289		obj = desired_class()
290
291		if hasattr(data, 'value'):
292		data = data.value
293
294		obj.unpack(data)
295
296		return obj
297
298		@staticmethod
299		def _get_data(req):
300		"""Get request data."""
301		data = req.get_json()
302		interface_list = []
303		if data:
304		interface_list = data.get('interfaces')
305		return interface_list
306
307		def _get_interfaces(self):
308		"""Get all interfaces."""
309		interfaces = []
310		switches = list(self.controller.switches.values())
311		for switch in switches:
312		interface_list = list(switch.interfaces.values())
313		interfaces = interfaces + interface_list
314		return interfaces
315
316		@staticmethod
317		def _get_interfaces_dict(interfaces):
318		"""Return a dict of interfaces."""
319		interfaces_dic = {}
320		interfaces_dic = {interface.id: interface for interface in interfaces}
321		return interfaces_dic
322
323		def _get_lldp_interfaces(self):
324		"""Get interfaces enabled to receive the LLDP package."""
325		enabled_lldp = []
326		interfaces = self._get_interfaces()
327		enabled_lldp = [inter.id for inter in interfaces if inter.lldp]
328		return enabled_lldp
329
330		@rest('v1/interfaces', methods=['GET'])
331		def get_lldp_interfaces(self):
332		"""Return of the interfaces available to receive the LLDP package."""
333		result = {"interfaces": self._get_lldp_interfaces()}
334		return jsonify(result), 200
335
336	View Code Duplication	@rest('v1/interfaces/disable', methods=['POST'])
		0 ignored issues – show Duplication introduced 2019-07-29 21:12 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
337		def disable_lldp(self):
338		"""Disables an interface to receive an LLDP package."""
339		interfaces_id = self._get_data(request)
340		error_list = []
341		if interfaces_id is None:
342		return jsonify("Bad format."), 401
343		interfaces = self._get_interfaces()
344
345		if not interfaces:
346		return jsonify("No interfaces were found"), 404
347
348		interfaces = self._get_interfaces_dict(interfaces)
349		for interface_id in interfaces_id:
350		if not interface_id:
351		continue
352		if interfaces.get(interface_id):
353		interface = interfaces[interface_id]
354		interface.lldp = False
355		else:
356		# List of interfaces that were not activated.
357		error_list.append(interface_id)
358		if not error_list:
359		return jsonify("All interface passed has been disabled."), 200
360
361		# Return a list of interfaces has not disabled.
362		# Other interfaces passed has enabled.
363		msg_partial_enabled = ""
364		msg_partial_enabled += "Some interfaces have not been disabled. "
365		msg_partial_enabled += "Interface list not disabled:"
366		return jsonify({msg_partial_enabled:
367		error_list}), 400
368
369	View Code Duplication	@rest('v1/interfaces/enable', methods=['POST'])
		0 ignored issues – show Duplication introduced 2019-07-29 21:12 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
370		def enable_lldp(self):
371		"""Enable an interface to receive an LLDP package."""
372		interfaces_id = self._get_data(request)
373		error_list = []
374		if interfaces_id is None:
375		return jsonify("Bad format."), 401
376		interfaces = self._get_interfaces()
377		if not interfaces:
378		return jsonify("No interfaces were found"), 404
379		interfaces = self._get_interfaces_dict(interfaces)
380		for interface_id in interfaces_id:
381		if not interface_id:
382		continue
383		if interfaces.get(interface_id):
384		interface = interfaces[interface_id]
385		interface.lldp = True
386		else:
387		# List of interfaces that were not activated.
388		error_list.append(interface_id)
389		if not error_list:
390		return jsonify("All interface passed has been enabled."), 200
391
392		# Return a list of interfaces has not enabled.
393		# The other interfaces have been activated.
394		msg_partial_enabled = ""
395		msg_partial_enabled += "Some interfaces have not been enabled. "
396		msg_partial_enabled += "Interface list not enabled:"
397		return jsonify({msg_partial_enabled:
398		error_list}), 400
399
400		@rest('v1/interfaces/disable/all', methods=['POST'])
401		def disable_lldp_all(self):
402		"""Disable LLDP on all interfaces."""
403		interfaces = self._get_interfaces()
404		if not interfaces:
405		return jsonify("No interfaces were found"), 404
406		for interface in interfaces:
407		interface.lldp = False
408
409		return jsonify("All interfaces has been disabled"), 200
410
411		@rest('v1/interfaces/enable/all', methods=['POST'])
412		def enable_all_lldp(self):
413		"""Enable LLLDP on all interfaces."""
414		interfaces = self._get_interfaces()
415		if not interfaces:
416		return jsonify("No interfaces were found."), 404
417		for interface in interfaces:
418		interface.lldp = True
419
420		return jsonify("All interfaces has been enabled."), 200
421

kytos / of_lldp

Pull Request — master (#32)

build.main.Main.enable_lldp() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like