build.main.Main.get_time() - Code Metrics - Inspection of "User can now set LLDP polling time by REST API #29" - kytos/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#33)

by Humberto

created 2019-12-20 18:26 UTC

build.main.Main.get_time() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	4
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	1
dl	0
loc	4
ccs	0
cts	0
cp	0
crap	2
rs	10
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
        self.polling_time = settings.POLLING_TIME
        if hasattr(settings, "FLOW_VLAN_VID"):
            self.vlan_id = settings.FLOW_VLAN_VID
        self.execute_as_loop(self.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

    @rest('v1/time', methods=['GET'])
    def get_time(self):
        """Get LLDP polling time."""
        return jsonify({"Polling time in seconds": self.polling_time}), 200

    @rest('v1/time/<time_sec>', methods=['POST'])
    def set_time(self, time_sec):
        """Set LLDP polling time."""
        # pylint: disable=attribute-defined-outside-init

        try:
            self.polling_time = int(time_sec)
            self.execute_as_loop(self.polling_time)
            return jsonify("Polling time has been updated."), 200
        except ValueError:
            return jsonify("Bad format."), 400


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

kytos / of_lldp

Pull Request — master (#33)

build.main.Main.get_time() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like