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

Test Failed

Pull Request — master (#405)

unknown

created 2017-07-13 20:01 UTC

BinaryData._pack() A

↳ Parent: BinaryData

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	2
ccs	1
cts	1
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
333		super().__init__(value, enum_ref=enum_ref)
334
335		def _pack_if_necessary(self, value):
336		if hasattr(value, 'pack') and callable(value.pack):
337		value = value.pack()
338
339		if not isinstance(value, bytes):
340	1	msg = 'BinaryData value must contain bytes or have pack method; '
341	1	msg += 'Received type {} value: {}'.format(type(value), value)
342		raise ValueError(msg)
343	1
344	1	def _pack(self):
345		return self._value
346	1
347		def pack(self, value):
348		value = self._pack_if_necessary(value)
349	1	return super().pack(value)
350
351	1	def unpack(self, buff, offset=0):
352	1	"""Unpack a binary message into this object's attributes.
353
354		Unpack the binary value buff and update this object attributes based
355		on the results. Since the buff is binary data, no conversion is done.
356
357		All the data in the buffer from the offset forward will be used,
358		so the buffer must be truncated using the desired size before passing
359	1	it to BinaryData.
360
361		Args:
362		buff (bytes): Binary data package to be unpacked.
363		offset (int): Where to begin unpacking.
364		"""
365		self._value = buff[offset:]
366
367		def _get_size(self):
368		return len(self._value)
369
370
371		class TypeList(GenericStruct, list):
372	1	"""Base class for lists that store objects of one single type."""
373	1
374		def __init__(self, items):
375	1	"""Initialize the list with one item or a list of items.
376	1
377		Args:
378		items (iterable, ``pyof_class``): Items to be stored.
379		"""
380	1	super().__init__()
381	1	if isinstance(items, list):
382		self.extend(items)
383	1	elif items:
384		self.append(items)
385
386		def extend(self, items):
387		"""Extend the list by adding all items of ``items``.
388
389		Args:
390		items (iterable): Items to be added to the list.
391
392		Raises:
393		:exc:`~.exceptions.WrongListItemType`: If an item has an unexpected
394	1	type.
395		"""
396	1	for item in items:
397		self.append(item)
398
399		def _pack(self):
400		bin_message = b''
401	1	try:
402		for item in self:
403		bin_message += item._pack()
404		return bin_message
405		except exceptions.PackException as err:
406		msg = "{} pack error: {}".format(type(self).__name__, err)
407		raise exceptions.PackException(msg)
408
409		def unpack(self, buff, item_class, offset=0):
410	1	"""Unpack the elements of the list.
411
412		This unpack method considers that all elements are of the same type.
413		To use this class with a pyof_class that accepts elements with
414		different classes, you must reimplement the unpack method.
415
416	1	Args:
417		buff (bytes): The binary data to be unpacked.
418	1	item_class (:obj:`type`): Class of the expected items on this list.
419		offset (int): If we need to shift the beginning of the data.
420		"""
421		begin = offset
422		limit_buff = len(buff)
423
424		while begin < limit_buff:
425		item = item_class()
426		item.unpack(buff, begin)
427	1	self.append(item)
428	1	begin += item.get_size()
429
430	1	def _get_size(self):
431	1	if not self:
432		# If this is a empty list, then returns zero
433	1	return 0
434	1	elif issubclass(type(self[0]), GenericType):
435		# If the type of the elements is GenericType, then returns the
436	1	# length of the list multiplied by the size of the GenericType.
437		return len(self) * self[0].get_size()
438	1
439		# Otherwise iter over the list accumulating the sizes.
440		return sum(item.get_size() for item in self)
441
442		def __str__(self):
443		"""Human-readable object representantion."""
444		return "{}".format([str(item) for item in self])
445
446
447		class FixedTypeList(TypeList):
448	1	"""A list that stores instances of one pyof class."""
449
450	1	_pyof_class = None
451
452		def __init__(self, pyof_class, items=None):
453		"""The constructor parameters follows.
454
455		Args:
456		pyof_class (:obj:`type`): Class of the items to be stored.
457		items (iterable, ``pyof_class``): Items to be stored.
458		"""
459		self._pyof_class = pyof_class
460		super().__init__(items)
461	1
462	1	def append(self, item):
463	1	"""Append one item to the list.
464	1
465		Args:
466	1	item: Item to be appended. Its type must match the one defined in
467		the constructor.
468
469	1	Raises:
470		:exc:`~.exceptions.WrongListItemType`: If the item has a different
471		type than the one specified in the constructor.
472	1	"""
473		if isinstance(item, list):
474		self.extend(item)
475		elif issubclass(item.__class__, self._pyof_class):
476		list.append(self, item)
477		else:
478	1	raise exceptions.WrongListItemType(item.__class__.__name__,
479	1	self._pyof_class.__name__)
480	1
481	1	def insert(self, index, item):
482		"""Insert an item at the specified index.
483
484	1	Args:
485		index (int): Position to insert the item.
486		item: Item to be inserted. It must have the type specified in the
487		constructor.
488
489		Raises:
490		:exc:`~.exceptions.WrongListItemType`: If the item has a different
491		type than the one specified in the constructor.
492		"""
493		if issubclass(item.__class__, self._pyof_class):
494	1	list.insert(self, index, item)
495	1	else:
496		raise exceptions.WrongListItemType(item.__class__.__name__,
497	1	self._pyof_class.__name__)
498
499		def unpack(self, buff, offset=0):
500		"""Unpack the elements of the list.
501
502		This unpack method considers that all elements have the same size.
503	1	To use this class with a pyof_class that accepts elements with
504	1	different sizes, you must reimplement the unpack method.
505
506	1	Args:
507	1	buff (bytes): The binary data to be unpacked.
508		offset (int): If we need to shift the beginning of the data.
509	1	"""
510	1	super().unpack(buff, self._pyof_class, offset)
511	1
512
513	1	class ConstantTypeList(TypeList):
514	1	"""List that contains only objects of the same type (class).
515	1
516	1	The types of all items are expected to be the same as the first item's.
517	1	Otherwise, :exc:`~.exceptions.WrongListItemType` is raised in many
518		list operations.
519		"""
520
521		def __init__(self, items=None): # noqa
522	1	"""The contructor can contain the items to be stored.
523
524		Args:
525		items (iterable, :class:`object`): Items to be stored.
526
527		Raises:
528		:exc:`~.exceptions.WrongListItemType`: If an item has a different
529		type than the first item to be stored.
530		"""
531		super().__init__(items)
532
533		def append(self, item):
534	1	"""Append one item to the list.
535	1
536		Args:
537	1	item: Item to be appended.
538	1
539	1	Raises:
540	1	:exc:`~.exceptions.WrongListItemType`: If an item has a different
541	1	type than the first item to be stored.
542		"""
543	1	if isinstance(item, list):
544		self.extend(item)
545		elif not self:
546		list.append(self, item)
547		elif item.__class__ == self[0].__class__:
548		list.append(self, item)
549		else:
550		raise exceptions.WrongListItemType(item.__class__.__name__,
551		self[0].__class__.__name__)
552
553		def insert(self, index, item):
554	1	"""Insert an item at the specified index.
555	1
556		Args:
557	1	index (int): Position to insert the item.
558	1	item: Item to be inserted.
559
560		Raises:
561		:exc:`~.exceptions.WrongListItemType`: If an item has a different
562		type than the first item to be stored.
563		"""
564	1	if not self:
565		list.append(self, item)
566	1	elif item.__class__ == self[0].__class__:
567		list.insert(self, index, item)
568	1	else:
569		raise exceptions.WrongListItemType(item.__class__.__name__,
570		self[0].__class__.__name__)
571
572
573	1	def get_custom_tlv_class(type_size=3, length_size=1):
574		""" Generate a CUstomTLV class
575
576	1	Create a CustomTLV class with the defined number of bytes for type and
577		length fields.
578	1
579		Args:
580		type_size (int): length in bytes for the type field of the TLV
581		length_size (int): length in bytes for the length field of the TLV
582		"""
583
584		size_classes = {1: UBInt8,
585	1	2: UBInt16,
586	1	3: UBInt24,
587		4: UBInt32}
588	1
589		custom_type = size_classes[type_size]
590		custom_length = size_classes[length_size]
591
592		class CustomTLV(GenericStruct):
593		""" a compact TLV class
594
595		Args:
596		tlv_type: type field of the TLV
597		tlv_value: length field of the TLV
598		"""
599	1	tlv_type = custom_type()
600		tlv_length = custom_length()
601	1	tlv_value = BinaryData()
602	1
603		def __init__(self, tlv_type=None, tlv_value=None):
604		super().__init__()
605		self.tlv_type = tlv_type
606		self.tlv_value = tlv_value
607	1	self._update_tlv_length()
608
609		def _update_tlv_length(self):
610		cls = type(self)
611		self.tlv_length = (type(cls.tlv_value)(self.tlv_value)).get_size()
612
613		def _pack(self):
614		self._update_tlv_length()
615		return super()._pack()
616
617		def unpack(self, buff, offset=0):
618		"""Unpack the buffer into a custom TLV
619
620		Args:
621		buff (bytes): The binary data to be unpacked.
622		offset (int): If we need to shift the beginning of the data.
623		"""
624
625	1	begin = offset
626		for name, value in list(self.get_class_attributes())[:-1]:
627		size = self._unpack_attribute(name, value, buff, begin)
628		begin += size
629		self._unpack_attribute('tlv_value', type(self).tlv_value,
630		buff[:begin + self.tlv_length],
631		begin)
632
633		return CustomTLV
634
635
636		CustomTLV_24_8 = get_custom_tlv_class(type_size=3, length_size=1)
637

kytos / python-openflow

Pull Request — master (#405)

BinaryData._pack() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like