Code Duplication - kytos/python-openflow - Measure and Improve Code Quality continuously with Scrutinizer

Code Duplication Length = 108-110 lines in 2 locations

pyof/v0x01/controller2switch/packet_out.py 1 location


_VIRT_IN_PORTS = (Port.OFPP_LOCAL, Port.OFPP_CONTROLLER, Port.OFPP_NONE)


class PacketOut(GenericMessage):
    """Send packet (controller -> datapath)."""

    #: :class:`~pyof.v0x01.common.header.Header`
    header = Header(message_type=Type.OFPT_PACKET_OUT)
    buffer_id = UBInt32()
    in_port = UBInt16()
    actions_len = UBInt16()
    actions = ListOfActions()
    data = BinaryData()

    def __init__(self, xid=None, buffer_id=NO_BUFFER, in_port=Port.OFPP_NONE,
                 actions=None, data=b''):
        """Create a PacketOut with the optional parameters below.

        Args:
            xid (int): xid of the message header.
            buffer_id (int): ID assigned by datapath (-1 if none).
            in_port (:class:`int` / :class:`~pyof.v0x01.common.phy_port.Port`):
                Packet's input port (:attr:`.Port.OFPP_NONE` if none). Virtual
                ports :attr:`Port.OFPP_IN_PORT`, :attr:`Port.OFPP_TABLE`,
                :attr:`Port.OFPP_NORMAL`, :attr:`Port.OFPP_FLOOD`, and
                :attr:`Port.OFPP_ALL` cannot be used as input port.
            actions (~pyof.v0x01.common.action.ListOfActions): List of Actions.
            data (bytes): Packet data. The length is inferred from the length
                field in the header. (Only meaningful if ``buffer_id`` == -1).
        """
        super().__init__(xid)
        self.buffer_id = buffer_id
        self.in_port = in_port
        self.actions = [] if actions is None else actions
        self.data = data

    def validate(self):
        """Validate the entire message."""
        if not super().is_valid():
            raise ValidationError()
        self._validate_in_port()

    def is_valid(self):
        """Answer if this message is valid."""
        try:
            self.validate()
            return True
        except ValidationError:
            return False

    def pack(self, value=None):
        """Update the action_len attribute and call super's pack."""
        if value is None:
            self._update_actions_len()
            return super().pack()
        if isinstance(value, type(self)):
            return value.pack()
        msg = "{} is not an instance of {}".format(value, type(self).__name__)
        raise PackException(msg)

    def unpack(self, buff, offset=0):
        """Unpack a binary message into this object's attributes.

        Unpack the binary value *buff* and update this object attributes based
        on the results. It is an inplace method and it receives the binary data
        of the message **without the header**.

        This class' unpack method is like the :meth:`.GenericMessage.unpack`
        one, except for the ``actions`` attribute which has a length determined
        by the ``actions_len`` attribute.

        Args:
            buff (bytes): Binary data package to be unpacked, without the
                header.
            offset (int): Where to begin unpacking.
        """
        begin = offset
        for attribute_name, class_attribute in self.get_class_attributes():
            if type(class_attribute).__name__ != "Header":
                attribute = deepcopy(class_attribute)
                if attribute_name == 'actions':
                    length = self.actions_len.value
                    attribute.unpack(buff[begin:begin+length])
                else:
                    attribute.unpack(buff, begin)
                setattr(self, attribute_name, attribute)
                begin += attribute.get_size()

    def _update_actions_len(self):
        """Update the actions_len field based on actions value."""
        if isinstance(self.actions, ListOfActions):
            self.actions_len = self.actions.get_size()
        else:
            self.actions_len = ListOfActions(self.actions).get_size()

    def _validate_in_port(self):
        """Validate in_port attribute.

        A valid port is either:

            * Greater than 0 and less than or equals to Port.OFPP_MAX
            * One of the valid virtual ports: Port.OFPP_LOCAL,
              Port.OFPP_CONTROLLER or Port.OFPP_NONE

        Raises:
            ValidationError: If in_port is an invalid port.

        """
        is_valid_range = self.in_port > 0 and self.in_port <= Port.OFPP_MAX
        is_valid_virtual_in_ports = self.in_port in _VIRT_IN_PORTS

        if (is_valid_range or is_valid_virtual_in_ports) is False:
            raise ValidationError(f'{self.in_port} is not a valid input port.')


pyof/v0x04/controller2switch/packet_out.py 1 location


