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 20:15 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	3
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)

        return value

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

kytos / python-openflow

Pull Request — master (#392)

BinaryData.pack() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like