BinaryData._pack() - Code Metrics - Inspection of "Match fix" - kytos/python-openflow - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#392)

unknown

created 2017-07-13 19:59 UTC

BinaryData._pack() A

↳ Parent: BinaryData

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

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

"""Basic types used in structures and messages."""

# System imports
import struct

# Local source tree imports
from pyof.foundation import exceptions
from pyof.foundation.base import GenericStruct, GenericType, GenericUBIntType

# Third-party imports

__all__ = ('BinaryData', 'Char', 'ConstantTypeList', 'FixedTypeList',
           'IPAddress', 'DPID', 'HWAddress', 'Pad', 'UBInt8', 'UBInt16',
           'UBInt24', 'UBInt32', 'UBInt64')


class Pad(GenericType):
    """Class for padding attributes."""

    _fmt = ''

    def __init__(self, length=0):
        """Pad up to ``length``, in bytes.

        Args:
            length (int): Total length, in bytes.
        """
        super().__init__()
        self._length = length

    def __repr__(self):
        return "{}({})".format(type(self).__name__, self._length)

    def __str__(self):
        return '0' * self._length

    def _get_size(self):
        return self._length

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

        Do nothing, since the _length is already defined and it is just a Pad.
        Keep buff and offset just for compability with other unpack methods.
        [this will check if bytes are 0 for validity in the future]

        Args:
            buff: Buffer where data is located.
            offset (int): Where data stream begins.
        """
        pass

    def _pack(self):
        return b'\x00' * self._length


class UBInt8(GenericType):
    """Format character for an Unsigned Char.

    Class for an 8-bit (1-byte) Unsigned Integer.
    """

    _fmt = "!B"


class UBInt16(GenericType):
    """Format character for an Unsigned Short.

    Class for an 16-bit (2-byte) Unsigned Integer.
    """

    _fmt = "!H"


class UBInt24(GenericUBIntType):
    """Format character for an Unsigned Short.

    Class for an 16-bit (2-byte) Unsigned Integer.
    """

    _buff_size = 3


class UBInt32(GenericType):
    """Format character for an Unsigned Int.

    Class for an 32-bit (4-byte) Unsigned Integer.
    """

    _fmt = "!I"


class UBInt64(GenericType):
    """Format character for an Unsigned Long Long.

    Class for an 64-bit (8-byte) Unsigned Integer.
    """

    _fmt = "!Q"


class DPID(GenericType):
    """DataPath ID. Identifies a switch."""

    _fmt = "!8B"

    def __init__(self, value=None):
        """Create an instance and optionally set its dpid value.

        Args:
            dpid (str): String with DPID value(e.g. `00:00:00:00:00:00:00:01`).
        """
        super().__init__(value=value)

    def __str__(self):
        return str(self._value)

    @property
    def value(self):
        """Return dpid value.

        Returns:
            str: DataPath ID stored by DPID class.
        """
        return self._value

    def _pack(self):
        return struct.pack('!8B', *[int(v, 16)
                                    for v in self.value.split(':')])

    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.
        """
        begin = offset
        hexas = []
        while begin < offset + 8:
            number = struct.unpack("!B", buff[begin:begin + 1])[0]
            hexas.append("%.2x" % number)
            begin += 1
        self._value = ':'.join(hexas)


class Char(GenericType):
    """Build a double char type according to the length."""

    def __init__(self, value=None, length=0):
        """The constructor takes the optional parameters below.

        Args:
            value: The character to be build.
            length (int): Character size.
        """
        super().__init__(value)
        self.length = length
        self._fmt = '!{}{}'.format(self.length, 's')

    def _get_new_instance(self, value):
        return type(self)(value, length=self.length)

    def _pack(self):
        packed = struct.pack(self._fmt, bytes(self.value, 'ascii'))
        return packed[:-1] + b'\0'  # null-terminated

    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.
        """
        try:
            begin = offset
            end = begin + self.length
            unpacked_data = struct.unpack(self._fmt, buff[begin:end])[0]
        except struct.error:
            raise Exception("%s: %s" % (offset, buff))

        self._value = unpacked_data.decode('ascii').rstrip('\0')


class IPAddress(GenericType):
    """Defines a IP address."""

    netmask = UBInt32()
    max_prefix = UBInt32(32)

    def __init__(self, value="0.0.0.0/32"):
        """The constructor takes the parameters below.

        Args:
            address (str): IP Address using ipv4. Defaults to '0.0.0.0/32'
        """
        if value.find('/') >= 0:
            value, netmask = value.split('/')
        else:
            netmask = 32

        super().__init__(value)
        self.netmask = int(netmask)

    def _pack(self):
        value = self._value
        try:
            value = value.split('.')
            return struct.pack('!4B', *[int(x) for x in value])
        except struct.error as err:
            msg = "IPAddress error. "
            msg += "Class: {}, struct error: {} ".format(type(value).__name__,
                                                         err)
            raise exceptions.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.

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

        Raises:
            Exception: If there is a struct unpacking error.
        """
        try:
            unpacked_data = struct.unpack('!4B', buff[offset:offset + 4])
            self._value = '.'.join([str(x) for x in unpacked_data])
        except struct.error as e:
            raise exceptions.UnpackException('%s; %s: %s' % (e, offset, buff))

    def _get_size(self):
        return 4