_VIRT_IN_PORTS = (PortNo.OFPP_LOCAL, PortNo.OFPP_CONTROLLER, PortNo.OFPP_ANY)


class PacketOut(GenericMessage):
    """Send packet (controller -> datapath)."""

    #: Openflow :class:`~pyof.v0x04.common.header.Header`
    header = Header(message_type=Type.OFPT_PACKET_OUT)
    #: ID assigned by datapath (OFP_NO_BUFFER if none).
    buffer_id = UBInt32()
    #: Packet’s input port or OFPP_CONTROLLER.
    in_port = UBInt32()
    #: Size of action array in bytes.
    actions_len = UBInt16()
    #: Padding
    pad = Pad(6)
    #: Action List.
    actions = ListOfActions()
    #: Packet data. The length is inferred from the length field in the header.
    #:    (Only meaningful if buffer_id == -1.)
    data = BinaryData()

    def __init__(self, xid=None, buffer_id=UBINT32_MAX_VALUE,
                 in_port=PortNo.OFPP_CONTROLLER, actions=None,
                 data=b''):
        """Create a PacketOut with the optional parameters below.

        Args:
            xid (int): xid of the message header.
            buffer_id (int): ID assigned by datapath (-1 if none). In this case
                UBINT32_MAX_VALUE is -1 for the field.
            in_port (:class:`int`, :class:`~pyof.v0x04.common.port.Port`):
                Packet's input port (:attr:`Port.OFPP_NONE` if none).
                Virtual ports OFPP_IN_PORT, OFPP_TABLE, OFPP_NORMAL,
                OFPP_FLOOD, and OFPP_ALL cannot be used as input port.
            actions (:class:`~pyof.v0x04.common.action.ListOfActions`):
                List of Action instances.
            data (bytes): Packet data. The length is inferred from the length
                field in the header. (Only meaningful if ``buffer_id`` == -1).
        """
        super().__init__(xid)
        self.buffer_id = buffer_id
        self.in_port = in_port
        self.actions = [] if actions is None else actions
        self.data = data

    def validate(self):
        """Validate the entire message."""
        if not super().is_valid():
            raise ValidationError()
        self._validate_in_port()

    def is_valid(self):
        """Answer if this message is valid."""
        try:
            self.validate()
            return True
        except ValidationError:
            return False

    def pack(self, value=None):
        """Update the action_len attribute and call super's pack."""
        if value is None:
            self._update_actions_len()
            return super().pack()
        if isinstance(value, type(self)):
            return value.pack()
        msg = "{} is not an instance of {}".format(value, type(self).__name__)
        raise PackException(msg)

    def unpack(self, buff, offset=0):
        """Unpack a binary message into this object's attributes.

        Unpack the binary value *buff* and update this object attributes based
        on the results. It is an inplace method and it receives the binary data
        of the message **without the header**.

        This class' unpack method is like the :meth:`.GenericMessage.unpack`
        one, except for the ``actions`` attribute which has a length determined
        by the ``actions_len`` attribute.

        Args:
            buff (bytes): Binary data package to be unpacked, without the
                header.
            offset (int): Where to begin unpacking.
        """
        begin = offset
        for attribute_name, class_attribute in self.get_class_attributes():
            if type(class_attribute).__name__ != "Header":
                attribute = deepcopy(class_attribute)
                if attribute_name == 'actions':
                    length = self.actions_len.value
                    attribute.unpack(buff[begin:begin+length])
                else:
                    attribute.unpack(buff, begin)
                setattr(self, attribute_name, attribute)
                begin += attribute.get_size()

    def _update_actions_len(self):
        """Update the actions_len field based on actions value."""
        if isinstance(self.actions, ListOfActions):
            self.actions_len = self.actions.get_size()
        else:
            self.actions_len = ListOfActions(self.actions).get_size()

    def _validate_in_port(self):
        is_valid_range = self.in_port > 0 and self.in_port <= PortNo.OFPP_MAX
        is_valid_virtual_in_ports = self.in_port in _VIRT_IN_PORTS

        if (is_valid_range or is_valid_virtual_in_ports) is False:
            raise ValidationError(f'{self.in_port} is not a valid input port.')


		@@ 20-129 (lines=110) @@
