DPID._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:34 UTC

DPID._pack() A

↳ Parent: DPID

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	2

Importance

Changes	1
Bugs	0	Features	0

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

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

kytos / python-openflow

Pull Request — master (#392)

DPID._pack() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like