class HWAddress(GenericType):
    """Defines a hardware address."""

    def __init__(self, value='00:00:00:00:00:00'):  # noqa
        """The constructor takes the parameters below.

        Args:
            value (bytes): Hardware address. Defaults to
                '00:00:00:00:00:00'.
        """
        if value == 0:
            value = '00:00:00:00:00:00'
        super().__init__(value)

    def _pack(self):
        value = self._value.split(':')

        try:
            return struct.pack('!6B', *[int(x, 16) for x in value])
        except struct.error as err:
            msg = "HWAddress error. "
            msg += "Class: {}, struct error: {} ".format(type(value).__name__,
                                                         err)
            raise exceptions.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.

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

        Raises:
            Exception: If there is a struct unpacking error.
        """
        def _int2hex(n):
            return "{0:0{1}x}".format(n, 2)

        try:
            unpacked_data = struct.unpack('!6B', buff[offset:offset + 6])
        except struct.error as e:
            raise exceptions.UnpackException('%s; %s: %s' % (e, offset, buff))

        transformed_data = ':'.join([_int2hex(x) for x in unpacked_data])
        self._value = transformed_data

    def _get_size(self):
        return 6

    def is_broadcast(self):
        """Return true if the value is a broadcast address. False otherwise."""
        return self.value == 'ff:ff:ff:ff:ff:ff'


class BinaryData(GenericType):
    """Class to create objects that represent binary data.

    This is used in the ``data`` attribute from
    :class:`~pyof.v0x01.asynchronous.packet_in.PacketIn` and
    :class:`~pyof.v0x01.controller2switch.packet_out.PacketOut` messages.
    Both the :meth:`pack` and :meth:`unpack` methods will return the
    binary data itself. :meth:`get_size` method will
    return the size of the instance using Python's :func:`len`.
    """

    def __init__(self, value=None, enum_ref=None):  # noqa
        """The constructor takes the parameter below.

        Args:
            value (bytes): The binary data. Defaults to an empty value.

        Raises:
            ValueError: If given value is not bytes.
        """
        if value is None:
            value = b''

        value = self._pack_if_necessary(value)
        super().__init__(value, enum_ref=enum_ref)

    def _pack_if_necessary(self, value):
        if hasattr(value, 'pack') and callable(value.pack):
            value = value.pack()

        if not isinstance(value, bytes):
            msg = 'BinaryData value must contain bytes or have pack method; '
            msg += 'Received type {} value: {}'.format(type(value), value)
            raise ValueError(msg)

    def _pack(self):
        return self._value

    def pack(self, value):
        value = self._pack_if_necessary(value)
        return super().pack(value)

    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. Since the *buff* is binary data, no conversion is done.

        All the data in the buffer from the offset forward will be used,
        so the buffer must be truncated using the desired size before passing
        it to BinaryData.

        Args:
            buff (bytes): Binary data package to be unpacked.
            offset (int): Where to begin unpacking.
        """
        self._value = buff[offset:]

    def _get_size(self):
        return len(self._value)


