pyof.v0x04.common.flow_instructions.ListOfInstruction.__init__() - Code Metrics - kytos/python-openflow - Measure and Improve Code Quality continuously with Scrutinizer

ListOfInstruction.init() A
last analyzed 2021-05-31 20:00 UTC

↳ Parent: pyof.v0x04.common.flow_instructions

Complexity

Conditions

Size

Total Lines	8
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

Metric	Value
eloc	2
dl	0
loc	8
ccs	2
cts	2
cp	1
rs	10
c	0
b	0
f	0
cc	1
nop	2
crap	1

"""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()
        if isinstance(value, type(self)):
            return value.pack()
        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			if isinstance(value, type(self)):
90			return value.pack()
91			msg = "{} is not an instance of {}".format(value, type(self).__name__)
92			raise PackException(msg)
93
94	1		def update_length(self):
95			"""Update length attribute."""
96	1		self.length = self.get_size()
97
98	1		def unpack(self, buff=None, offset=0):
99			"""Unpack buff into this object.
100
101			This method will convert a binary data into a readable value according
102			to the attribute format.
103
104			Args:
105			buff (bytes): Binary buffer.
106			offset (int): Where to begin unpacking.
107
108			Raises:
109			:exc:`~.exceptions.UnpackException`: If unpack fails.
110
111			"""
112	1		instruction_type = UBInt16(enum_ref=InstructionType)
113	1		instruction_type.unpack(buff, offset)
114	1		self.__class__ = InstructionType(instruction_type.value).find_class()
115
116	1		length = UBInt16()
117	1		length.unpack(buff, offset=offset+2)
118
119	1		super().unpack(buff[:offset+length.value], offset)
120
121
122	1		class InstructionApplyAction(Instruction):
123			"""Instruction structure for OFPIT_APPLY_ACTIONS.
124
125			The :attr:`~actions` field is treated as a list, and the actions are
126			applied to the packet in-order.
127			"""
128
129			#: Align to 64-bits
130	1		pad = Pad(4)
131			#: Actions associated with OFPIT_APPLY_ACTIONS
132	1		actions = ListOfActions()
133
134	1		def __init__(self, actions=None):
135			"""Create a InstructionApplyAction with the optional parameters below.
136
137			Args:
138			actions (:class:`~.actions.ListOfActions`):
139			Actions associated with OFPIT_APPLY_ACTIONS.
140			"""
141	1		super().__init__(InstructionType.OFPIT_APPLY_ACTIONS)
142	1		self.actions = actions if actions else []
143
144
145	1		class InstructionClearAction(Instruction):
146			"""Instruction structure for OFPIT_CLEAR_ACTIONS.
147
148			This structure does not contain any actions.
149			"""
150
151			#: Align to 64-bits
152	1		pad = Pad(4)
153			#: OFPIT_CLEAR_ACTIONS does not have any action on the list of actions.
154	1		actions = ListOfActions()
155
156	1		def __init__(self, actions=None):
157			"""Create a InstructionClearAction with the optional parameters below.
158
159			Args:
160			actions (:class:`~.actions.ListOfActions`):
161			Actions associated with OFPIT_CLEAR_ACTIONS.
162			"""
163			super().__init__(InstructionType.OFPIT_CLEAR_ACTIONS)
164			self.actions = actions if actions else []
165
166
167	1		class InstructionGotoTable(Instruction):
168			"""Instruction structure for OFPIT_GOTO_TABLE."""
169
170			#: Set next table in the lookup pipeline.
171	1		table_id = UBInt8()
172			#: Pad to 64 bits.
173	1		pad = Pad(3)
174
175	1		def __init__(self, table_id=Meter.OFPM_ALL):
176			"""Create a InstructionGotoTable with the optional parameters below.
177
178			Args:
179			length (int): Length of this struct in bytes.
180			table_id (int): set next table in the lookup pipeline.
181			"""
182			super().__init__(InstructionType.OFPIT_GOTO_TABLE)
183			self.table_id = table_id
184
185
186	1		class InstructionMeter(Instruction):
187			"""Instruction structure for OFPIT_METER.
188
189			meter_id indicates which meter to apply on the packet.
190			"""
191
192			#: Meter instance.
193	1		meter_id = UBInt32()
194
195	1		def __init__(self, meter_id=Meter.OFPM_ALL):
196			"""Create a InstructionMeter with the optional parameters below.
197
198			Args:
199			meter_id (int): Meter instance.
200			"""
201			super().__init__(InstructionType.OFPIT_METER)
202			self.meter_id = meter_id
203
204
205	1		class InstructionWriteAction(Instruction):
206			"""Instruction structure for OFPIT_WRITE_ACTIONS.
207
208			The actions field must be treated as a SET, so the actions are not
209			repeated.
210			"""
211
212			#: Align to 64-bits
213	1		pad = Pad(4)
214			#: Actions associated with OFPIT_WRITE_ACTIONS
215	1		actions = ListOfActions()
216
217	1		def __init__(self, actions=None):
218			"""Create a InstructionWriteAction with the optional parameters below.
219
220			Args:
221			actions (:class:`~.actions.ListOfActions`):
222			Actions associated with OFPIT_WRITE_ACTIONS.
223			"""
224			super().__init__(InstructionType.OFPIT_WRITE_ACTIONS)
225			self.actions = actions if actions else []
226
227
228	1		class InstructionWriteMetadata(Instruction):
229			"""Instruction structure for OFPIT_WRITE_METADATA."""
230
231			#: Align to 64-bits
232	1		pad = Pad(4)
233			#: Metadata value to write
234	1		metadata = UBInt64()
235			#: Metadata write bitmask
236	1		metadata_mask = UBInt64()
237
238	1		def __init__(self, metadata=0, metadata_mask=0):
239			"""Create InstructionWriteMetadata with the optional parameters below.
240
241			Args:
242			metadata (int): Metadata value to write.
243			metadata_mask (int): Metadata write bitmask.
244			"""
245			super().__init__(InstructionType.OFPIT_WRITE_METADATA)
246			self.metadata = metadata
247			self.metadata_mask = metadata_mask
248
249
250	1		class ListOfInstruction(FixedTypeList):
251			"""List of Instructions.
252
253			Represented by instances of Instruction.
254			"""
255
256	1		def __init__(self, items=None):
257			"""Create ListOfInstruction with the optional parameters below.
258
259			Args:
260			items (:class:`~pyof.v0x04.common.flow_instructions.Instruction`):
261			Instance or a list of instances.
262			"""
263			super().__init__(pyof_class=Instruction, items=items)
264

kytos / python-openflow

ListOfInstruction.__init__() A last analyzed 2021-05-31 20:00 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

ListOfInstruction.init() A
last analyzed 2021-05-31 20:00 UTC