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

Code Duplication Length = 110-112 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()
        elif isinstance(value, type(self)):
            return value.pack()
        else:
            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()
        elif isinstance(value, type(self)):
            return value.pack()
        else:
            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-131 (lines=112) @@
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		elif isinstance(value, type(self)):
73		return value.pack()
74		else:
75		msg = "{} is not an instance of {}".format(value,
76		type(self).__name__)
77		raise PackException(msg)
78
79		def unpack(self, buff, offset=0):
80		"""Unpack a binary message into this object's attributes.
81
82		Unpack the binary value buff and update this object attributes based
83		on the results. It is an inplace method and it receives the binary data
84		of the message without the header.
85
86		This class' unpack method is like the :meth:`.GenericMessage.unpack`
87		one, except for the ``actions`` attribute which has a length determined
88		by the ``actions_len`` attribute.
89
90		Args:
91		buff (bytes): Binary data package to be unpacked, without the
92		header.
93		offset (int): Where to begin unpacking.
94		"""
95		begin = offset
96		for attribute_name, class_attribute in self.get_class_attributes():
97		if type(class_attribute).__name__ != "Header":
98		attribute = deepcopy(class_attribute)
99		if attribute_name == 'actions':
100		length = self.actions_len.value
101		attribute.unpack(buff[begin:begin+length])
102		else:
103		attribute.unpack(buff, begin)
104		setattr(self, attribute_name, attribute)
105		begin += attribute.get_size()
106
107		def _update_actions_len(self):
108		"""Update the actions_len field based on actions value."""
109		if isinstance(self.actions, ListOfActions):
110		self.actions_len = self.actions.get_size()
111		else:
112		self.actions_len = ListOfActions(self.actions).get_size()
113
114		def _validate_in_port(self):
115		"""Validate in_port attribute.
116
117		A valid port is either:
118
119		* Greater than 0 and less than or equals to Port.OFPP_MAX
120		* One of the valid virtual ports: Port.OFPP_LOCAL,
121		Port.OFPP_CONTROLLER or Port.OFPP_NONE
122
123		Raises:
124		ValidationError: If in_port is an invalid port.
125
126		"""
127		is_valid_range = self.in_port > 0 and self.in_port <= Port.OFPP_MAX
128		is_valid_virtual_in_ports = self.in_port in _VIRT_IN_PORTS
129
130		if (is_valid_range or is_valid_virtual_in_ports) is False:
131		raise ValidationError(f'{self.in_port} is not a valid input port.')
132

		@@ 20-129 (lines=110) @@
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		elif isinstance(value, type(self)):
83		return value.pack()
84		else:
85		msg = "{} is not an instance of {}".format(value,
86		type(self).__name__)
87		raise PackException(msg)
88
89		def unpack(self, buff, offset=0):
90		"""Unpack a binary message into this object's attributes.
91
92		Unpack the binary value buff and update this object attributes based
93		on the results. It is an inplace method and it receives the binary data
94		of the message without the header.
95
96		This class' unpack method is like the :meth:`.GenericMessage.unpack`
97		one, except for the ``actions`` attribute which has a length determined
98		by the ``actions_len`` attribute.
99
100		Args:
101		buff (bytes): Binary data package to be unpacked, without the
102		header.
103		offset (int): Where to begin unpacking.
104		"""
105		begin = offset
106		for attribute_name, class_attribute in self.get_class_attributes():
107		if type(class_attribute).__name__ != "Header":
108		attribute = deepcopy(class_attribute)
109		if attribute_name == 'actions':
110		length = self.actions_len.value
111		attribute.unpack(buff[begin:begin+length])
112		else:
113		attribute.unpack(buff, begin)
114		setattr(self, attribute_name, attribute)
115		begin += attribute.get_size()
116
117		def _update_actions_len(self):
118		"""Update the actions_len field based on actions value."""
119		if isinstance(self.actions, ListOfActions):
120		self.actions_len = self.actions.get_size()
121		else:
122		self.actions_len = ListOfActions(self.actions).get_size()
123
124		def _validate_in_port(self):
125		is_valid_range = self.in_port > 0 and self.in_port <= PortNo.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

kytos / python-openflow

Code Duplication Length = 110-112 lines in 2 locations

pyof/v0x01/controller2switch/packet_out.py 1 location

pyof/v0x04/controller2switch/packet_out.py 1 location