class TypeList(GenericStruct, list):
    """Base class for lists that store objects of one single type."""

    def __init__(self, items):
        """Initialize the list with one item or a list of items.

        Args:
            items (iterable, ``pyof_class``): Items to be stored.
        """
        super().__init__()
        if isinstance(items, list):
            self.extend(items)
        elif items:
            self.append(items)

    def extend(self, items):
        """Extend the list by adding all items of ``items``.

        Args:
            items (iterable): Items to be added to the list.

        Raises:
            :exc:`~.exceptions.WrongListItemType`: If an item has an unexpected
                type.
        """
        for item in items:
            self.append(item)

    def _pack(self):
        bin_message = b''
        try:
            for item in self:
                bin_message += item._pack()
            return bin_message
        except exceptions.PackException as err:
            msg = "{} pack error: {}".format(type(self).__name__, err)
            raise exceptions.PackException(msg)

    def unpack(self, buff, item_class, offset=0):
        """Unpack the elements of the list.

        This unpack method considers that all elements are of the same type.
        To use this class with a pyof_class that accepts elements with
        different classes, you must reimplement the unpack method.

        Args:
            buff (bytes): The binary data to be unpacked.
            item_class (:obj:`type`): Class of the expected items on this list.
            offset (int): If we need to shift the beginning of the data.
        """
        begin = offset
        limit_buff = len(buff)

        while begin < limit_buff:
            item = item_class()
            item.unpack(buff, begin)
            self.append(item)
            begin += item.get_size()

    def _get_size(self):
        if not self:
            # If this is a empty list, then returns zero
            return 0
        elif issubclass(type(self[0]), GenericType):
            # If the type of the elements is GenericType, then returns the
            # length of the list multiplied by the size of the GenericType.
            return len(self) * self[0].get_size()

        # Otherwise iter over the list accumulating the sizes.
        return sum(item.get_size() for item in self)

    def __str__(self):
        """Human-readable object representantion."""
        return "{}".format([str(item) for item in self])


class FixedTypeList(TypeList):
    """A list that stores instances of one pyof class."""

    _pyof_class = None

    def __init__(self, pyof_class, items=None):
        """The constructor parameters follows.

        Args:
            pyof_class (:obj:`type`): Class of the items to be stored.
            items (iterable, ``pyof_class``): Items to be stored.
        """
        self._pyof_class = pyof_class
        super().__init__(items)

    def append(self, item):
        """Append one item to the list.

        Args:
            item: Item to be appended. Its type must match the one defined in
                the constructor.

        Raises:
            :exc:`~.exceptions.WrongListItemType`: If the item has a different
                type than the one specified in the constructor.
        """
        if isinstance(item, list):
            self.extend(item)
        elif issubclass(item.__class__, self._pyof_class):
            list.append(self, item)
        else:
            raise exceptions.WrongListItemType(item.__class__.__name__,
                                               self._pyof_class.__name__)

    def insert(self, index, item):
        """Insert an item at the specified index.

        Args:
            index (int): Position to insert the item.
            item: Item to be inserted. It must have the type specified in the
                constructor.

        Raises:
            :exc:`~.exceptions.WrongListItemType`: If the item has a different
                type than the one specified in the constructor.
        """
        if issubclass(item.__class__, self._pyof_class):
            list.insert(self, index, item)
        else:
            raise exceptions.WrongListItemType(item.__class__.__name__,
                                               self._pyof_class.__name__)

    def unpack(self, buff, offset=0):
        """Unpack the elements of the list.

        This unpack method considers that all elements have the same size.
        To use this class with a pyof_class that accepts elements with
        different sizes, you must reimplement the unpack method.

        Args:
            buff (bytes): The binary data to be unpacked.
            offset (int): If we need to shift the beginning of the data.
        """
        super().unpack(buff, self._pyof_class, offset)


class ConstantTypeList(TypeList):
    """List that contains only objects of the same type (class).

    The types of all items are expected to be the same as the first item's.
    Otherwise, :exc:`~.exceptions.WrongListItemType` is raised in many
    list operations.
    """

    def __init__(self, items=None):  # noqa
        """The contructor can contain the items to be stored.

        Args:
            items (iterable, :class:`object`): Items to be stored.

        Raises:
            :exc:`~.exceptions.WrongListItemType`: If an item has a different
                type than the first item to be stored.
        """
        super().__init__(items)

    def append(self, item):
        """Append one item to the list.

        Args:
            item: Item to be appended.

        Raises:
            :exc:`~.exceptions.WrongListItemType`: If an item has a different
                type than the first item to be stored.
        """
        if isinstance(item, list):
            self.extend(item)
        elif not self:
            list.append(self, item)
        elif item.__class__ == self[0].__class__:
            list.append(self, item)
        else:
            raise exceptions.WrongListItemType(item.__class__.__name__,
                                               self[0].__class__.__name__)

    def insert(self, index, item):
        """Insert an item at the specified index.

        Args:
            index (int): Position to insert the item.
            item: Item to be inserted.

        Raises:
            :exc:`~.exceptions.WrongListItemType`: If an item has a different
                type than the first item to be stored.
        """
        if not self:
            list.append(self, item)
        elif item.__class__ == self[0].__class__:
            list.insert(self, index, item)
        else:
            raise exceptions.WrongListItemType(item.__class__.__name__,
                                               self[0].__class__.__name__)


