pyof.v0x04.common.action.ActionSetField._update_length() - Code Metrics - Inspection of "Customize unpack to Bucket class to support dynami..." - kytos/python-openflow - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( ebaa29...24e02e )

by Humberto

created 2019-10-18 18:25 UTC

ActionSetField._update_length() A

↳ Parent: pyof.v0x04.common.action

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	2.0185

Importance

Changes

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

"""Defines actions that may be associated with flows packets."""
# System imports
from enum import IntEnum
from math import ceil

# Local source tree imports
from pyof.foundation.base import GenericStruct
from pyof.foundation.basic_types import (
    BinaryData, FixedTypeList, Pad, UBInt8, UBInt16, UBInt32)
from pyof.v0x04.common.flow_match import OxmTLV

# Third-party imports

__all__ = ('ActionExperimenter', 'ActionGroup', 'ActionHeader',
           'ActionCopyTTLIn', 'ActionCopyTTLOut', 'ActionDecMPLSTTL',
           'ActionSetMPLSTTL', 'ActionDecNWTTL', 'ActionSetNWTTL',
           'ActionOutput', 'ActionPopMPLS', 'ActionPopPBB', 'ActionPopVLAN',
           'ActionPush', 'ActionSetField', 'ActionSetQueue', 'ActionType',
           'ControllerMaxLen', 'ListOfActions')

# Enums


class ActionType(IntEnum):
    """Actions associated with flows and packets."""

    #: Output to switch port.
    OFPAT_OUTPUT = 0
    #: Copy TTL "outwards" -- from next-to-outermost to outermost
    OFPAT_COPY_TTL_OUT = 11
    #: Copy TTL "inwards" -- from outermost to next-to-outermost
    OFPAT_COPY_TTL_IN = 12
    #: MPLS TTL
    OFPAT_SET_MPLS_TTL = 15
    #: Decrement MPLS TTL
    OFPAT_DEC_MPLS_TTL = 16
    #: Push a new VLAN tag
    OFPAT_PUSH_VLAN = 17
    #: Pop the outer VLAN tag
    OFPAT_POP_VLAN = 18
    #: Push a new MPLS tag
    OFPAT_PUSH_MPLS = 19
    #: Pop the outer MPLS tag
    OFPAT_POP_MPLS = 20
    #: Set queue id when outputting to a port
    OFPAT_SET_QUEUE = 21
    #: Apply group.
    OFPAT_GROUP = 22
    #: IP TTL.
    OFPAT_SET_NW_TTL = 23
    #: Decrement IP TTL.
    OFPAT_DEC_NW_TTL = 24
    #: Set a header field using OXM TLV format.
    OFPAT_SET_FIELD = 25
    #: Push a new PBB service tag (I-TAG)
    OFPAT_PUSH_PBB = 26
    #: Pop the outer PBB service tag (I-TAG)
    OFPAT_POP_PBB = 27
    #: Experimenter type
    OFPAT_EXPERIMENTER = 0xffff


class ControllerMaxLen(IntEnum):
    """A max_len of OFPCML_NO_BUFFER means not to buffer.

    The packet should be sent.
    """

    #: maximum max_len value which can be used to request a specific byte
    #:     length.
    OFPCML_MAX = 0xffe5
    #: indicates that no buffering should be applied and the whole packet is to
    #:     be sent to the controller.
    OFPCML_NO_BUFFER = 0xffff


# Classes


