build.main.Main.setup() - Code Metrics - Inspection of "Create unit tests" - kytos/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#42)

by Gleyberson

created 2020-06-03 13:29 UTC

build.main.Main.setup() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	6
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	2

Importance

Changes

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

"""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 packet
                if not interface.lldp:
                    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 packet 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)
                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()  # Valid format { "interfaces": [...] }
        return data.get('interfaces', [])

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

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

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

    @rest('v1/interfaces', methods=['GET'])
    def get_lldp_interfaces(self):
        """Return all the interfaces that have LLDP traffic enabled."""
        return jsonify({"interfaces": self._get_lldp_interfaces()}), 200

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

    def disable_lldp(self):
        """Disables an interface to receive LLDP packets."""
        interface_ids = self._get_data(request)
        error_list = []  # List of interfaces that were not activated.
        interface_ids = filter(None, interface_ids)
        interfaces = self._get_interfaces()
        if not interfaces:
            return jsonify("No interfaces were found."), 404
        interfaces = self._get_interfaces_dict(interfaces)
        for id_ in interface_ids:
            interface = interfaces.get(id_)
            if interface:
                interface.lldp = False
            else:
                error_list.append(id_)
        if not error_list:
            return jsonify(
                "All the requested interfaces have been disabled."), 200

        # Return a list of interfaces that couldn't be disabled
        msg_error = "Some interfaces couldn't be found and deactivated: "
        return jsonify({msg_error:
                        error_list}), 400

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

    def enable_lldp(self):
        """Enable an interface to receive LLDP packets."""
        interface_ids = self._get_data(request)
        error_list = []  # List of interfaces that were not activated.
        interface_ids = filter(None, interface_ids)
        interfaces = self._get_interfaces()
        if not interfaces:
            return jsonify("No interfaces were found."), 404
        interfaces = self._get_interfaces_dict(interfaces)
        for id_ in interface_ids:
            interface = interfaces.get(id_)
            if interface:
                interface.lldp = True
            else:
                error_list.append(id_)
        if not error_list:
            return jsonify(
                "All the requested interfaces have been enabled."), 200

        # Return a list of interfaces that couldn't be enabled
        msg_error = "Some interfaces couldn't be found and activated: "
        return jsonify({msg_error:
                        error_list}), 400


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

kytos / of_lldp

Pull Request — master (#42)

build.main.Main.setup() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like