def get_custom_tlv_class(type_size=3, length_size=1):
    """ Generate a CUstomTLV class

    Create a CustomTLV class with the defined number of bytes for type and
    length fields.

    Args:
        type_size (int): length in bytes for the type field of the TLV
        length_size (int): length in bytes for the length field of the TLV
    """

    size_classes = {1: UBInt8,
                    2: UBInt16,
                    3: UBInt24,
                    4: UBInt32}

    custom_type = size_classes[type_size]
    custom_length = size_classes[length_size]

    class CustomTLV(GenericStruct):
        """ a compact TLV class

        Args:
            tlv_type: type field of the TLV
            tlv_value: length field of the TLV
        """
        tlv_type = custom_type()
        tlv_length = custom_length()
        tlv_value = BinaryData()

        def __init__(self, tlv_type=None, tlv_value=None):
            super().__init__()
            self.tlv_type = tlv_type
            self.tlv_value = tlv_value
            self._update_tlv_length()

        def _update_tlv_length(self):
            cls = type(self)
            self.tlv_length = (type(cls.tlv_value)(self.tlv_value)).get_size()

        def _pack(self):
            self._update_tlv_length()
            return super()._pack()

        def unpack(self, buff, offset=0):
            """Unpack the buffer into a custom TLV

            Args:
                buff (bytes): The binary data to be unpacked.
                offset (int): If we need to shift the beginning of the data.
            """

            begin = offset
            for name, value in list(self.get_class_attributes())[:-1]:
                size = self._unpack_attribute(name, value, buff, begin)
                begin += size
            self._unpack_attribute('tlv_value', type(self).tlv_value,
                                   buff[:begin + self.tlv_length],
                                   begin)

    return CustomTLV


CustomTLV_24_8 = get_custom_tlv_class(type_size=3, length_size=1)