17		_VIRT_IN_PORTS = (Port.OFPP_LOCAL, Port.OFPP_CONTROLLER, Port.OFPP_NONE)
18
19
20		class PacketOut(GenericMessage):
21		"""Send packet (controller -> datapath)."""
22
23		#: :class:`~pyof.v0x01.common.header.Header`
24		header = Header(message_type=Type.OFPT_PACKET_OUT)
25		buffer_id = UBInt32()
26		in_port = UBInt16()
27		actions_len = UBInt16()
28		actions = ListOfActions()
29		data = BinaryData()
30
31		def __init__(self, xid=None, buffer_id=NO_BUFFER, in_port=Port.OFPP_NONE,
32		actions=None, data=b''):
33		"""Create a PacketOut with the optional parameters below.
34
35		Args:
36		xid (int): xid of the message header.
37		buffer_id (int): ID assigned by datapath (-1 if none).
38		in_port (:class:`int` / :class:`~pyof.v0x01.common.phy_port.Port`):
39		Packet's input port (:attr:`.Port.OFPP_NONE` if none). Virtual
40		ports :attr:`Port.OFPP_IN_PORT`, :attr:`Port.OFPP_TABLE`,
41		:attr:`Port.OFPP_NORMAL`, :attr:`Port.OFPP_FLOOD`, and
42		:attr:`Port.OFPP_ALL` cannot be used as input port.
43		actions (~pyof.v0x01.common.action.ListOfActions): List of Actions.
44		data (bytes): Packet data. The length is inferred from the length
45		field in the header. (Only meaningful if ``buffer_id`` == -1).
46		"""
47		super().__init__(xid)
48		self.buffer_id = buffer_id
49		self.in_port = in_port
50		self.actions = [] if actions is None else actions
51		self.data = data
52
53		def validate(self):
54		"""Validate the entire message."""
55		if not super().is_valid():
56		raise ValidationError()
57		self._validate_in_port()
58
59		def is_valid(self):
60		"""Answer if this message is valid."""
61		try:
62		self.validate()
63		return True
64		except ValidationError:
65		return False
66
67		def pack(self, value=None):
68		"""Update the action_len attribute and call super's pack."""
69		if value is None:
70		self._update_actions_len()
71		return super().pack()
72		if isinstance(value, type(self)):
73		return value.pack()
74		msg = "{} is not an instance of {}".format(value, type(self).__name__)
75		raise PackException(msg)
76
77		def unpack(self, buff, offset=0):
78		"""Unpack a binary message into this object's attributes.
79
80		Unpack the binary value buff and update this object attributes based
81		on the results. It is an inplace method and it receives the binary data
82		of the message without the header.
83
84		This class' unpack method is like the :meth:`.GenericMessage.unpack`
85		one, except for the ``actions`` attribute which has a length determined
86		by the ``actions_len`` attribute.
87
88		Args:
89		buff (bytes): Binary data package to be unpacked, without the
90		header.
91		offset (int): Where to begin unpacking.
92		"""
93		begin = offset
94		for attribute_name, class_attribute in self.get_class_attributes():
95		if type(class_attribute).__name__ != "Header":
96		attribute = deepcopy(class_attribute)
97		if attribute_name == 'actions':
98		length = self.actions_len.value
99		attribute.unpack(buff[begin:begin+length])
100		else:
101		attribute.unpack(buff, begin)
102		setattr(self, attribute_name, attribute)
103		begin += attribute.get_size()
104
105		def _update_actions_len(self):
106		"""Update the actions_len field based on actions value."""
107		if isinstance(self.actions, ListOfActions):
108		self.actions_len = self.actions.get_size()
109		else:
110		self.actions_len = ListOfActions(self.actions).get_size()
111
112		def _validate_in_port(self):
113		"""Validate in_port attribute.
114
115		A valid port is either:
116
117		* Greater than 0 and less than or equals to Port.OFPP_MAX
118		* One of the valid virtual ports: Port.OFPP_LOCAL,
119		Port.OFPP_CONTROLLER or Port.OFPP_NONE
120
121		Raises:
122		ValidationError: If in_port is an invalid port.
123
124		"""
125		is_valid_range = self.in_port > 0 and self.in_port <= Port.OFPP_MAX
126		is_valid_virtual_in_ports = self.in_port in _VIRT_IN_PORTS
127
128		if (is_valid_range or is_valid_virtual_in_ports) is False:
129		raise ValidationError(f'{self.in_port} is not a valid input port.')
130

		@@ 20-127 (lines=108) @@
