pyof.v0x04.common.flow_instructions.InstructionWriteMetadata.__init__() - Code Metrics - Inspection of "Create some raw files and of_1.3 tests" - kytos/python-openflow - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#577)

by Gleyberson

created 2019-10-18 17:45 UTC

InstructionWriteMetadata.init() A

↳ Parent: pyof.v0x04.common.flow_instructions

Complexity

Conditions

Size

Total Lines	10
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.4218

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	10
ccs	1
cts	4
cp	0.25
rs	10
c	0
b	0
f	0
cc	1
nop	3
crap	1.4218

"""Flow instructions to be executed.

The flow instructions associated with a flow table entry are executed when a
flow matches the entry.
"""
# System imports
from enum import IntEnum

# Local source tree imports
from pyof.foundation.base import GenericStruct
from pyof.foundation.basic_types import (
    FixedTypeList, Pad, UBInt8, UBInt16, UBInt32, UBInt64)
from pyof.foundation.exceptions import PackException
from pyof.v0x04.common.action import ListOfActions
from pyof.v0x04.controller2switch.meter_mod import Meter

# Third-party imports


__all__ = ('InstructionApplyAction', 'InstructionClearAction',
           'InstructionGotoTable', 'InstructionMeter', 'InstructionType',
           'InstructionWriteAction', 'InstructionWriteMetadata',
           'ListOfInstruction')

# Enums


class InstructionType(IntEnum):
    """List of instructions that are currently defined."""

    #: Setup the next table in the lookup pipeline
    OFPIT_GOTO_TABLE = 1
    #: Setup the metadata field for use later in pipeline
    OFPIT_WRITE_METADATA = 2
    #: Write the action(s) onto the datapath action set
    OFPIT_WRITE_ACTIONS = 3
    #: Applies the action(s) immediately
    OFPIT_APPLY_ACTIONS = 4
    #: Clears all actions from the datapath action set
    OFPIT_CLEAR_ACTIONS = 5
    #: Apply meter (rate limiter)
    OFPIT_METER = 6
    #: Experimenter instruction
    OFPIT_EXPERIMENTER = 0xFFFF

    def find_class(self):
        """Return a class related with this type."""
        classes = {1: InstructionGotoTable, 2: InstructionWriteMetadata,
                   3: InstructionWriteAction, 4: InstructionApplyAction,
                   5: InstructionClearAction, 6: InstructionMeter}
        return classes.get(self.value, None)


# Classes

class Instruction(GenericStruct):
    """Generic Instruction class.

    This class represents a Generic Instruction that can be instanciated as
    'InstructionApplyAction', 'InstructionClearAction', 'InstructionGotoTable',
    'InstructionMeter', 'InstructionWriteAction', 'InstructionWriteMetadata'.
    """

    instruction_type = UBInt16(enum_ref=InstructionType)
    length = UBInt16()

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

        Args:
            instruction_type(InstructionType): Type of instruction.
        """
        super().__init__()
        self.instruction_type = instruction_type

    def pack(self, value=None):

        """Update the length and pack the massege into binary data.

        Returns:
            bytes: A binary data that represents the Message.

        Raises:
            Exception: If there are validation errors.

        """
        if value is None:
            self.update_length()
            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 update_length(self):
        """Update length attribute."""
        self.length = self.get_size()

    def unpack(self, buff=None, offset=0):
        """Unpack *buff* into this object.

        This method will convert a binary data into a readable value according
        to the attribute format.

        Args:
            buff (bytes): Binary buffer.
            offset (int): Where to begin unpacking.

        Raises:
            :exc:`~.exceptions.UnpackException`: If unpack fails.

        """
        instruction_type = UBInt16(enum_ref=InstructionType)
        instruction_type.unpack(buff, offset)
        self.__class__ = InstructionType(instruction_type.value).find_class()

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

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


class InstructionApplyAction(Instruction):
    """Instruction structure for OFPIT_APPLY_ACTIONS.

    The :attr:`~actions` field is treated as a list, and the actions are
    applied to the packet in-order.
    """

    #: Align to 64-bits
    pad = Pad(4)
    #: Actions associated with OFPIT_APPLY_ACTIONS
    actions = ListOfActions()

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

        Args:
            actions (:class:`~.actions.ListOfActions`):
                Actions associated with OFPIT_APPLY_ACTIONS.
        """
        super().__init__(InstructionType.OFPIT_APPLY_ACTIONS)
        self.actions = actions if actions else []


