IPAddress.get_size() - 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-05 22:25 UTC

IPAddress.get_size() A

↳ Parent: BinaryData

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

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

"""Basic types used in structures and messages."""

# System imports
import struct

# Local source tree imports
from pyof.foundation import exceptions
from pyof.foundation.base import GenericStruct, GenericType, GenericUBIntType

# Third-party imports

__all__ = ('BinaryData', 'Char', 'ConstantTypeList', 'FixedTypeList',
           'IPAddress', 'DPID', 'HWAddress', 'Pad', 'UBInt8', 'UBInt16',
           'UBInt24', 'UBInt32', 'UBInt64')


class Pad(GenericType):
    """Class for padding attributes."""

    _fmt = ''

    def __init__(self, length=0):
        """Pad up to ``length``, in bytes.

        Args:
            length (int): Total length, in bytes.
        """
        super().__init__()
        self._length = length

    def __repr__(self):
        return "{}({})".format(type(self).__name__, self._length)

    def __str__(self):
        return '0' * self._length

    def _get_size(self):
        return self._length

    def unpack(self, buff, offset=0):
        """Unpack *buff* into this object.

        Do nothing, since the _length is already defined and it is just a Pad.
        Keep buff and offset just for compability with other unpack methods.

        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): 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."""
        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.

        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(list, GenericStruct):
    """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 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.
            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):

    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):
        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):
            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
46		Args:
47		buff: Buffer where data is located.
48	1	offset (int): Where data stream begins.
49		"""
50	1	pass
51
52		def _pack(self):
53		return b'\x00' * self._length
54
55
56		class UBInt8(GenericType):
57		"""Format character for an Unsigned Char.
58
59		Class for an 8-bit (1-byte) Unsigned Integer.
60	1	"""
61
62	1	_fmt = "!B"
63
64
65		class UBInt16(GenericType):
66		"""Format character for an Unsigned Short.
67
68		Class for an 16-bit (2-byte) Unsigned Integer.
69		"""
70
71		_fmt = "!H"
72
73	1
74		class UBInt24(GenericUBIntType):
75		"""Format character for an Unsigned Short.
76	1
77		Class for an 16-bit (2-byte) Unsigned Integer.
78		"""
79
80		_buff_size = 3
81
82	1
83		class UBInt32(GenericType):
84		"""Format character for an Unsigned Int.
85	1
86		Class for an 32-bit (4-byte) Unsigned Integer.
87		"""
88
89		_fmt = "!I"
90
91	1
92		class UBInt64(GenericType):
93		"""Format character for an Unsigned Long Long.
94	1
95		Class for an 64-bit (8-byte) Unsigned Integer.
96		"""
97
98		_fmt = "!Q"
99
100	1
101		class DPID(GenericType):
102		"""DataPath ID. Identifies a switch."""
103	1
104		_fmt = "!8B"
105
106		def __init__(self, dpid=None):
107		"""Create an instance and optionally set its dpid value.
108
109	1	Args:
110		dpid (str): E.g. 00:00:00:00:00:00:00:01.
111		"""
112	1	super().__init__(value=dpid)
113
114		def __str__(self):
115	1	return str(self._value)
116
117	1	@property
118		def value(self):
119		"""Return dpid value."""
120		return self._value
121
122		def _pack(self):
123	1	return struct.pack('!8B', *[int(v, 16)
124		for v in self.value.split(':')])
125	1
126		def unpack(self, buff, offset=0):
127		"""Unpack a binary message into this object's attributes.
128	1
129		Unpack the binary value buff and update this object attributes based
130		on the results.
131
132		Args:
133	1	buff (bytes): Binary data package to be unpacked.
134		offset (int): Where to begin unpacking.
135
136		Raises:
137		Exception: If there is a struct unpacking error.
138		"""
139		begin = offset
140		hexas = []
141		while begin < offset + 8:
142	1	number = struct.unpack("!B", buff[begin:begin + 1])[0]
143	1	hexas.append("%.2x" % number)
144	1	begin += 1
145	1	self._value = ':'.join(hexas)
146	1
147
148	1	class Char(GenericType):
149		"""Build a double char type according to the length."""
150
151		def __init__(self, value=None, length=0):
152		"""The constructor takes the optional parameters below.
153
154		Args:
155		value: The character to be build.
156		length (int): Character size.
157		"""
158		super().__init__(value)
159		self.length = length
160		self._fmt = '!{}{}'.format(self.length, 's')
161	1
162	1	def _pack(self):
163	1	packed = struct.pack(self._fmt, bytes(self.value, 'ascii'))
164	1	return packed[:-1] + b'\0' # null-terminated
165	1
166	1	def unpack(self, buff, offset=0):
167	1	"""Unpack a binary message into this object's attributes.
168
169		Unpack the binary value buff and update this object attributes based
170	1	on the results.
171
172		Args:
173	1	buff (bytes): Binary data package to be unpacked.
174		offset (int): Where to begin unpacking.
175
176		Raises:
177		Exception: If there is a struct unpacking error.
178		"""
179		try:
180	1	begin = offset
181	1	end = begin + self.length
182	1	unpacked_data = struct.unpack(self._fmt, buff[begin:end])[0]
183		except struct.error:
184	1	raise Exception("%s: %s" % (offset, buff))
185
186		self._value = unpacked_data.decode('ascii').rstrip('\0')
187
188
189		class IPAddress(GenericType):
190		"""Defines a IP address."""
191
192		netmask = UBInt32()
193	1	max_prefix = UBInt32(32)
194	1
195		def __init__(self, address="0.0.0.0/32"):
196	1	"""The constructor takes the parameters below.
197	1
198	1	Args:
199	1	address (str): IP Address using ipv4.
200	1	Defaults to '0.0.0.0/32'
201		"""
202		if address.find('/') >= 0:
203		address, netmask = address.split('/')
204		else:
205		netmask = 32
206
207	1	super().__init__(address)
208		self.netmask = int(netmask)
209
210		def _pack(self):
211		value = self._value
212		try:
213		value = value.split('.')
214		return struct.pack('!4B', *[int(x) for x in value])
215		except struct.error as err:
216		msg = "IPAddress error. "
217		msg += "Class: {}, struct error: {} ".format(type(value).__name__,
218		err)
219		raise exceptions.PackException(msg)
220	1
221	1	def unpack(self, buff, offset=0):
222	1	"""Unpack a binary message into this object's attributes.
223	1
224		Unpack the binary value buff and update this object attributes based
225		on the results.
226
227	1	Args:
228		buff (bytes): Binary data package to be unpacked.
229		offset (int): Where to begin unpacking.
230	1
231		Raises:
232		Exception: If there is a struct unpacking error.
233	1	"""
234	1	try:
235		unpacked_data = struct.unpack('!4B', buff[offset:offset + 4])
236	1	self._value = '.'.join([str(x) for x in unpacked_data])
237		except struct.error as e:
238		raise exceptions.UnpackException('%s; %s: %s' % (e, offset, buff))
239
240		def _get_size(self):
241		return 4
242
243	1
244	1	class HWAddress(GenericType):
245		"""Defines a hardware address."""
246	1
247		def __init__(self, hw_address='00:00:00:00:00:00'): # noqa
248	1	"""The constructor takes the parameters below.
249	1
250		Args:
251	1	hw_address (bytes): Hardware address. Defaults to
252		'00:00:00:00:00:00'.
253		"""
254		if hw_address == 0:
255		hw_address = '00:00:00:00:00:00'
256		super().__init__(hw_address)
257
258		def _pack(self):
259		value = self._value.split(':')
260
261		try:
262		return struct.pack('!6B', *[int(x, 16) for x in value])
263		except struct.error as err:
264		msg = "HWAddress error. "
265	1	msg += "Class: {}, struct error: {} ".format(type(value).__name__,
266	1	err)
267		raise exceptions.PackException(msg)
268	1
269	1	def unpack(self, buff, offset=0):
270		"""Unpack a binary message into this object's attributes.
271	1
272		Unpack the binary value buff and update this object attributes based
273		on the results.
274	1
275	1	Args:
276	1	buff (bytes): Binary data package to be unpacked.
277		offset (int): Where to begin unpacking.
278
279		Raises:
280		Exception: If there is a struct unpacking error.
281		"""
282		def _int2hex(n):
283	1	return "{0:0{1}x}".format(n, 2)
284
285		try:
286		unpacked_data = struct.unpack('!6B', buff[offset:offset + 6])
287		except struct.error as e:
288		raise exceptions.UnpackException('%s; %s: %s' % (e, offset, buff))
289
290		transformed_data = ':'.join([_int2hex(x) for x in unpacked_data])
291		self._value = transformed_data
292
293		def _get_size(self):
294		return 6
295
296	1	def is_broadcast(self):
297	1	"""Return true if the value is a broadcast address. False otherwise."""
298	1	return self.value == 'ff:ff:ff:ff:ff:ff'
299
300
301		class BinaryData(GenericType):
302	1	"""Class to create objects that represent binary data.
303
304		This is used in the ``data`` attribute from
305		:class:`~pyof.v0x01.asynchronous.packet_in.PacketIn` and
306		:class:`~pyof.v0x01.controller2switch.packet_out.PacketOut` messages.
307		Both the :meth:`pack` and :meth:`unpack` methods will return the
308		binary data itself. :meth:`get_size` method will
309		return the size of the instance using Python's :func:`len`.
310		"""
311
312		def __init__(self, value=None, enum_ref=None): # noqa
313	1	"""The constructor takes the parameter below.
314
315		Args:
316	1	value (bytes): The binary data. Defaults to an empty value.
317
318		Raises:
319	1	ValueError: If given value is not bytes.
320		"""
321		if value is None:
322		value = b''
323		if not isinstance(value, bytes):
324		raise ValueError('BinaryData must contain bytes.')
325		super().__init__(value, enum_ref=enum_ref)
326	1
327		def _pack(self):
328	1	return self._value
329
330		def unpack(self, buff, offset=0):
331		"""Unpack a binary message into this object's attributes.
332
333		Unpack the binary value buff and update this object attributes based
334		on the results. Since the buff is binary data, no conversion is done.
335
336		Args:
337		buff (bytes): Binary data package to be unpacked.
338		offset (int): Where to begin unpacking.
339		"""
340	1	self._value = buff[offset:]
341	1
342		def _get_size(self):
343	1	return len(self._value)
344	1
345
346	1	class TypeList(list, GenericStruct):
347		"""Base class for lists that store objects of one single type."""
348
349	1	def __init__(self, items):
350		"""Initialize the list with one item or a list of items.
351	1
352	1	Args:
353		items (iterable, ``pyof_class``): Items to be stored.
354		"""
355		super().__init__()
356		if isinstance(items, list):
357		self.extend(items)
358		elif items:
359	1	self.append(items)
360
361		def extend(self, items):
362		"""Extend the list by adding all items of ``items``.
363
364		Args:
365		items (iterable): Items to be added to the list.
366
367		Raises:
368		:exc:`~.exceptions.WrongListItemType`: If an item has an unexpected
369		type.
370		"""
371		for item in items:
372	1	self.append(item)
373	1
374		def _pack(self):
375	1	bin_message = b''
376	1	try:
377		for item in self:
378		bin_message += item.pack()
379		return bin_message
380	1	except exceptions.PackException as err:
381	1	msg = "{} pack error: {}".format(type(self).__name__, err)
382		raise exceptions.PackException(msg)
383	1
384		def unpack(self, buff, item_class, offset=0):
385		"""Unpack the elements of the list.
386
387		This unpack method considers that all elements have the same size.
388		To use this class with a pyof_class that accepts elements with
389		different sizes, you must reimplement the unpack method.
390
391		Args:
392		buff (bytes): The binary data to be unpacked.
393		item_class (:obj:`type`): Class of the expected items on this list.
394	1	offset (int): If we need to shift the beginning of the data.
395		"""
396	1	begin = offset
397		limit_buff = len(buff)
398
399		while begin < limit_buff:
400		item = item_class()
401	1	item.unpack(buff, begin)
402		self.append(item)
403		begin += item.get_size()
404
405		def _get_size(self):
406		if not self:
407		# If this is a empty list, then returns zero
408		return 0
409		elif issubclass(type(self[0]), GenericType):
410	1	# If the type of the elements is GenericType, then returns the
411		# length of the list multiplied by the size of the GenericType.
412		return len(self) * self[0].get_size()
413
414		# Otherwise iter over the list accumulating the sizes.
415		return sum(item.get_size() for item in self)
416	1
417		def __str__(self):
418	1	"""Human-readable object representantion."""
419		return "{}".format([str(item) for item in self])
420
421
422		class FixedTypeList(TypeList):
423		"""A list that stores instances of one pyof class."""
424
425		_pyof_class = None
426
427	1	def __init__(self, pyof_class, items=None):
428	1	"""The constructor parameters follows.
429
430	1	Args:
431	1	pyof_class (:obj:`type`): Class of the items to be stored.
432		items (iterable, ``pyof_class``): Items to be stored.
433	1	"""
434	1	self._pyof_class = pyof_class
435		super().__init__(items)
436	1
437		def append(self, item):
438	1	"""Append one item to the list.
439
440		Args:
441		item: Item to be appended. Its type must match the one defined in
442		the constructor.
443
444		Raises:
445		:exc:`~.exceptions.WrongListItemType`: If the item has a different
446		type than the one specified in the constructor.
447		"""
448	1	if isinstance(item, list):
449		self.extend(item)
450	1	elif issubclass(item.__class__, self._pyof_class):
451		list.append(self, item)
452		else:
453		raise exceptions.WrongListItemType(item.__class__.__name__,
454		self._pyof_class.__name__)
455
456		def insert(self, index, item):
457		"""Insert an item at the specified index.
458
459		Args:
460		index (int): Position to insert the item.
461	1	item: Item to be inserted. It must have the type specified in the
462	1	constructor.
463	1
464	1	Raises:
465		:exc:`~.exceptions.WrongListItemType`: If the item has a different
466	1	type than the one specified in the constructor.
467		"""
468		if issubclass(item.__class__, self._pyof_class):
469	1	list.insert(self, index, item)
470		else:
471		raise exceptions.WrongListItemType(item.__class__.__name__,
472	1	self._pyof_class.__name__)
473
474		def unpack(self, buff, offset=0):
475		"""Unpack the elements of the list.
476
477		This unpack method considers that all elements have the same size.
478	1	To use this class with a pyof_class that accepts elements with
479	1	different sizes, you must reimplement the unpack method.
480	1
481	1	Args:
482		buff (bytes): The binary data to be unpacked.
483		offset (int): If we need to shift the beginning of the data.
484	1	"""
485		super().unpack(buff, self._pyof_class, offset)
486
487
488		class ConstantTypeList(TypeList):
489		"""List that contains only objects of the same type (class).
490
491		The types of all items are expected to be the same as the first item's.
492		Otherwise, :exc:`~.exceptions.WrongListItemType` is raised in many
493		list operations.
494	1	"""
495	1
496		def __init__(self, items=None): # noqa
497	1	"""The contructor can contain the items to be stored.
498
499		Args:
500		items (iterable, :class:`object`): Items to be stored.
501
502		Raises:
503	1	:exc:`~.exceptions.WrongListItemType`: If an item has a different
504	1	type than the first item to be stored.
505		"""
506	1	super().__init__(items)
507	1
508		def append(self, item):
509	1	"""Append one item to the list.
510	1
511	1	Args:
512		item: Item to be appended.
513	1
514	1	Raises:
515	1	:exc:`~.exceptions.WrongListItemType`: If an item has a different
516	1	type than the first item to be stored.
517	1	"""
518		if isinstance(item, list):
519		self.extend(item)
520		elif not self:
521		list.append(self, item)
522	1	elif item.__class__ == self[0].__class__:
523		list.append(self, item)
524		else:
525		raise exceptions.WrongListItemType(item.__class__.__name__,
526		self[0].__class__.__name__)
527
528		def insert(self, index, item):
529		"""Insert an item at the specified index.
530
531		Args:
532		index (int): Position to insert the item.
533		item: Item to be inserted.
534	1
535	1	Raises:
536		:exc:`~.exceptions.WrongListItemType`: If an item has a different
537	1	type than the first item to be stored.
538	1	"""
539	1	if not self:
540	1	list.append(self, item)
541	1	elif item.__class__ == self[0].__class__:
542		list.insert(self, index, item)
543	1	else:
544		raise exceptions.WrongListItemType(item.__class__.__name__,
545		self[0].__class__.__name__)
546
547
548		def get_custom_tlv_class(type_size=3, length_size=1):
549
550		size_classes = {1: UBInt8,
551		2: UBInt16,
552		3: UBInt24,
553		4: UBInt32}
554	1
555	1	custom_type = size_classes[type_size]
556		custom_length = size_classes[length_size]
557	1
558	1	class CustomTLV(GenericStruct):
559		tlv_type = custom_type()
560		tlv_length = custom_length()
561		tlv_value = BinaryData()
562
563		def __init__(self, tlv_type=None, tlv_value=None):
564	1	super().__init__()
565		self.tlv_type = tlv_type
566	1	self.tlv_value = tlv_value
567		self._update_tlv_length()
568	1
569		def _update_tlv_length(self):
570		cls = type(self)
571		self.tlv_length = (type(cls.tlv_value)(self.tlv_value)).get_size()
572
573	1	def _pack(self):
574		self._update_tlv_length()
575		return super()._pack()
576	1
577		def unpack(self, buff, offset=0):
578	1	begin = offset
579		for name, value in list(self.get_class_attributes())[:-1]:
580		size = self._unpack_attribute(name, value, buff, begin)
581		begin += size
582		self._unpack_attribute('tlv_value', type(self).tlv_value,
583		buff[:begin + self.tlv_length],
584		begin)
585	1
586	1	return CustomTLV
587
588	1
589		CustomTLV_24_8 = get_custom_tlv_class(type_size=3, length_size=1)
590

kytos / python-openflow

Pull Request — master (#392)

IPAddress.get_size() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like