17		_VIRT_IN_PORTS = (PortNo.OFPP_LOCAL, PortNo.OFPP_CONTROLLER, PortNo.OFPP_ANY)
18
19
20		class PacketOut(GenericMessage):
21		"""Send packet (controller -> datapath)."""
22
23		#: Openflow :class:`~pyof.v0x04.common.header.Header`
24		header = Header(message_type=Type.OFPT_PACKET_OUT)
25		#: ID assigned by datapath (OFP_NO_BUFFER if none).
26		buffer_id = UBInt32()
27		#: Packet’s input port or OFPP_CONTROLLER.
28		in_port = UBInt32()
29		#: Size of action array in bytes.
30		actions_len = UBInt16()
31		#: Padding
32		pad = Pad(6)
33		#: Action List.
34		actions = ListOfActions()
35		#: Packet data. The length is inferred from the length field in the header.
36		#: (Only meaningful if buffer_id == -1.)
37		data = BinaryData()
38
39		def __init__(self, xid=None, buffer_id=UBINT32_MAX_VALUE,
40		in_port=PortNo.OFPP_CONTROLLER, actions=None,
41		data=b''):
42		"""Create a PacketOut with the optional parameters below.
43
44		Args:
45		xid (int): xid of the message header.
46		buffer_id (int): ID assigned by datapath (-1 if none). In this case
47		UBINT32_MAX_VALUE is -1 for the field.
48		in_port (:class:`int`, :class:`~pyof.v0x04.common.port.Port`):
49		Packet's input port (:attr:`Port.OFPP_NONE` if none).
50		Virtual ports OFPP_IN_PORT, OFPP_TABLE, OFPP_NORMAL,
51		OFPP_FLOOD, and OFPP_ALL cannot be used as input port.
52		actions (:class:`~pyof.v0x04.common.action.ListOfActions`):
53		List of Action instances.
54		data (bytes): Packet data. The length is inferred from the length
55		field in the header. (Only meaningful if ``buffer_id`` == -1).
56		"""
57		super().__init__(xid)
58		self.buffer_id = buffer_id
59		self.in_port = in_port
60		self.actions = [] if actions is None else actions
61		self.data = data
62
63		def validate(self):
64		"""Validate the entire message."""
65		if not super().is_valid():
66		raise ValidationError()
67		self._validate_in_port()
68
69		def is_valid(self):
70		"""Answer if this message is valid."""
71		try:
72		self.validate()
73		return True
74		except ValidationError:
75		return False
76
77		def pack(self, value=None):
78		"""Update the action_len attribute and call super's pack."""
79		if value is None:
80		self._update_actions_len()
81		return super().pack()
82		if isinstance(value, type(self)):
83		return value.pack()
84		msg = "{} is not an instance of {}".format(value, type(self).__name__)
85		raise PackException(msg)
86
87		def unpack(self, buff, offset=0):
88		"""Unpack a binary message into this object's attributes.
89
90		Unpack the binary value buff and update this object attributes based
91		on the results. It is an inplace method and it receives the binary data
92		of the message without the header.
93
94		This class' unpack method is like the :meth:`.GenericMessage.unpack`
95		one, except for the ``actions`` attribute which has a length determined
96		by the ``actions_len`` attribute.
97
98		Args:
99		buff (bytes): Binary data package to be unpacked, without the
100		header.
101		offset (int): Where to begin unpacking.
102		"""
103		begin = offset
104		for attribute_name, class_attribute in self.get_class_attributes():
105		if type(class_attribute).__name__ != "Header":
106		attribute = deepcopy(class_attribute)
107		if attribute_name == 'actions':
108		length = self.actions_len.value
109		attribute.unpack(buff[begin:begin+length])
110		else:
111		attribute.unpack(buff, begin)
112		setattr(self, attribute_name, attribute)
113		begin += attribute.get_size()
114
115		def _update_actions_len(self):
116		"""Update the actions_len field based on actions value."""
117		if isinstance(self.actions, ListOfActions):
118		self.actions_len = self.actions.get_size()
119		else:
120		self.actions_len = ListOfActions(self.actions).get_size()
121
122		def _validate_in_port(self):
123		is_valid_range = self.in_port > 0 and self.in_port <= PortNo.OFPP_MAX
124		is_valid_virtual_in_ports = self.in_port in _VIRT_IN_PORTS
125
126		if (is_valid_range or is_valid_virtual_in_ports) is False:
127		raise ValidationError(f'{self.in_port} is not a valid input port.')
128

kytos / python-openflow

Code Duplication Length = 108-110 lines in 2 locations

pyof/v0x01/controller2switch/packet_out.py 1 location

pyof/v0x04/controller2switch/packet_out.py 1 location