IPAddress._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:21 UTC

IPAddress._pack() A

↳ Parent: IPAddress

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	3.3332

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
dl	0
loc	10
ccs	4
cts	6
cp	0.6667
rs	9.4285
c	1
b	0
f	0
cc	3
crap	3.3332

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

kytos / python-openflow

Pull Request — master (#405)

IPAddress._pack() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like