MeterFlags

Test Failed

Pull Request — master (#505)

by macartur

created 2017-11-22 19:27 UTC

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	11
Duplicated Lines	0 %

Test Coverage

Coverage

Importance

Changes

Metric	Value
wmc	0
dl	0
loc	11
ccs	0
cts	5
cp	0
c	0
b	0
f	0

"""Meter modification message."""
from enum import IntEnum

from pyof.foundation.base import GenericBitMask, GenericMessage
from pyof.foundation.basic_types import (
    FixedTypeList, GenericStruct, Pad, UBInt8, UBInt16, UBInt32)
from pyof.v0x04.common.header import Header, Type

__all__ = ('MeterMod', 'Meter', 'MeterModCommand', 'MeterFlags',
           'MeterBandHeader', 'MeterBandType', 'MeterBandDrop',
           'MeterBandDscpRemark', 'MeterBandExperimenter')


class Meter(IntEnum):
    """Meter numbering. Flow meters can use any number up to OFPM_MAX."""

    #: Last usable meter.
    OFPM_MAX = 0xffff0000

    # Virtual meters.
    #: Meter for slow datapath, if any.
    OFPM_SLOWPATH = 0xfffffffd
    #: Meter for controller connection.
    OFPM_CONTROLLER = 0xfffffffe
    #: Represents all meters for stat requests commands.
    OFPM_ALL = 0xffffffff


class MeterModCommand(IntEnum):
    """Meter commands."""

    #: New meter.
    OFPMC_ADD = 0
    #: Modify specified meter.
    OFPMC_MODIFY = 1
    #: Delete specified meter.
    OFPMC_DELETE = 2


class MeterFlags(GenericBitMask):
    """Meter configuration flags."""

    #: Rate value in kb/s (kilo-bit per second).
    OFPMF_KBPS = 1 << 0
    #: Rate value in packet/sec.
    OFPMF_PKTPS = 1 << 1
    #: Do burst size.
    OFPMF_BURST = 1 << 2
    #: Collect statistics.
    OFPMF_STATS = 1 << 3


class MeterBandType(IntEnum):
    """Meter band types."""

    #: Drop packet.
    OFPMBT_DROP = 1
    #: Remark DSCP in the IP header.
    OFPMBT_DSCP_REMARK = 2
    #: Experimenter meter band.
    OFPMBT_EXPERIMENTER = 0xFFFF

    def find_class(self):
        """Return a class related with this type."""
        types = {1: MeterBandDrop, 2: MeterBandDscpRemark,
                 3: MeterBandExperimenter}
        return types[self.value]


class MeterBandHeader(GenericStruct):
    """Common header for all meter bands."""

    band_type = UBInt16(enum_ref=MeterBandType)
    length = UBInt16()
    rate = UBInt32()
    burst_size = UBInt32()

    def __init__(self, band_type=None, rate=None, burst_size=None):
        """Create a MeterBandHeader with the optional parameters below.

        Args:
            band_type (MeterBandType): One of OFPMBT_*.
            rate (int): Rate for this band.
            burst_size (int): Size of bursts.
        """
        super().__init__()
        self.band_type = band_type
        self.rate = rate
        self.burst_size = burst_size
        self.update_length()

    def update_length(self):
        """Update the length attribute of current instance."""
        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.

        """
        band_type = UBInt16(enum_ref=MeterBandType)
        band_type.unpack(buff, offset)
        self.__class__ = MeterBandType(band_type.value).find_class()

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

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


class MeterMod(GenericMessage):
    """Meter configuration."""

    header = Header(message_type=Type.OFPT_METER_MOD)
    command = UBInt16(enum_ref=MeterModCommand)
    flags = UBInt16(enum_ref=MeterFlags)
    meter_id = UBInt32()
    bands = FixedTypeList(MeterBandHeader)

    def __init__(self, xid=None, command=None, flags=None, meter_id=None,
                 bands=None):
        """Create a MeterMod with the optional parameters below.

        Args:
            xid (int): Headers transaction id. Defaults to random.
            command (MeterModCommand): One of OFPMC_*.
            flags (MeterFlags): One of OFPMF_*.
            meter_id (int): Meter instance.
            bands (MeterBandHeader): The bands length is inferred from the
                length field in the header.
        """
        super().__init__(xid)
        self.command = command
        self.flags = flags
        self.meter_id = meter_id
        self.bands = bands


class MeterBandDrop(MeterBandHeader):
    """OFPMBT_DROP band - drop packets."""

    pad = Pad(4)

    def __init__(self, rate=None, burst_size=None):
        """Create a MeterBandDrop with the optional parameters below.

        Args:
            rate (int): Rate for dropping packets.
            burst_size (int): Size of bursts.
        """
        super().__init__(MeterBandType.OFPMBT_DROP, rate, burst_size)


class MeterBandDscpRemark(MeterBandHeader):
    """OFPMBT_DSCP_REMARK band - Remark DSCP in the IP header."""

    prec_level = UBInt8()
    pad = Pad(3)

    def __init__(self, rate=None, burst_size=None, prec_level=None):
        """Create a MeterBandDscpRemark with the optional parameters below.

        Args:
            rate (int): Rate for remarking packets.
            burst_size (int): Size of bursts.
            prec_level (int): Number of precendence level to substract.
        """
        super().__init__(MeterBandType.OFPMBT_DSCP_REMARK, rate, burst_size)
        self.prec_level = prec_level


class MeterBandExperimenter(MeterBandHeader):
    """OFPMBT_EXPERIMENTER band - Write actions in action set."""

    experimenter = UBInt32()

    def __init__(self, rate=None, burst_size=None, experimenter=None):
        """Create a MeterBandExperimenter with the optional parameters below.

        Args:
            rate (int): Rate for remarking packets.
            burst_size (int): Size of bursts.
            experimenter (int): Experimenter ID which takes the same form as in
                :class:`.ExperimenterHeader`.
        """
        super().__init__(MeterBandType.OFPMBT_EXPERIMENTER, rate, burst_size)
        self.experimenter = experimenter


class ListOfMeterBandHeader(FixedTypeList):
    """List of MeterBandHeader.

    Represented by instances of MeterBandHeader.
    """

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

        Args:
        items (MeterBandHeader): Instance or a list of instances.
        """
        super().__init__(pyof_class=MeterBandHeader, items=items)


1			"""Meter modification message."""
2			from enum import IntEnum
3
4			from pyof.foundation.base import GenericBitMask, GenericMessage
5			from pyof.foundation.basic_types import (
6			FixedTypeList, GenericStruct, Pad, UBInt8, UBInt16, UBInt32)
7			from pyof.v0x04.common.header import Header, Type
8
9			__all__ = ('MeterMod', 'Meter', 'MeterModCommand', 'MeterFlags',
10			'MeterBandHeader', 'MeterBandType', 'MeterBandDrop',
11			'MeterBandDscpRemark', 'MeterBandExperimenter')
12
13
14			class Meter(IntEnum):
15			"""Meter numbering. Flow meters can use any number up to OFPM_MAX."""
16
17			#: Last usable meter.
18			OFPM_MAX = 0xffff0000
19
20			# Virtual meters.
21			#: Meter for slow datapath, if any.
22			OFPM_SLOWPATH = 0xfffffffd
23			#: Meter for controller connection.
24			OFPM_CONTROLLER = 0xfffffffe
25			#: Represents all meters for stat requests commands.
26			OFPM_ALL = 0xffffffff
27
28
29			class MeterModCommand(IntEnum):
30			"""Meter commands."""
31
32			#: New meter.
33			OFPMC_ADD = 0
34			#: Modify specified meter.
35			OFPMC_MODIFY = 1
36			#: Delete specified meter.
37			OFPMC_DELETE = 2
38
39
40			class MeterFlags(GenericBitMask):
41			"""Meter configuration flags."""
42
43			#: Rate value in kb/s (kilo-bit per second).
44			OFPMF_KBPS = 1 << 0
45			#: Rate value in packet/sec.
46			OFPMF_PKTPS = 1 << 1
47			#: Do burst size.
48			OFPMF_BURST = 1 << 2
49			#: Collect statistics.
50			OFPMF_STATS = 1 << 3
51
52
53			class MeterBandType(IntEnum):
54			"""Meter band types."""
55
56			#: Drop packet.
57			OFPMBT_DROP = 1
58			#: Remark DSCP in the IP header.
59			OFPMBT_DSCP_REMARK = 2
60			#: Experimenter meter band.
61			OFPMBT_EXPERIMENTER = 0xFFFF
62
63			def find_class(self):
64			"""Return a class related with this type."""
65			types = {1: MeterBandDrop, 2: MeterBandDscpRemark,
66			3: MeterBandExperimenter}
67			return types[self.value]
68
69
70			class MeterBandHeader(GenericStruct):
71			"""Common header for all meter bands."""
72
73			band_type = UBInt16(enum_ref=MeterBandType)
74			length = UBInt16()
75			rate = UBInt32()
76			burst_size = UBInt32()
77
78			def __init__(self, band_type=None, rate=None, burst_size=None):
79			"""Create a MeterBandHeader with the optional parameters below.
80
81			Args:
82			band_type (MeterBandType): One of OFPMBT_*.
83			rate (int): Rate for this band.
84			burst_size (int): Size of bursts.
85			"""
86			super().__init__()
87			self.band_type = band_type
88			self.rate = rate
89			self.burst_size = burst_size
90			self.update_length()
91
92			def update_length(self):
93			"""Update the length attribute of current instance."""
94			self.length = self.get_size()
95
96			def unpack(self, buff=None, offset=0):
97			"""Unpack buff into this object.
98
99			This method will convert a binary data into a readable value according
100			to the attribute format.
101
102			Args:
103			buff (bytes): Binary buffer.
104			offset (int): Where to begin unpacking.
105
106			Raises:
107			:exc:`~.exceptions.UnpackException`: If unpack fails.
108
109			"""
110			band_type = UBInt16(enum_ref=MeterBandType)
111			band_type.unpack(buff, offset)
112			self.__class__ = MeterBandType(band_type.value).find_class()
113
114			length = UBInt16()
115			length.unpack(buff, offset=offset+2)
116
117			super().unpack(buff[:offset+length.value], offset)
118
119
120			class MeterMod(GenericMessage):
121			"""Meter configuration."""
122
123			header = Header(message_type=Type.OFPT_METER_MOD)
124			command = UBInt16(enum_ref=MeterModCommand)
125			flags = UBInt16(enum_ref=MeterFlags)
126			meter_id = UBInt32()
127			bands = FixedTypeList(MeterBandHeader)
128
129			def __init__(self, xid=None, command=None, flags=None, meter_id=None,
130			bands=None):
131			"""Create a MeterMod with the optional parameters below.
132
133			Args:
134			xid (int): Headers transaction id. Defaults to random.
135			command (MeterModCommand): One of OFPMC_*.
136			flags (MeterFlags): One of OFPMF_*.
137			meter_id (int): Meter instance.
138			bands (MeterBandHeader): The bands length is inferred from the
139			length field in the header.
140			"""
141			super().__init__(xid)
142			self.command = command
143			self.flags = flags
144			self.meter_id = meter_id
145			self.bands = bands
146
147
148			class MeterBandDrop(MeterBandHeader):
149			"""OFPMBT_DROP band - drop packets."""
150
151			pad = Pad(4)
152
153			def __init__(self, rate=None, burst_size=None):
154			"""Create a MeterBandDrop with the optional parameters below.
155
156			Args:
157			rate (int): Rate for dropping packets.
158			burst_size (int): Size of bursts.
159			"""
160			super().__init__(MeterBandType.OFPMBT_DROP, rate, burst_size)
161
162
163			class MeterBandDscpRemark(MeterBandHeader):
164			"""OFPMBT_DSCP_REMARK band - Remark DSCP in the IP header."""
165
166			prec_level = UBInt8()
167			pad = Pad(3)
168
169			def __init__(self, rate=None, burst_size=None, prec_level=None):
170			"""Create a MeterBandDscpRemark with the optional parameters below.
171
172			Args:
173			rate (int): Rate for remarking packets.
174			burst_size (int): Size of bursts.
175			prec_level (int): Number of precendence level to substract.
176			"""
177			super().__init__(MeterBandType.OFPMBT_DSCP_REMARK, rate, burst_size)
178			self.prec_level = prec_level
179
180
181			class MeterBandExperimenter(MeterBandHeader):
182			"""OFPMBT_EXPERIMENTER band - Write actions in action set."""
183
184			experimenter = UBInt32()
185
186			def __init__(self, rate=None, burst_size=None, experimenter=None):
187			"""Create a MeterBandExperimenter with the optional parameters below.
188
189			Args:
190			rate (int): Rate for remarking packets.
191			burst_size (int): Size of bursts.
192			experimenter (int): Experimenter ID which takes the same form as in
193			:class:`.ExperimenterHeader`.
194			"""
195			super().__init__(MeterBandType.OFPMBT_EXPERIMENTER, rate, burst_size)
196			self.experimenter = experimenter
197
198
199			class ListOfMeterBandHeader(FixedTypeList):
200			"""List of MeterBandHeader.
201
202			Represented by instances of MeterBandHeader.
203			"""
204
205			def __init__(self, items=None):
206			"""Create a ListOfMeterBandHeader with the optional parameters below.
207
208			Args:
209			items (MeterBandHeader): Instance or a list of instances.
210			"""
211			super().__init__(pyof_class=MeterBandHeader, items=items)
212

kytos / python-openflow

Pull Request — master (#505)

MeterFlags

Complexity

Size/Duplication

Test Coverage

Importance

Duplication Side-by-Side

Filter issues like