class InstructionClearAction(Instruction):
    """Instruction structure for OFPIT_CLEAR_ACTIONS.

    This structure does not contain any actions.
    """

    #: Align to 64-bits
    pad = Pad(4)
    #: OFPIT_CLEAR_ACTIONS does not have any action on the list of actions.
    actions = ListOfActions()

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

        Args:
            actions (:class:`~.actions.ListOfActions`):
                Actions associated with OFPIT_CLEAR_ACTIONS.
        """
        super().__init__(InstructionType.OFPIT_CLEAR_ACTIONS)
        self.actions = actions if actions else []


class InstructionGotoTable(Instruction):
    """Instruction structure for OFPIT_GOTO_TABLE."""

    #: Set next table in the lookup pipeline.
    table_id = UBInt8()
    #: Pad to 64 bits.
    pad = Pad(3)

    def __init__(self, table_id=Meter.OFPM_ALL):
        """Create a InstructionGotoTable with the optional parameters below.

        Args:
            length (int): Length of this struct in bytes.
            table_id (int): set next table in the lookup pipeline.
        """
        super().__init__(InstructionType.OFPIT_GOTO_TABLE)
        self.table_id = table_id


class InstructionMeter(Instruction):
    """Instruction structure for OFPIT_METER.

    meter_id indicates which meter to apply on the packet.
    """

    #: Meter instance.
    meter_id = UBInt32()

    def __init__(self, meter_id=Meter.OFPM_ALL):
        """Create a InstructionMeter with the optional parameters below.

        Args:
            meter_id (int): Meter instance.
        """
        super().__init__(InstructionType.OFPIT_METER)
        self.meter_id = meter_id


class InstructionWriteAction(Instruction):
    """Instruction structure for OFPIT_WRITE_ACTIONS.

    The actions field must be treated as a SET, so the actions are not
    repeated.
    """

    #: Align to 64-bits
    pad = Pad(4)
    #: Actions associated with OFPIT_WRITE_ACTIONS
    actions = ListOfActions()

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

        Args:
            actions (:class:`~.actions.ListOfActions`):
                Actions associated with OFPIT_WRITE_ACTIONS.
        """
        super().__init__(InstructionType.OFPIT_WRITE_ACTIONS)
        self.actions = actions if actions else []


class InstructionWriteMetadata(Instruction):
    """Instruction structure for OFPIT_WRITE_METADATA."""

    #: Align to 64-bits
    pad = Pad(4)
    #: Metadata value to write
    metadata = UBInt64()
    #: Metadata write bitmask
    metadata_mask = UBInt64()

    def __init__(self, metadata=0, metadata_mask=0):
        """Create InstructionWriteMetadata with the optional parameters below.

        Args:
            metadata (int): Metadata value to write.
            metadata_mask (int): Metadata write bitmask.
        """
        super().__init__(InstructionType.OFPIT_WRITE_METADATA)
        self.metadata = metadata
        self.metadata_mask = metadata_mask