1		"""Basic types used in structures and messages."""
2
3		# System imports
4	1	import struct
5
6		# Local source tree imports
7	1	from pyof.foundation import exceptions
8	1	from pyof.foundation.base import GenericStruct, GenericType, GenericUBIntType
9
10		# Third-party imports
11
12	1	__all__ = ('BinaryData', 'Char', 'ConstantTypeList', 'FixedTypeList',
13		'IPAddress', 'DPID', 'HWAddress', 'Pad', 'UBInt8', 'UBInt16',
14		'UBInt24', 'UBInt32', 'UBInt64')
15
16
17	1	class Pad(GenericType):
18		"""Class for padding attributes."""
19
20	1	_fmt = ''
21
22	1	def __init__(self, length=0):
23		"""Pad up to ``length``, in bytes.
24
25		Args:
26		length (int): Total length, in bytes.
27		"""
28	1	super().__init__()
29	1	self._length = length
30
31	1	def __repr__(self):
32		return "{}({})".format(type(self).__name__, self._length)
33
34	1	def __str__(self):
35		return '0' * self._length
36
37	1	def _get_size(self):
38		return self._length
39
40		def unpack(self, buff, offset=0):
41		"""Unpack buff into this object.
42
43		Do nothing, since the _length is already defined and it is just a Pad.
44		Keep buff and offset just for compability with other unpack methods.
45		[this will check if bytes are 0 for validity in the future]
46
47		Args:
48	1	buff: Buffer where data is located.
49		offset (int): Where data stream begins.
50	1	"""
51		pass
52
53		def _pack(self):
54		return b'\x00' * self._length
55
56
57		class UBInt8(GenericType):
58		"""Format character for an Unsigned Char.
59
60	1	Class for an 8-bit (1-byte) Unsigned Integer.
61		"""
62	1
63		_fmt = "!B"
64
65
66		class UBInt16(GenericType):
67		"""Format character for an Unsigned Short.
68
69		Class for an 16-bit (2-byte) Unsigned Integer.
70		"""
71
72		_fmt = "!H"
73	1
74
75		class UBInt24(GenericUBIntType):
76	1	"""Format character for an Unsigned Short.
77
78		Class for an 16-bit (2-byte) Unsigned Integer.
79		"""
80
81		_buff_size = 3
82	1
83
84		class UBInt32(GenericType):
85	1	"""Format character for an Unsigned Int.
86
87		Class for an 32-bit (4-byte) Unsigned Integer.
88		"""
89
90		_fmt = "!I"
91	1
92
93		class UBInt64(GenericType):
94	1	"""Format character for an Unsigned Long Long.
95
96		Class for an 64-bit (8-byte) Unsigned Integer.
97		"""
98
99		_fmt = "!Q"
100	1
101
102		class DPID(GenericType):
103	1	"""DataPath ID. Identifies a switch."""
104
105		_fmt = "!8B"
106
107		def __init__(self, value=None):
108		"""Create an instance and optionally set its dpid value.
109	1
110		Args:
111		dpid (str): String with DPID value(e.g. `00:00:00:00:00:00:00:01`).
112	1	"""
113		super().__init__(value=value)
114
115	1	def __str__(self):
116		return str(self._value)
117	1
118		@property
119		def value(self):
120		"""Return dpid value.
121
122		Returns:
123	1	str: DataPath ID stored by DPID class.
124		"""
125	1	return self._value
126
127		def _pack(self):
128	1	return struct.pack('!8B', *[int(v, 16)
129		for v in self.value.split(':')])
130
131		def unpack(self, buff, offset=0):
132		"""Unpack a binary message into this object's attributes.
133	1
134		Unpack the binary value buff and update this object attributes based
135		on the results.
136
137		Args:
138		buff (bytes): Binary data package to be unpacked.
139		offset (int): Where to begin unpacking.
140
141		Raises:
142	1	Exception: If there is a struct unpacking error.
143	1	"""
144	1	begin = offset
145	1	hexas = []
146	1	while begin < offset + 8:
147		number = struct.unpack("!B", buff[begin:begin + 1])[0]
148	1	hexas.append("%.2x" % number)
149		begin += 1
150		self._value = ':'.join(hexas)
151
152
153		class Char(GenericType):
154		"""Build a double char type according to the length."""
155
156		def __init__(self, value=None, length=0):
157		"""The constructor takes the optional parameters below.
158
159		Args:
160		value: The character to be build.
161	1	length (int): Character size.
162	1	"""
163	1	super().__init__(value)
164	1	self.length = length
165	1	self._fmt = '!{}{}'.format(self.length, 's')
166	1
167	1	def _get_new_instance(self, value):
168		return type(self)(value, length=self.length)
169
170	1	def _pack(self):
171		packed = struct.pack(self._fmt, bytes(self.value, 'ascii'))
172		return packed[:-1] + b'\0' # null-terminated
173	1
174		def unpack(self, buff, offset=0):
175		"""Unpack a binary message into this object's attributes.
176
177		Unpack the binary value buff and update this object attributes based
178		on the results.
179
180	1	Args:
181	1	buff (bytes): Binary data package to be unpacked.
182	1	offset (int): Where to begin unpacking.
183
184	1	Raises:
185		Exception: If there is a struct unpacking error.
186		"""
187		try:
188		begin = offset
189		end = begin + self.length
190		unpacked_data = struct.unpack(self._fmt, buff[begin:end])[0]
191		except struct.error:
192		raise Exception("%s: %s" % (offset, buff))
193	1
194	1	self._value = unpacked_data.decode('ascii').rstrip('\0')
195
196	1
197	1	class IPAddress(GenericType):
198	1	"""Defines a IP address."""
199	1
200	1	netmask = UBInt32()
201		max_prefix = UBInt32(32)
202
203		def __init__(self, value="0.0.0.0/32"):
204		"""The constructor takes the parameters below.
205
206		Args:
207	1	address (str): IP Address using ipv4. Defaults to '0.0.0.0/32'
208		"""
209		if value.find('/') >= 0:
210		value, netmask = value.split('/')
211		else:
212		netmask = 32
213
214		super().__init__(value)
215		self.netmask = int(netmask)
216
217		def _pack(self):
218		value = self._value
219		try:
220	1	value = value.split('.')
221	1	return struct.pack('!4B', *[int(x) for x in value])
222	1	except struct.error as err:
223	1	msg = "IPAddress error. "
224		msg += "Class: {}, struct error: {} ".format(type(value).__name__,
225		err)
226		raise exceptions.PackException(msg)
227	1
228		def unpack(self, buff, offset=0):
229		"""Unpack a binary message into this object's attributes.
230	1
231		Unpack the binary value buff and update this object attributes based
232		on the results.
233	1
234	1	Args:
235		buff (bytes): Binary data package to be unpacked.
236	1	offset (int): Where to begin unpacking.
237
238		Raises:
239		Exception: If there is a struct unpacking error.
240		"""
241		try:
242		unpacked_data = struct.unpack('!4B', buff[offset:offset + 4])
243	1	self._value = '.'.join([str(x) for x in unpacked_data])
244	1	except struct.error as e:
245		raise exceptions.UnpackException('%s; %s: %s' % (e, offset, buff))
246	1
247		def _get_size(self):
248	1	return 4
249	1
250
251	1	class HWAddress(GenericType):
252		"""Defines a hardware address."""
253
254		def __init__(self, value='00:00:00:00:00:00'): # noqa
255		"""The constructor takes the parameters below.
256
257		Args:
258		value (bytes): Hardware address. Defaults to
259		'00:00:00:00:00:00'.
260		"""
261		if value == 0:
262		value = '00:00:00:00:00:00'
263		super().__init__(value)
264
265	1	def _pack(self):
266	1	value = self._value.split(':')
267
268	1	try:
269	1	return struct.pack('!6B', *[int(x, 16) for x in value])
270		except struct.error as err:
271	1	msg = "HWAddress error. "
272		msg += "Class: {}, struct error: {} ".format(type(value).__name__,
273		err)
274	1	raise exceptions.PackException(msg)
275	1
276	1	def unpack(self, buff, offset=0):
277		"""Unpack a binary message into this object's attributes.
278
279		Unpack the binary value buff and update this object attributes based
280		on the results.
281
282		Args:
283	1	buff (bytes): Binary data package to be unpacked.
284		offset (int): Where to begin unpacking.
285
286		Raises:
287		Exception: If there is a struct unpacking error.
288		"""
289		def _int2hex(n):
290		return "{0:0{1}x}".format(n, 2)
291
292		try:
293		unpacked_data = struct.unpack('!6B', buff[offset:offset + 6])
294		except struct.error as e:
295		raise exceptions.UnpackException('%s; %s: %s' % (e, offset, buff))
296	1
297	1	transformed_data = ':'.join([_int2hex(x) for x in unpacked_data])
298	1	self._value = transformed_data
299
300		def _get_size(self):
301		return 6
302	1
303		def is_broadcast(self):
304		"""Return true if the value is a broadcast address. False otherwise."""
305		return self.value == 'ff:ff:ff:ff:ff:ff'
306
307
308		class BinaryData(GenericType):
309		"""Class to create objects that represent binary data.
310
311		This is used in the ``data`` attribute from
312		:class:`~pyof.v0x01.asynchronous.packet_in.PacketIn` and
313	1	:class:`~pyof.v0x01.controller2switch.packet_out.PacketOut` messages.
314		Both the :meth:`pack` and :meth:`unpack` methods will return the
315		binary data itself. :meth:`get_size` method will
316	1	return the size of the instance using Python's :func:`len`.
317		"""
318
319	1	def __init__(self, value=None, enum_ref=None): # noqa
320		"""The constructor takes the parameter below.
321
322		Args:
323		value (bytes): The binary data. Defaults to an empty value.
324
325		Raises:
326	1	ValueError: If given value is not bytes.
327		"""
328	1	if value is None:
329		value = b''
330
331		value = self._pack_if_necessary(value)
332		super().__init__(value, enum_ref=enum_ref)
333
334		def _pack_if_necessary(self, value):
335		if hasattr(value, 'pack') and callable(value.pack):
336		value = value.pack()
337
338		if not isinstance(value, bytes):
339		msg = 'BinaryData value must contain bytes or have pack method; '
340	1	msg += 'Received type {} value: {}'.format(type(value), value)
341	1	raise ValueError(msg)
342
343	1	def _pack(self):
344	1	return self._value
345
346	1	def pack(self, value):
347		value = self._pack_if_necessary(value)
348		return super().pack(value)
349	1
350		def unpack(self, buff, offset=0):
351	1	"""Unpack a binary message into this object's attributes.
352	1
353		Unpack the binary value buff and update this object attributes based
354		on the results. Since the buff is binary data, no conversion is done.
355
356		All the data in the buffer from the offset forward will be used,
357		so the buffer must be truncated using the desired size before passing
358		it to BinaryData.
359	1
360		Args:
361		buff (bytes): Binary data package to be unpacked.
362		offset (int): Where to begin unpacking.
363		"""
364		self._value = buff[offset:]
365
366		def _get_size(self):
367		return len(self._value)
368
369
370		class TypeList(GenericStruct, list):
371		"""Base class for lists that store objects of one single type."""
372	1
373	1	def __init__(self, items):
374		"""Initialize the list with one item or a list of items.
375	1
376	1	Args:
377		items (iterable, ``pyof_class``): Items to be stored.
378		"""
379		super().__init__()
380	1	if isinstance(items, list):
381	1	self.extend(items)
382		elif items:
383	1	self.append(items)
384
385		def extend(self, items):
386		"""Extend the list by adding all items of ``items``.
387
388		Args:
389		items (iterable): Items to be added to the list.
390
391		Raises:
392		:exc:`~.exceptions.WrongListItemType`: If an item has an unexpected
393		type.
394	1	"""
395		for item in items:
396	1	self.append(item)
397
398		def _pack(self):
399		bin_message = b''
400		try:
401	1	for item in self:
402		bin_message += item._pack()
403		return bin_message
404		except exceptions.PackException as err:
405		msg = "{} pack error: {}".format(type(self).__name__, err)
406		raise exceptions.PackException(msg)
407
408		def unpack(self, buff, item_class, offset=0):
409		"""Unpack the elements of the list.
410	1
411		This unpack method considers that all elements are of the same type.
412		To use this class with a pyof_class that accepts elements with
413		different classes, you must reimplement the unpack method.
414
415		Args:
416	1	buff (bytes): The binary data to be unpacked.
417		item_class (:obj:`type`): Class of the expected items on this list.
418	1	offset (int): If we need to shift the beginning of the data.
419		"""
420		begin = offset
421		limit_buff = len(buff)
422
423		while begin < limit_buff:
424		item = item_class()
425		item.unpack(buff, begin)
426		self.append(item)
427	1	begin += item.get_size()
428	1
429		def _get_size(self):
430	1	if not self:
431	1	# If this is a empty list, then returns zero
432		return 0
433	1	elif issubclass(type(self[0]), GenericType):
434	1	# If the type of the elements is GenericType, then returns the
435		# length of the list multiplied by the size of the GenericType.
436	1	return len(self) * self[0].get_size()
437
438	1	# Otherwise iter over the list accumulating the sizes.
439		return sum(item.get_size() for item in self)
440
441		def __str__(self):
442		"""Human-readable object representantion."""
443		return "{}".format([str(item) for item in self])
444
445
446		class FixedTypeList(TypeList):
447		"""A list that stores instances of one pyof class."""
448	1
449		_pyof_class = None
450	1
451		def __init__(self, pyof_class, items=None):
452		"""The constructor parameters follows.
453
454		Args:
455		pyof_class (:obj:`type`): Class of the items to be stored.
456		items (iterable, ``pyof_class``): Items to be stored.
457		"""
458		self._pyof_class = pyof_class
459		super().__init__(items)
460
461	1	def append(self, item):
462	1	"""Append one item to the list.
463	1
464	1	Args:
465		item: Item to be appended. Its type must match the one defined in
466	1	the constructor.
467
468		Raises:
469	1	:exc:`~.exceptions.WrongListItemType`: If the item has a different
470		type than the one specified in the constructor.
471		"""
472	1	if isinstance(item, list):
473		self.extend(item)
474		elif issubclass(item.__class__, self._pyof_class):
475		list.append(self, item)
476		else:
477		raise exceptions.WrongListItemType(item.__class__.__name__,
478	1	self._pyof_class.__name__)
479	1
480	1	def insert(self, index, item):
481	1	"""Insert an item at the specified index.
482
483		Args:
484	1	index (int): Position to insert the item.
485		item: Item to be inserted. It must have the type specified in the
486		constructor.
487
488		Raises:
489		:exc:`~.exceptions.WrongListItemType`: If the item has a different
490		type than the one specified in the constructor.
491		"""
492		if issubclass(item.__class__, self._pyof_class):
493		list.insert(self, index, item)
494	1	else:
495	1	raise exceptions.WrongListItemType(item.__class__.__name__,
496		self._pyof_class.__name__)
497	1
498		def unpack(self, buff, offset=0):
499		"""Unpack the elements of the list.
500
501		This unpack method considers that all elements have the same size.
502		To use this class with a pyof_class that accepts elements with
503	1	different sizes, you must reimplement the unpack method.
504	1
505		Args:
506	1	buff (bytes): The binary data to be unpacked.
507	1	offset (int): If we need to shift the beginning of the data.
508		"""
509	1	super().unpack(buff, self._pyof_class, offset)
510	1
511	1
512		class ConstantTypeList(TypeList):
513	1	"""List that contains only objects of the same type (class).
514	1
515	1	The types of all items are expected to be the same as the first item's.
516	1	Otherwise, :exc:`~.exceptions.WrongListItemType` is raised in many
517	1	list operations.
518		"""
519
520		def __init__(self, items=None): # noqa
521		"""The contructor can contain the items to be stored.
522	1
523		Args:
524		items (iterable, :class:`object`): Items to be stored.
525
526		Raises:
527		:exc:`~.exceptions.WrongListItemType`: If an item has a different
528		type than the first item to be stored.
529		"""
530		super().__init__(items)
531
532		def append(self, item):
533		"""Append one item to the list.
534	1
535	1	Args:
536		item: Item to be appended.
537	1
538	1	Raises:
539	1	:exc:`~.exceptions.WrongListItemType`: If an item has a different
540	1	type than the first item to be stored.
541	1	"""
542		if isinstance(item, list):
543	1	self.extend(item)
544		elif not self:
545		list.append(self, item)
546		elif item.__class__ == self[0].__class__:
547		list.append(self, item)
548		else:
549		raise exceptions.WrongListItemType(item.__class__.__name__,
550		self[0].__class__.__name__)
551
552		def insert(self, index, item):
553		"""Insert an item at the specified index.
554	1
555	1	Args:
556		index (int): Position to insert the item.
557	1	item: Item to be inserted.
558	1
559		Raises:
560		:exc:`~.exceptions.WrongListItemType`: If an item has a different
561		type than the first item to be stored.
562		"""
563		if not self:
564	1	list.append(self, item)
565		elif item.__class__ == self[0].__class__:
566	1	list.insert(self, index, item)
567		else:
568	1	raise exceptions.WrongListItemType(item.__class__.__name__,
569		self[0].__class__.__name__)
570
571
572		def get_custom_tlv_class(type_size=3, length_size=1):
573	1	""" Generate a CUstomTLV class
574
575		Create a CustomTLV class with the defined number of bytes for type and
576	1	length fields.
577
578	1	Args:
579		type_size (int): length in bytes for the type field of the TLV
580		length_size (int): length in bytes for the length field of the TLV
581		"""
582
583		size_classes = {1: UBInt8,
584		2: UBInt16,
585	1	3: UBInt24,
586	1	4: UBInt32}
587
588	1	custom_type = size_classes[type_size]
589		custom_length = size_classes[length_size]
590
591		class CustomTLV(GenericStruct):
592		""" a compact TLV class
593
594		Args:
595		tlv_type: type field of the TLV
596		tlv_value: length field of the TLV
597		"""
598		tlv_type = custom_type()
599	1	tlv_length = custom_length()
600		tlv_value = BinaryData()
601	1
602	1	def __init__(self, tlv_type=None, tlv_value=None):
603		super().__init__()
604		self.tlv_type = tlv_type
605		self.tlv_value = tlv_value
606		self._update_tlv_length()
607	1
608		def _update_tlv_length(self):
609		cls = type(self)
610		self.tlv_length = (type(cls.tlv_value)(self.tlv_value)).get_size()
611
612		def _pack(self):
613		self._update_tlv_length()
614		return super()._pack()
615
616		def unpack(self, buff, offset=0):
617		"""Unpack the buffer into a custom TLV
618
619		Args:
620		buff (bytes): The binary data to be unpacked.
621		offset (int): If we need to shift the beginning of the data.
622		"""
623
624		begin = offset
625	1	for name, value in list(self.get_class_attributes())[:-1]:
626		size = self._unpack_attribute(name, value, buff, begin)
627		begin += size
628		self._unpack_attribute('tlv_value', type(self).tlv_value,
629		buff[:begin + self.tlv_length],
630		begin)
631
632		return CustomTLV
633
634
635		CustomTLV_24_8 = get_custom_tlv_class(type_size=3, length_size=1)
636

kytos / python-openflow

Pull Request — master (#392)

BinaryData._pack() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like