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 16:36 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, dpid=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=dpid)

    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 _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, address="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 address.find('/') >= 0:
            address, netmask = address.split('/')
        else:
            netmask = 32

        super().__init__(address)
        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, hw_address='00:00:00:00:00:00'):  # noqa
        """The constructor takes the parameters below.

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

    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''
        if not isinstance(value, bytes):
            raise ValueError('BinaryData must contain bytes.')
        super().__init__(value, enum_ref=enum_ref)

    def _pack(self):
        return self._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, dpid=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=dpid)
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 _pack(self):
168		packed = struct.pack(self._fmt, bytes(self.value, 'ascii'))
169		return packed[:-1] + b'\0' # null-terminated
170	1
171		def unpack(self, buff, offset=0):
172		"""Unpack a binary message into this object's attributes.
173	1
174		Unpack the binary value buff and update this object attributes based
175		on the results.
176
177		Args:
178		buff (bytes): Binary data package to be unpacked.
179		offset (int): Where to begin unpacking.
180	1
181	1	Raises:
182	1	Exception: If there is a struct unpacking error.
183		"""
184	1	try:
185		begin = offset
186		end = begin + self.length
187		unpacked_data = struct.unpack(self._fmt, buff[begin:end])[0]
188		except struct.error:
189		raise Exception("%s: %s" % (offset, buff))
190
191		self._value = unpacked_data.decode('ascii').rstrip('\0')
192
193	1
194	1	class IPAddress(GenericType):
195		"""Defines a IP address."""
196	1
197	1	netmask = UBInt32()
198	1	max_prefix = UBInt32(32)
199	1
200	1	def __init__(self, address="0.0.0.0/32"):
201		"""The constructor takes the parameters below.
202
203		Args:
204		address (str): IP Address using ipv4. Defaults to '0.0.0.0/32'
205		"""
206		if address.find('/') >= 0:
207	1	address, netmask = address.split('/')
208		else:
209		netmask = 32
210
211		super().__init__(address)
212		self.netmask = int(netmask)
213
214		def _pack(self):
215		value = self._value
216		try:
217		value = value.split('.')
218		return struct.pack('!4B', *[int(x) for x in value])
219		except struct.error as err:
220	1	msg = "IPAddress error. "
221	1	msg += "Class: {}, struct error: {} ".format(type(value).__name__,
222	1	err)
223	1	raise exceptions.PackException(msg)
224
225		def unpack(self, buff, offset=0):
226		"""Unpack a binary message into this object's attributes.
227	1
228		Unpack the binary value buff and update this object attributes based
229		on the results.
230	1
231		Args:
232		buff (bytes): Binary data package to be unpacked.
233	1	offset (int): Where to begin unpacking.
234	1
235		Raises:
236	1	Exception: If there is a struct unpacking error.
237		"""
238		try:
239		unpacked_data = struct.unpack('!4B', buff[offset:offset + 4])
240		self._value = '.'.join([str(x) for x in unpacked_data])
241		except struct.error as e:
242		raise exceptions.UnpackException('%s; %s: %s' % (e, offset, buff))
243	1
244	1	def _get_size(self):
245		return 4
246	1
247
248	1	class HWAddress(GenericType):
249	1	"""Defines a hardware address."""
250
251	1	def __init__(self, hw_address='00:00:00:00:00:00'): # noqa
252		"""The constructor takes the parameters below.
253
254		Args:
255		hw_address (bytes): Hardware address. Defaults to
256		'00:00:00:00:00:00'.
257		"""
258		if hw_address == 0:
259		hw_address = '00:00:00:00:00:00'
260		super().__init__(hw_address)
261
262		def _pack(self):
263		value = self._value.split(':')
264
265	1	try:
266	1	return struct.pack('!6B', *[int(x, 16) for x in value])
267		except struct.error as err:
268	1	msg = "HWAddress error. "
269	1	msg += "Class: {}, struct error: {} ".format(type(value).__name__,
270		err)
271	1	raise exceptions.PackException(msg)
272
273		def unpack(self, buff, offset=0):
274	1	"""Unpack a binary message into this object's attributes.
275	1
276	1	Unpack the binary value buff and update this object attributes based
277		on the results.
278
279		Args:
280		buff (bytes): Binary data package to be unpacked.
281		offset (int): Where to begin unpacking.
282
283	1	Raises:
284		Exception: If there is a struct unpacking error.
285		"""
286		def _int2hex(n):
287		return "{0:0{1}x}".format(n, 2)
288
289		try:
290		unpacked_data = struct.unpack('!6B', buff[offset:offset + 6])
291		except struct.error as e:
292		raise exceptions.UnpackException('%s; %s: %s' % (e, offset, buff))
293
294		transformed_data = ':'.join([_int2hex(x) for x in unpacked_data])
295		self._value = transformed_data
296	1
297	1	def _get_size(self):
298	1	return 6
299
300		def is_broadcast(self):
301		"""Return true if the value is a broadcast address. False otherwise."""
302	1	return self.value == 'ff:ff:ff:ff:ff:ff'
303
304
305		class BinaryData(GenericType):
306		"""Class to create objects that represent binary data.
307
308		This is used in the ``data`` attribute from
309		:class:`~pyof.v0x01.asynchronous.packet_in.PacketIn` and
310		:class:`~pyof.v0x01.controller2switch.packet_out.PacketOut` messages.
311		Both the :meth:`pack` and :meth:`unpack` methods will return the
312		binary data itself. :meth:`get_size` method will
313	1	return the size of the instance using Python's :func:`len`.
314		"""
315
316	1	def __init__(self, value=None, enum_ref=None): # noqa
317		"""The constructor takes the parameter below.
318
319	1	Args:
320		value (bytes): The binary data. Defaults to an empty value.
321
322		Raises:
323		ValueError: If given value is not bytes.
324		"""
325		if value is None:
326	1	value = b''
327		if not isinstance(value, bytes):
328	1	raise ValueError('BinaryData must contain bytes.')
329		super().__init__(value, enum_ref=enum_ref)
330
331		def _pack(self):
332		return self._value
333
334		def unpack(self, buff, offset=0):
335		"""Unpack a binary message into this object's attributes.
336
337		Unpack the binary value buff and update this object attributes based
338		on the results. Since the buff is binary data, no conversion is done.
339
340	1	All the data in the buffer from the offset forward will be used,
341	1	so the buffer must be truncated using the desired size before passing
342		it to BinaryData.
343	1
344	1	Args:
345		buff (bytes): Binary data package to be unpacked.
346	1	offset (int): Where to begin unpacking.
347		"""
348		self._value = buff[offset:]
349	1
350		def _get_size(self):
351	1	return len(self._value)
352	1
353
354		class TypeList(GenericStruct, list):
355		"""Base class for lists that store objects of one single type."""
356
357		def __init__(self, items):
358		"""Initialize the list with one item or a list of items.
359	1
360		Args:
361		items (iterable, ``pyof_class``): Items to be stored.
362		"""
363		super().__init__()
364		if isinstance(items, list):
365		self.extend(items)
366		elif items:
367		self.append(items)
368
369		def extend(self, items):
370		"""Extend the list by adding all items of ``items``.
371
372	1	Args:
373	1	items (iterable): Items to be added to the list.
374
375	1	Raises:
376	1	:exc:`~.exceptions.WrongListItemType`: If an item has an unexpected
377		type.
378		"""
379		for item in items:
380	1	self.append(item)
381	1
382		def _pack(self):
383	1	bin_message = b''
384		try:
385		for item in self:
386		bin_message += item._pack()
387		return bin_message
388		except exceptions.PackException as err:
389		msg = "{} pack error: {}".format(type(self).__name__, err)
390		raise exceptions.PackException(msg)
391
392		def unpack(self, buff, item_class, offset=0):
393		"""Unpack the elements of the list.
394	1
395		This unpack method considers that all elements are of the same type.
396	1	To use this class with a pyof_class that accepts elements with
397		different classes, you must reimplement the unpack method.
398
399		Args:
400		buff (bytes): The binary data to be unpacked.
401	1	item_class (:obj:`type`): Class of the expected items on this list.
402		offset (int): If we need to shift the beginning of the data.
403		"""
404		begin = offset
405		limit_buff = len(buff)
406
407		while begin < limit_buff:
408		item = item_class()
409		item.unpack(buff, begin)
410	1	self.append(item)
411		begin += item.get_size()
412
413		def _get_size(self):
414		if not self:
415		# If this is a empty list, then returns zero
416	1	return 0
417		elif issubclass(type(self[0]), GenericType):
418	1	# If the type of the elements is GenericType, then returns the
419		# length of the list multiplied by the size of the GenericType.
420		return len(self) * self[0].get_size()
421
422		# Otherwise iter over the list accumulating the sizes.
423		return sum(item.get_size() for item in self)
424
425		def __str__(self):
426		"""Human-readable object representantion."""
427	1	return "{}".format([str(item) for item in self])
428	1
429
430	1	class FixedTypeList(TypeList):
431	1	"""A list that stores instances of one pyof class."""
432
433	1	_pyof_class = None
434	1
435		def __init__(self, pyof_class, items=None):
436	1	"""The constructor parameters follows.
437
438	1	Args:
439		pyof_class (:obj:`type`): Class of the items to be stored.
440		items (iterable, ``pyof_class``): Items to be stored.
441		"""
442		self._pyof_class = pyof_class
443		super().__init__(items)
444
445		def append(self, item):
446		"""Append one item to the list.
447
448	1	Args:
449		item: Item to be appended. Its type must match the one defined in
450	1	the constructor.
451
452		Raises:
453		:exc:`~.exceptions.WrongListItemType`: If the item has a different
454		type than the one specified in the constructor.
455		"""
456		if isinstance(item, list):
457		self.extend(item)
458		elif issubclass(item.__class__, self._pyof_class):
459		list.append(self, item)
460		else:
461	1	raise exceptions.WrongListItemType(item.__class__.__name__,
462	1	self._pyof_class.__name__)
463	1
464	1	def insert(self, index, item):
465		"""Insert an item at the specified index.
466	1
467		Args:
468		index (int): Position to insert the item.
469	1	item: Item to be inserted. It must have the type specified in the
470		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 issubclass(item.__class__, self._pyof_class):
477		list.insert(self, index, item)
478	1	else:
479	1	raise exceptions.WrongListItemType(item.__class__.__name__,
480	1	self._pyof_class.__name__)
481	1
482		def unpack(self, buff, offset=0):
483		"""Unpack the elements of the list.
484	1
485		This unpack method considers that all elements have the same size.
486		To use this class with a pyof_class that accepts elements with
487		different sizes, you must reimplement the unpack method.
488
489		Args:
490		buff (bytes): The binary data to be unpacked.
491		offset (int): If we need to shift the beginning of the data.
492		"""
493		super().unpack(buff, self._pyof_class, offset)
494	1
495	1
496		class ConstantTypeList(TypeList):
497	1	"""List that contains only objects of the same type (class).
498
499		The types of all items are expected to be the same as the first item's.
500		Otherwise, :exc:`~.exceptions.WrongListItemType` is raised in many
501		list operations.
502		"""
503	1
504	1	def __init__(self, items=None): # noqa
505		"""The contructor can contain the items to be stored.
506	1
507	1	Args:
508		items (iterable, :class:`object`): Items to be stored.
509	1
510	1	Raises:
511	1	:exc:`~.exceptions.WrongListItemType`: If an item has a different
512		type than the first item to be stored.
513	1	"""
514	1	super().__init__(items)
515	1
516	1	def append(self, item):
517	1	"""Append one item to the list.
518
519		Args:
520		item: Item to be appended.
521
522	1	Raises:
523		:exc:`~.exceptions.WrongListItemType`: If an item has a different
524		type than the first item to be stored.
525		"""
526		if isinstance(item, list):
527		self.extend(item)
528		elif not self:
529		list.append(self, item)
530		elif item.__class__ == self[0].__class__:
531		list.append(self, item)
532		else:
533		raise exceptions.WrongListItemType(item.__class__.__name__,
534	1	self[0].__class__.__name__)
535	1
536		def insert(self, index, item):
537	1	"""Insert an item at the specified index.
538	1
539	1	Args:
540	1	index (int): Position to insert the item.
541	1	item: Item to be inserted.
542
543	1	Raises:
544		:exc:`~.exceptions.WrongListItemType`: If an item has a different
545		type than the first item to be stored.
546		"""
547		if not self:
548		list.append(self, item)
549		elif item.__class__ == self[0].__class__:
550		list.insert(self, index, item)
551		else:
552		raise exceptions.WrongListItemType(item.__class__.__name__,
553		self[0].__class__.__name__)
554	1
555	1
556		def get_custom_tlv_class(type_size=3, length_size=1):
557	1	""" Generate a CUstomTLV class
558	1
559		Create a CustomTLV class with the defined number of bytes for type and
560		length fields.
561
562		Args:
563		type_size (int): length in bytes for the type field of the TLV
564	1	length_size (int): length in bytes for the length field of the TLV
565		"""
566	1
567		size_classes = {1: UBInt8,
568	1	2: UBInt16,
569		3: UBInt24,
570		4: UBInt32}
571
572		custom_type = size_classes[type_size]
573	1	custom_length = size_classes[length_size]
574
575		class CustomTLV(GenericStruct):
576	1	""" a compact TLV class
577
578	1	Args:
579		tlv_type: type field of the TLV
580		tlv_value: length field of the TLV
581		"""
582		tlv_type = custom_type()
583		tlv_length = custom_length()
584		tlv_value = BinaryData()
585	1
586	1	def __init__(self, tlv_type=None, tlv_value=None):
587		super().__init__()
588	1	self.tlv_type = tlv_type
589		self.tlv_value = tlv_value
590		self._update_tlv_length()
591
592		def _update_tlv_length(self):
593		cls = type(self)
594		self.tlv_length = (type(cls.tlv_value)(self.tlv_value)).get_size()
595
596		def _pack(self):
597		self._update_tlv_length()
598		return super()._pack()
599	1
600		def unpack(self, buff, offset=0):
601	1	"""Unpack the buffer into a custom TLV
602	1
603		Args:
604		buff (bytes): The binary data to be unpacked.
605		offset (int): If we need to shift the beginning of the data.
606		"""
607	1
608		begin = offset
609		for name, value in list(self.get_class_attributes())[:-1]:
610		size = self._unpack_attribute(name, value, buff, begin)
611		begin += size
612		self._unpack_attribute('tlv_value', type(self).tlv_value,
613		buff[:begin + self.tlv_length],
614		begin)
615
616		return CustomTLV
617
618
619		CustomTLV_24_8 = get_custom_tlv_class(type_size=3, length_size=1)
620

kytos / python-openflow

Pull Request — master (#392)

DPID._pack() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like