class ListOfInstruction(FixedTypeList):
    """List of Instructions.

    Represented by instances of Instruction.
    """

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

        Args:
            items (:class:`~pyof.v0x04.common.flow_instructions.Instruction`):
                Instance or a list of instances.
        """
        super().__init__(pyof_class=Instruction, items=items)


1			"""Flow instructions to be executed.
2
3			The flow instructions associated with a flow table entry are executed when a
4			flow matches the entry.
5			"""
6			# System imports
7	1		from enum import IntEnum
8
9			# Local source tree imports
10	1		from pyof.foundation.base import GenericStruct
11	1		from pyof.foundation.basic_types import (
12			FixedTypeList, Pad, UBInt8, UBInt16, UBInt32, UBInt64)
13	1		from pyof.foundation.exceptions import PackException
14	1		from pyof.v0x04.common.action import ListOfActions
15	1		from pyof.v0x04.controller2switch.meter_mod import Meter
16
17			# Third-party imports
18
19
20	1		__all__ = ('InstructionApplyAction', 'InstructionClearAction',
21			'InstructionGotoTable', 'InstructionMeter', 'InstructionType',
22			'InstructionWriteAction', 'InstructionWriteMetadata',
23			'ListOfInstruction')
24
25			# Enums
26
27
28	1		class InstructionType(IntEnum):
29			"""List of instructions that are currently defined."""
30
31			#: Setup the next table in the lookup pipeline
32	1		OFPIT_GOTO_TABLE = 1
33			#: Setup the metadata field for use later in pipeline
34	1		OFPIT_WRITE_METADATA = 2
35			#: Write the action(s) onto the datapath action set
36	1		OFPIT_WRITE_ACTIONS = 3
37			#: Applies the action(s) immediately
38	1		OFPIT_APPLY_ACTIONS = 4
39			#: Clears all actions from the datapath action set
40	1		OFPIT_CLEAR_ACTIONS = 5
41			#: Apply meter (rate limiter)
42	1		OFPIT_METER = 6
43			#: Experimenter instruction
44	1		OFPIT_EXPERIMENTER = 0xFFFF
45
46	1		def find_class(self):
47			"""Return a class related with this type."""
48	1		classes = {1: InstructionGotoTable, 2: InstructionWriteMetadata,
49			3: InstructionWriteAction, 4: InstructionApplyAction,
50			5: InstructionClearAction, 6: InstructionMeter}
51	1		return classes.get(self.value, None)
52
53
54			# Classes
55
56	1		class Instruction(GenericStruct):
57			"""Generic Instruction class.
58
59			This class represents a Generic Instruction that can be instanciated as
60			'InstructionApplyAction', 'InstructionClearAction', 'InstructionGotoTable',
61			'InstructionMeter', 'InstructionWriteAction', 'InstructionWriteMetadata'.
62			"""
63
64	1		instruction_type = UBInt16(enum_ref=InstructionType)
65	1		length = UBInt16()
66
67	1		def __init__(self, instruction_type=None):
68			"""Create a Instruction with the optional parameters below.
69
70			Args:
71			instruction_type(InstructionType): Type of instruction.
72			"""
73	1		super().__init__()
74	1		self.instruction_type = instruction_type
75
76	1	View Code Duplication	def pack(self, value=None):
			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...
77			"""Update the length and pack the massege into binary data.
78
79			Returns:
80			bytes: A binary data that represents the Message.
81
82			Raises:
83			Exception: If there are validation errors.
84
85			"""
86	1		if value is None:
87	1		self.update_length()
88	1		return super().pack()
89			elif isinstance(value, type(self)):
90			return value.pack()
91			else:
92			msg = "{} is not an instance of {}".format(value,
93			type(self).__name__)
94			raise PackException(msg)
95
96	1		def update_length(self):
97			"""Update length attribute."""
98	1		self.length = self.get_size()
99
100	1		def unpack(self, buff=None, offset=0):
101			"""Unpack buff into this object.
102
103			This method will convert a binary data into a readable value according
104			to the attribute format.
105
106			Args:
107			buff (bytes): Binary buffer.
108			offset (int): Where to begin unpacking.
109
110			Raises:
111			:exc:`~.exceptions.UnpackException`: If unpack fails.
112
113			"""
114	1		instruction_type = UBInt16(enum_ref=InstructionType)
115	1		instruction_type.unpack(buff, offset)
116	1		self.__class__ = InstructionType(instruction_type.value).find_class()
117
118	1		length = UBInt16()
119	1		length.unpack(buff, offset=offset+2)
120
121	1		super().unpack(buff[:offset+length.value], offset)
122
123
124	1		class InstructionApplyAction(Instruction):
125			"""Instruction structure for OFPIT_APPLY_ACTIONS.
126
127			The :attr:`~actions` field is treated as a list, and the actions are
128			applied to the packet in-order.
129			"""
130
131			#: Align to 64-bits
132	1		pad = Pad(4)
133			#: Actions associated with OFPIT_APPLY_ACTIONS
134	1		actions = ListOfActions()
135
136	1		def __init__(self, actions=None):
137			"""Create a InstructionApplyAction with the optional parameters below.
138
139			Args:
140			actions (:class:`~.actions.ListOfActions`):
141			Actions associated with OFPIT_APPLY_ACTIONS.
142			"""
143	1		super().__init__(InstructionType.OFPIT_APPLY_ACTIONS)
144	1		self.actions = actions if actions else []
145
146
147	1		class InstructionClearAction(Instruction):
148			"""Instruction structure for OFPIT_CLEAR_ACTIONS.
149
150			This structure does not contain any actions.
151			"""
152
153			#: Align to 64-bits
154	1		pad = Pad(4)
155			#: OFPIT_CLEAR_ACTIONS does not have any action on the list of actions.
156	1		actions = ListOfActions()
157
158	1		def __init__(self, actions=None):
159			"""Create a InstructionClearAction with the optional parameters below.
160
161			Args:
162			actions (:class:`~.actions.ListOfActions`):
163			Actions associated with OFPIT_CLEAR_ACTIONS.
164			"""
165			super().__init__(InstructionType.OFPIT_CLEAR_ACTIONS)
166			self.actions = actions if actions else []
167
168
169	1		class InstructionGotoTable(Instruction):
170			"""Instruction structure for OFPIT_GOTO_TABLE."""
171
172			#: Set next table in the lookup pipeline.
173	1		table_id = UBInt8()
174			#: Pad to 64 bits.
175	1		pad = Pad(3)
176
177	1		def __init__(self, table_id=Meter.OFPM_ALL):
178			"""Create a InstructionGotoTable with the optional parameters below.
179
180			Args:
181			length (int): Length of this struct in bytes.
182			table_id (int): set next table in the lookup pipeline.
183			"""
184			super().__init__(InstructionType.OFPIT_GOTO_TABLE)
185			self.table_id = table_id
186
187
188	1		class InstructionMeter(Instruction):
189			"""Instruction structure for OFPIT_METER.
190
191			meter_id indicates which meter to apply on the packet.
192			"""
193
194			#: Meter instance.
195	1		meter_id = UBInt32()
196
197	1		def __init__(self, meter_id=Meter.OFPM_ALL):
198			"""Create a InstructionMeter with the optional parameters below.
199
200			Args:
201			meter_id (int): Meter instance.
202			"""
203			super().__init__(InstructionType.OFPIT_METER)
204			self.meter_id = meter_id
205
206
207	1		class InstructionWriteAction(Instruction):
208			"""Instruction structure for OFPIT_WRITE_ACTIONS.
209
210			The actions field must be treated as a SET, so the actions are not
211			repeated.
212			"""
213
214			#: Align to 64-bits
215	1		pad = Pad(4)
216			#: Actions associated with OFPIT_WRITE_ACTIONS
217	1		actions = ListOfActions()
218
219	1		def __init__(self, actions=None):
220			"""Create a InstructionWriteAction with the optional parameters below.
221
222			Args:
223			actions (:class:`~.actions.ListOfActions`):
224			Actions associated with OFPIT_WRITE_ACTIONS.
225			"""
226			super().__init__(InstructionType.OFPIT_WRITE_ACTIONS)
227			self.actions = actions if actions else []
228
229
230	1		class InstructionWriteMetadata(Instruction):
231			"""Instruction structure for OFPIT_WRITE_METADATA."""
232
233			#: Align to 64-bits
234	1		pad = Pad(4)
235			#: Metadata value to write
236	1		metadata = UBInt64()
237			#: Metadata write bitmask
238	1		metadata_mask = UBInt64()
239
240	1		def __init__(self, metadata=0, metadata_mask=0):
241			"""Create InstructionWriteMetadata with the optional parameters below.
242
243			Args:
244			metadata (int): Metadata value to write.
245			metadata_mask (int): Metadata write bitmask.
246			"""
247			super().__init__(InstructionType.OFPIT_WRITE_METADATA)
248			self.metadata = metadata
249			self.metadata_mask = metadata_mask
250
251
252	1		class ListOfInstruction(FixedTypeList):
253			"""List of Instructions.
254
255			Represented by instances of Instruction.
256			"""
257
258	1		def __init__(self, items=None):
259			"""Create ListOfInstruction with the optional parameters below.
260
261			Args:
262			items (:class:`~pyof.v0x04.common.flow_instructions.Instruction`):
263			Instance or a list of instances.
264			"""
265			super().__init__(pyof_class=Instruction, items=items)
266

kytos / python-openflow

Pull Request — master (#577)

InstructionWriteMetadata.__init__() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

InstructionWriteMetadata.init() A