class ActionHeader(GenericStruct):
    """Action header that is common to all actions.

    The length includes the header and any padding used to make the action
    64-bit aligned.
    NB: The length of an action *must* always be a multiple of eight.
    """

    #: One of OFPAT_*.
    action_type = UBInt16(enum_ref=ActionType)
    #: Length of action, including this header. This is the length of actions,
    #:    including any padding to make it 64-bit aligned.
    length = UBInt16()
    # Pad for 64-bit alignment.
    # This should not be implemented, as each action type has its own padding.
    # pad = Pad(4)

    _allowed_types = ()

    def __init__(self, action_type=None, length=None):
        """Create an ActionHeader with the optional parameters below.

        Args:
            action_type (~pyof.v0x04.common.action.ActionType):
                The type of the action.
            length (int): Length of action, including this header.
        """
        super().__init__()
        self.action_type = action_type
        self.length = length

    def get_size(self, value=None):
        """Return the action length including the padding (multiple of 8)."""
        if isinstance(value, ActionHeader):
            return value.get_size()
        elif value is None:
            current_size = super().get_size()
            return ceil(current_size / 8) * 8
        raise ValueError(f'Invalid value "{value}" for Action*.get_size()')

    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.

        Args:
            buff (bytes): Binary data package to be unpacked.
            offset (int): Where to begin unpacking.

        Raises:
            Exception: If there is a struct unpacking error.

        """
        self.action_type = UBInt16(enum_ref=ActionType)
        self.action_type.unpack(buff, offset)

        self.length = UBInt16()
        self.length.unpack(buff, offset=offset+2)

        for cls in ActionHeader.__subclasses__():
            if self.action_type.value in cls.get_allowed_types():
                self.__class__ = cls
                break

        super().unpack(buff[:offset+self.length], offset)

    @classmethod
    def get_allowed_types(cls):
        """Return allowed types for the class."""
        return cls._allowed_types


class ActionExperimenter(ActionHeader):
    """Action structure for OFPAT_EXPERIMENTER."""

    experimenter = UBInt32()
    body = BinaryData()

    _allowed_types = ActionType.OFPAT_EXPERIMENTER,

    def __init__(self, length=None, experimenter=None, body=None):
        """Create ActionExperimenterHeader with the optional parameters below.

        Args:
            experimenter (int): The experimenter field is the Experimenter ID,
                which takes the same form as in struct ofp_experimenter.
            body(bytes): The body of the experimenter. It is vendor-defined,
                so it is left as it is.
        """
        super().__init__(action_type=ActionType.OFPAT_EXPERIMENTER)
        self.length = length
        self.experimenter = experimenter
        self.body = body


class ActionGroup(ActionHeader):
    """Action structure for OFPAT_GROUP."""

    group_id = UBInt32()

    _allowed_types = ActionType.OFPAT_GROUP,

    def __init__(self, group_id=None):
        """Create an ActionGroup with the optional parameters below.

        Args:
            group_id (int): The group_id indicates the group used to process
                this packet. The set of buckets to apply depends on the group
                type.
        """
        super().__init__(action_type=ActionType.OFPAT_GROUP, length=8)
        self.group_id = group_id


class ActionDecMPLSTTL(ActionHeader):
    """Action structure for OFPAT_DEC_MPLS_TTL."""

    pad = Pad(4)

    _allowed_types = ActionType.OFPAT_DEC_MPLS_TTL,

    def __init__(self):
        """Create an ActionDecMPLSTTL."""
        super().__init__(action_type=ActionType.OFPAT_DEC_MPLS_TTL, length=8)


class ActionSetMPLSTTL(ActionHeader):
    """Action structure for OFPAT_SET_MPLS_TTL."""

    mpls_ttl = UBInt8()
    pad = Pad(3)

    _allowed_types = ActionType.OFPAT_SET_MPLS_TTL,

    def __init__(self, mpls_ttl=None):
        """Create an ActionSetMPLSTTL with the optional parameters below.

        Args:
            mpls_ttl (int): The mpls_ttl field is the MPLS TTL to set.
        """
        super().__init__(action_type=ActionType.OFPAT_SET_MPLS_TTL, length=8)
        self.mpls_ttl = mpls_ttl


class ActionCopyTTLIn(ActionHeader):
    """Action structure for OFPAT_COPY_TTL_IN."""

    pad = Pad(4)

    _allowed_types = ActionType.OFPAT_COPY_TTL_IN,

    def __init__(self):
        """Create an ActionCopyTTLIn."""
        super().__init__(action_type=ActionType.OFPAT_COPY_TTL_IN, length=8)


class ActionCopyTTLOut(ActionHeader):
    """Action structure for OFPAT_COPY_TTL_OUT."""

    pad = Pad(4)

    _allowed_types = ActionType.OFPAT_COPY_TTL_OUT,

    def __init__(self):
        """Create an ActionCopyTTLOut."""
        super().__init__(action_type=ActionType.OFPAT_COPY_TTL_OUT, length=8)


class ActionPopVLAN(ActionHeader):
    """Action structure for OFPAT_POP_VLAN."""

    pad = Pad(4)

    _allowed_types = ActionType.OFPAT_POP_VLAN,

    def __init__(self):
        """Create an ActionPopVLAN."""
        super().__init__(action_type=ActionType.OFPAT_POP_VLAN, length=8)


class ActionPopPBB(ActionHeader):
    """Action structure for OFPAT_POP_PBB."""

    pad = Pad(4)

    _allowed_types = ActionType.OFPAT_POP_PBB,

    def __init__(self):
        """Create an ActionPopPBB."""
        super().__init__(action_type=ActionType.OFPAT_POP_PBB, length=8)


class ActionDecNWTTL(ActionHeader):
    """Action structure for OFPAT_DEC_NW_TTL."""

    pad = Pad(4)

    _allowed_types = ActionType.OFPAT_DEC_NW_TTL,

    def __init__(self):
        """Create a ActionDecNWTTL."""
        super().__init__(action_type=ActionType.OFPAT_DEC_NW_TTL, length=8)


class ActionSetNWTTL(ActionHeader):
    """Action structure for OFPAT_SET_NW_TTL."""

    nw_ttl = UBInt8()
    pad = Pad(3)

    _allowed_types = ActionType.OFPAT_SET_NW_TTL,

    def __init__(self, nw_ttl=None):
        """Create an ActionSetNWTTL with the optional parameters below.

        Args:
            nw_ttl (int): the TTL address to set in the IP header.
        """
        super().__init__(action_type=ActionType.OFPAT_SET_NW_TTL, length=8)
        self.nw_ttl = nw_ttl


class ActionOutput(ActionHeader):
    """Defines the actions output.

    Action structure for :attr:`ActionType.OFPAT_OUTPUT`, which sends packets
    out :attr:`port`. When the :attr:`port` is the
    :attr:`.Port.OFPP_CONTROLLER`, :attr:`max_length` indicates the max number
    of bytes to send. A :attr:`max_length` of zero means no bytes of the packet
    should be sent.
    """

    port = UBInt32()
    max_length = UBInt16()
    pad = Pad(6)

    _allowed_types = ActionType.OFPAT_OUTPUT,

    def __init__(self, port=None,
                 max_length=ControllerMaxLen.OFPCML_NO_BUFFER):
        """Create a ActionOutput with the optional parameters below.

        Args:
            port (:class:`Port` or :class:`int`): Output port.
            max_length (int): Max length to send to controller.
        """
        super().__init__(action_type=ActionType.OFPAT_OUTPUT, length=16)
        self.port = port
        self.max_length = max_length


class ActionPopMPLS(ActionHeader):
    """Action structure for OFPAT_POP_MPLS."""

    ethertype = UBInt16()
    pad = Pad(2)

    _allowed_types = ActionType.OFPAT_POP_MPLS,

    def __init__(self, ethertype=None):
        """Create an ActionPopMPLS with the optional parameters below.

        Args:
            ethertype (int): indicates the Ethertype of the payload.
        """
        super().__init__(action_type=ActionType.OFPAT_POP_MPLS)
        self.ethertype = ethertype


class ActionPush(ActionHeader):
    """Action structure for OFPAT_PUSH_[VLAN/MPLS/PBB]."""

    ethertype = UBInt16()
    pad = Pad(2)

    _allowed_types = (ActionType.OFPAT_PUSH_VLAN, ActionType.OFPAT_PUSH_MPLS,
                      ActionType.OFPAT_PUSH_PBB)

    def __init__(self, action_type=None, ethertype=None):
        """Create a ActionPush with the optional parameters below.

        Args:
            action_type (:class:`ActionType`): indicates which tag will be
                pushed (VLAN, MPLS, PBB).
            ethertype (int): indicates the Ethertype of the new tag.
        """
        super().__init__(action_type, length=8)
        self.ethertype = ethertype


class ActionSetField(ActionHeader):
    """Action structure for OFPAT_SET_FIELD."""

    field = OxmTLV()

    _allowed_types = ActionType.OFPAT_SET_FIELD,

    def __init__(self, field=None):
        """Create a ActionSetField with the optional parameters below.

        Args:
            length (int): length padded to 64 bits, followed by exactly
                          oxm_len bytes containing a single OXM TLV, then
                          exactly ((oxm_len + 4) + 7)/8*8 - (oxm_len + 4)
                          (between 0 and 7) bytes of all-zero bytes
            field (:class:`OxmTLV`): OXM field and value.
        """
        super().__init__(action_type=ActionType.OFPAT_SET_FIELD)
        self.field = OxmTLV() if field is None else field

    def pack(self, value=None):
        """Pack this structure updating the length and padding it."""
        self._update_length()
        packet = super().pack()
        return self._complete_last_byte(packet)

    def _update_length(self):
        """Update the length field of the struct."""
        action_length = 4 + len(self.field.pack())
        overflow = action_length % 8
        self.length = action_length
        if overflow:
            self.length = action_length + 8 - overflow

    def _complete_last_byte(self, packet):
        """Pad until the packet length is a multiple of 8 (bytes)."""
        padded_size = self.length
        padding_bytes = padded_size - len(packet)
        if padding_bytes > 0:
            packet += Pad(padding_bytes).pack()
        return packet


class ActionSetQueue(ActionHeader):
    """Action structure for OFPAT_SET_QUEUE."""

    queue_id = UBInt32()

    _allowed_types = ActionType.OFPAT_SET_QUEUE,

    def __init__(self, queue_id=None):
        """Create an ActionSetQueue with the optional parameters below.

        Args:
            queue_id (int): The queue_id send packets to given queue on port.
        """
        super().__init__(action_type=ActionType.OFPAT_SET_QUEUE, length=8)
        self.queue_id = queue_id


class ListOfActions(FixedTypeList):
    """List of actions.

    Represented by instances of ActionHeader and used on ActionHeader objects.
    """

    def __init__(self, items=None):
        """Create a ListOfActions with the optional parameters below.

        Args:
            items (~pyof.v0x04.common.action.ActionHeader):
                Instance or a list of instances.
        """
        super().__init__(pyof_class=ActionHeader, items=items)


1			"""Defines actions that may be associated with flows packets."""
2			# System imports
3	1		from enum import IntEnum
4	1		from math import ceil
5
6			# Local source tree imports
7	1		from pyof.foundation.base import GenericStruct
8	1		from pyof.foundation.basic_types import (
9			BinaryData, FixedTypeList, Pad, UBInt8, UBInt16, UBInt32)
10	1		from pyof.v0x04.common.flow_match import OxmTLV
11
12			# Third-party imports
13
14	1		__all__ = ('ActionExperimenter', 'ActionGroup', 'ActionHeader',
15			'ActionCopyTTLIn', 'ActionCopyTTLOut', 'ActionDecMPLSTTL',
16			'ActionSetMPLSTTL', 'ActionDecNWTTL', 'ActionSetNWTTL',
17			'ActionOutput', 'ActionPopMPLS', 'ActionPopPBB', 'ActionPopVLAN',
18			'ActionPush', 'ActionSetField', 'ActionSetQueue', 'ActionType',
19			'ControllerMaxLen', 'ListOfActions')
20
21			# Enums
22
23
24	1		class ActionType(IntEnum):
25			"""Actions associated with flows and packets."""
26
27			#: Output to switch port.
28	1		OFPAT_OUTPUT = 0
29			#: Copy TTL "outwards" -- from next-to-outermost to outermost
30	1		OFPAT_COPY_TTL_OUT = 11
31			#: Copy TTL "inwards" -- from outermost to next-to-outermost
32	1		OFPAT_COPY_TTL_IN = 12
33			#: MPLS TTL
34	1		OFPAT_SET_MPLS_TTL = 15
35			#: Decrement MPLS TTL
36	1		OFPAT_DEC_MPLS_TTL = 16
37			#: Push a new VLAN tag
38	1		OFPAT_PUSH_VLAN = 17
39			#: Pop the outer VLAN tag
40	1		OFPAT_POP_VLAN = 18
41			#: Push a new MPLS tag
42	1		OFPAT_PUSH_MPLS = 19
43			#: Pop the outer MPLS tag
44	1		OFPAT_POP_MPLS = 20
45			#: Set queue id when outputting to a port
46	1		OFPAT_SET_QUEUE = 21
47			#: Apply group.
48	1		OFPAT_GROUP = 22
49			#: IP TTL.
50	1		OFPAT_SET_NW_TTL = 23
51			#: Decrement IP TTL.
52	1		OFPAT_DEC_NW_TTL = 24
53			#: Set a header field using OXM TLV format.
54	1		OFPAT_SET_FIELD = 25
55			#: Push a new PBB service tag (I-TAG)
56	1		OFPAT_PUSH_PBB = 26
57			#: Pop the outer PBB service tag (I-TAG)
58	1		OFPAT_POP_PBB = 27
59			#: Experimenter type
60	1		OFPAT_EXPERIMENTER = 0xffff
61
62
63	1		class ControllerMaxLen(IntEnum):
64			"""A max_len of OFPCML_NO_BUFFER means not to buffer.
65
66			The packet should be sent.
67			"""
68
69			#: maximum max_len value which can be used to request a specific byte
70			#: length.
71	1		OFPCML_MAX = 0xffe5
72			#: indicates that no buffering should be applied and the whole packet is to
73			#: be sent to the controller.
74	1		OFPCML_NO_BUFFER = 0xffff
75
76
77			# Classes
78
79
80	1		class ActionHeader(GenericStruct):
81			"""Action header that is common to all actions.
82
83			The length includes the header and any padding used to make the action
84			64-bit aligned.
85			NB: The length of an action must always be a multiple of eight.
86			"""
87
88			#: One of OFPAT_*.
89	1		action_type = UBInt16(enum_ref=ActionType)
90			#: Length of action, including this header. This is the length of actions,
91			#: including any padding to make it 64-bit aligned.
92	1		length = UBInt16()
93			# Pad for 64-bit alignment.
94			# This should not be implemented, as each action type has its own padding.
95			# pad = Pad(4)
96
97	1		_allowed_types = ()
98
99	1		def __init__(self, action_type=None, length=None):
100			"""Create an ActionHeader with the optional parameters below.
101
102			Args:
103			action_type (~pyof.v0x04.common.action.ActionType):
104			The type of the action.
105			length (int): Length of action, including this header.
106			"""
107	1		super().__init__()
108	1		self.action_type = action_type
109	1		self.length = length
110
111	1		def get_size(self, value=None):
112			"""Return the action length including the padding (multiple of 8)."""
113	1		if isinstance(value, ActionHeader):
114			return value.get_size()
115	1		elif value is None:
116	1		current_size = super().get_size()
117	1		return ceil(current_size / 8) * 8
118			raise ValueError(f'Invalid value "{value}" for Action*.get_size()')
119
120	1	View Code Duplication	def unpack(self, buff, offset=0):
			0 ignored issues – show Duplication introduced 2019-08-27 13:20 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
121			"""Unpack a binary message into this object's attributes.
122
123			Unpack the binary value buff and update this object attributes based
124			on the results.
125
126			Args:
127			buff (bytes): Binary data package to be unpacked.
128			offset (int): Where to begin unpacking.
129
130			Raises:
131			Exception: If there is a struct unpacking error.
132
133			"""
134	1		self.action_type = UBInt16(enum_ref=ActionType)
135	1		self.action_type.unpack(buff, offset)
136
137	1		self.length = UBInt16()
138	1		self.length.unpack(buff, offset=offset+2)
139
140	1		for cls in ActionHeader.__subclasses__():
141	1		if self.action_type.value in cls.get_allowed_types():
142	1		self.__class__ = cls
143	1		break
144
145	1		super().unpack(buff[:offset+self.length], offset)
146
147	1		@classmethod
148			def get_allowed_types(cls):
149			"""Return allowed types for the class."""
150	1		return cls._allowed_types
151
152
153	1		class ActionExperimenter(ActionHeader):
154			"""Action structure for OFPAT_EXPERIMENTER."""
155
156	1		experimenter = UBInt32()
157	1		body = BinaryData()
158
159	1		_allowed_types = ActionType.OFPAT_EXPERIMENTER,
160
161	1		def __init__(self, length=None, experimenter=None, body=None):
162			"""Create ActionExperimenterHeader with the optional parameters below.
163
164			Args:
165			experimenter (int): The experimenter field is the Experimenter ID,
166			which takes the same form as in struct ofp_experimenter.
167			body(bytes): The body of the experimenter. It is vendor-defined,
168			so it is left as it is.
169			"""
170	1		super().__init__(action_type=ActionType.OFPAT_EXPERIMENTER)
171	1		self.length = length
172	1		self.experimenter = experimenter
173	1		self.body = body
174
175
176	1		class ActionGroup(ActionHeader):
177			"""Action structure for OFPAT_GROUP."""
178
179	1		group_id = UBInt32()
180
181	1		_allowed_types = ActionType.OFPAT_GROUP,
182
183	1		def __init__(self, group_id=None):
184			"""Create an ActionGroup with the optional parameters below.
185
186			Args:
187			group_id (int): The group_id indicates the group used to process
188			this packet. The set of buckets to apply depends on the group
189			type.
190			"""
191			super().__init__(action_type=ActionType.OFPAT_GROUP, length=8)
192			self.group_id = group_id
193
194
195	1		class ActionDecMPLSTTL(ActionHeader):
196			"""Action structure for OFPAT_DEC_MPLS_TTL."""
197
198	1		pad = Pad(4)
199
200	1		_allowed_types = ActionType.OFPAT_DEC_MPLS_TTL,
201
202	1		def __init__(self):
203			"""Create an ActionDecMPLSTTL."""
204			super().__init__(action_type=ActionType.OFPAT_DEC_MPLS_TTL, length=8)
205
206
207	1		class ActionSetMPLSTTL(ActionHeader):
208			"""Action structure for OFPAT_SET_MPLS_TTL."""
209
210	1		mpls_ttl = UBInt8()
211	1		pad = Pad(3)
212
213	1		_allowed_types = ActionType.OFPAT_SET_MPLS_TTL,
214
215	1		def __init__(self, mpls_ttl=None):
216			"""Create an ActionSetMPLSTTL with the optional parameters below.
217
218			Args:
219			mpls_ttl (int): The mpls_ttl field is the MPLS TTL to set.
220			"""
221			super().__init__(action_type=ActionType.OFPAT_SET_MPLS_TTL, length=8)
222			self.mpls_ttl = mpls_ttl
223
224
225	1		class ActionCopyTTLIn(ActionHeader):
226			"""Action structure for OFPAT_COPY_TTL_IN."""
227
228	1		pad = Pad(4)
229
230	1		_allowed_types = ActionType.OFPAT_COPY_TTL_IN,
231
232	1		def __init__(self):
233			"""Create an ActionCopyTTLIn."""
234			super().__init__(action_type=ActionType.OFPAT_COPY_TTL_IN, length=8)
235
236
237	1		class ActionCopyTTLOut(ActionHeader):
238			"""Action structure for OFPAT_COPY_TTL_OUT."""
239
240	1		pad = Pad(4)
241
242	1		_allowed_types = ActionType.OFPAT_COPY_TTL_OUT,
243
244	1		def __init__(self):
245			"""Create an ActionCopyTTLOut."""
246			super().__init__(action_type=ActionType.OFPAT_COPY_TTL_OUT, length=8)
247
248
249	1		class ActionPopVLAN(ActionHeader):
250			"""Action structure for OFPAT_POP_VLAN."""
251
252	1		pad = Pad(4)
253
254	1		_allowed_types = ActionType.OFPAT_POP_VLAN,
255
256	1		def __init__(self):
257			"""Create an ActionPopVLAN."""
258			super().__init__(action_type=ActionType.OFPAT_POP_VLAN, length=8)
259
260
261	1		class ActionPopPBB(ActionHeader):
262			"""Action structure for OFPAT_POP_PBB."""
263
264	1		pad = Pad(4)
265
266	1		_allowed_types = ActionType.OFPAT_POP_PBB,
267
268	1		def __init__(self):
269			"""Create an ActionPopPBB."""
270			super().__init__(action_type=ActionType.OFPAT_POP_PBB, length=8)
271
272
273	1		class ActionDecNWTTL(ActionHeader):
274			"""Action structure for OFPAT_DEC_NW_TTL."""
275
276	1		pad = Pad(4)
277
278	1		_allowed_types = ActionType.OFPAT_DEC_NW_TTL,
279
280	1		def __init__(self):
281			"""Create a ActionDecNWTTL."""
282			super().__init__(action_type=ActionType.OFPAT_DEC_NW_TTL, length=8)
283
284
285	1		class ActionSetNWTTL(ActionHeader):
286			"""Action structure for OFPAT_SET_NW_TTL."""
287
288	1		nw_ttl = UBInt8()
289	1		pad = Pad(3)
290
291	1		_allowed_types = ActionType.OFPAT_SET_NW_TTL,
292
293	1		def __init__(self, nw_ttl=None):
294			"""Create an ActionSetNWTTL with the optional parameters below.
295
296			Args:
297			nw_ttl (int): the TTL address to set in the IP header.
298			"""
299			super().__init__(action_type=ActionType.OFPAT_SET_NW_TTL, length=8)
300			self.nw_ttl = nw_ttl
301
302
303	1		class ActionOutput(ActionHeader):
304			"""Defines the actions output.
305
306			Action structure for :attr:`ActionType.OFPAT_OUTPUT`, which sends packets
307			out :attr:`port`. When the :attr:`port` is the
308			:attr:`.Port.OFPP_CONTROLLER`, :attr:`max_length` indicates the max number
309			of bytes to send. A :attr:`max_length` of zero means no bytes of the packet
310			should be sent.
311			"""
312
313	1		port = UBInt32()
314	1		max_length = UBInt16()
315	1		pad = Pad(6)
316
317	1		_allowed_types = ActionType.OFPAT_OUTPUT,
318
319	1		def __init__(self, port=None,
320			max_length=ControllerMaxLen.OFPCML_NO_BUFFER):
321			"""Create a ActionOutput with the optional parameters below.
322
323			Args:
324			port (:class:`Port` or :class:`int`): Output port.
325			max_length (int): Max length to send to controller.
326			"""
327			super().__init__(action_type=ActionType.OFPAT_OUTPUT, length=16)
328			self.port = port
329			self.max_length = max_length
330
331
332	1		class ActionPopMPLS(ActionHeader):
333			"""Action structure for OFPAT_POP_MPLS."""
334
335	1		ethertype = UBInt16()
336	1		pad = Pad(2)
337
338	1		_allowed_types = ActionType.OFPAT_POP_MPLS,
339
340	1		def __init__(self, ethertype=None):
341			"""Create an ActionPopMPLS with the optional parameters below.
342
343			Args:
344			ethertype (int): indicates the Ethertype of the payload.
345			"""
346			super().__init__(action_type=ActionType.OFPAT_POP_MPLS)
347			self.ethertype = ethertype
348
349
350	1		class ActionPush(ActionHeader):
351			"""Action structure for OFPAT_PUSH_[VLAN/MPLS/PBB]."""
352
353	1		ethertype = UBInt16()
354	1		pad = Pad(2)
355
356	1		_allowed_types = (ActionType.OFPAT_PUSH_VLAN, ActionType.OFPAT_PUSH_MPLS,
357			ActionType.OFPAT_PUSH_PBB)
358
359	1		def __init__(self, action_type=None, ethertype=None):
360			"""Create a ActionPush with the optional parameters below.
361
362			Args:
363			action_type (:class:`ActionType`): indicates which tag will be
364			pushed (VLAN, MPLS, PBB).
365			ethertype (int): indicates the Ethertype of the new tag.
366			"""
367			super().__init__(action_type, length=8)
368			self.ethertype = ethertype
369
370
371	1		class ActionSetField(ActionHeader):
372			"""Action structure for OFPAT_SET_FIELD."""
373
374	1		field = OxmTLV()
375
376	1		_allowed_types = ActionType.OFPAT_SET_FIELD,
377
378	1		def __init__(self, field=None):
379			"""Create a ActionSetField with the optional parameters below.
380
381			Args:
382			length (int): length padded to 64 bits, followed by exactly
383			oxm_len bytes containing a single OXM TLV, then
384			exactly ((oxm_len + 4) + 7)/8*8 - (oxm_len + 4)
385			(between 0 and 7) bytes of all-zero bytes
386			field (:class:`OxmTLV`): OXM field and value.
387			"""
388	1		super().__init__(action_type=ActionType.OFPAT_SET_FIELD)
389	1		self.field = OxmTLV() if field is None else field
390
391	1		def pack(self, value=None):
392			"""Pack this structure updating the length and padding it."""
393	1		self._update_length()
394	1		packet = super().pack()
395	1		return self._complete_last_byte(packet)
396
397	1		def _update_length(self):
398			"""Update the length field of the struct."""
399	1		action_length = 4 + len(self.field.pack())
400	1		overflow = action_length % 8
401	1		self.length = action_length
402	1		if overflow:
403			self.length = action_length + 8 - overflow
404
405	1		def _complete_last_byte(self, packet):
406			"""Pad until the packet length is a multiple of 8 (bytes)."""
407	1		padded_size = self.length
408	1		padding_bytes = padded_size - len(packet)
409	1		if padding_bytes > 0:
410			packet += Pad(padding_bytes).pack()
411	1		return packet
412
413
414	1		class ActionSetQueue(ActionHeader):
415			"""Action structure for OFPAT_SET_QUEUE."""
416
417	1		queue_id = UBInt32()
418
419	1		_allowed_types = ActionType.OFPAT_SET_QUEUE,
420
421	1		def __init__(self, queue_id=None):
422			"""Create an ActionSetQueue with the optional parameters below.
423
424			Args:
425			queue_id (int): The queue_id send packets to given queue on port.
426			"""
427			super().__init__(action_type=ActionType.OFPAT_SET_QUEUE, length=8)
428			self.queue_id = queue_id
429
430
431	1		class ListOfActions(FixedTypeList):
432			"""List of actions.
433
434			Represented by instances of ActionHeader and used on ActionHeader objects.
435			"""
436
437	1		def __init__(self, items=None):
438			"""Create a ListOfActions with the optional parameters below.
439
440			Args:
441			items (~pyof.v0x04.common.action.ActionHeader):
442			Instance or a list of instances.
443			"""
444			super().__init__(pyof_class=ActionHeader, items=items)
445

kytos / python-openflow

Push — master ( ebaa29...24e02e )

ActionSetField._update_length() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like