BinaryData.pack() - Code Metrics - Inspection of "Binarydata fixes" - kytos/python-openflow - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#445)

by macartur

created 2017-09-12 19:36 UTC

BinaryData.pack() A

↳ Parent: BinaryData

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	2

Importance

Changes

Metric	Value
dl	0
loc	13
ccs	4
cts	4
cp	1
rs	9.4285
c	0
b	0
f	0
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

# Third-party imports

__all__ = ('BinaryData', 'Char', 'ConstantTypeList', 'FixedTypeList',
           'IPAddress', 'DPID', 'HWAddress', 'Pad', 'UBInt8', 'UBInt16',
           '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, value=None):
        """Return the type size in bytes.

        Args:
            value (int): In structs, the user can assign other value instead of
                this class' instance. Here, in such cases, ``self`` is a class
                attribute of the struct.

        Returns:
            int: Size in bytes.

        """
        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, value=None):
        """Pack the object.

        Args:
            value (int): In structs, the user can assign other value instead of
                this class' instance. Here, in such cases, ``self`` is a class
                attribute of the struct.

        Returns:
            bytes: the byte 0 (zero) *length* times.

        """
        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 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 self._value

    @property
    def value(self):
        """Return dpid value.

        Returns:
            str: DataPath ID stored by DPID class.

        """
        return self._value

    def pack(self, value=None):
        """Pack the value as a binary representation.

        Returns:
            bytes: The binary representation.

        Raises:
            struct.error: If the value does not fit the binary format.

        """
        if isinstance(value, type(self)):
            return value.pack()
        if value is None:
            value = self._value
        return struct.pack('!8B', *[int(v, 16) for v in 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):
        """Create a Char with 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, value=None):
        """Pack the value as a binary representation.

        Returns:
            bytes: The binary representation.

        Raises:
            struct.error: If the value does not fit the binary format.

        """
        if isinstance(value, type(self)):
            return value.pack()

        try:
            if value is None:
                value = self.value
            packed = struct.pack(self._fmt, bytes(value, 'ascii'))
            return packed[:-1] + b'\0'  # null-terminated
        except struct.error as err:
            msg = "Char Pack 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:
            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"):
        """Create an IPAddress with 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=None):
        """Pack the value as a binary representation.

        If the value is None the self._value will be used to pack.

        Args:
            value (str): IP Address with ipv4 format.

        Returns:
            bytes: The binary representation.

        Raises:
            struct.error: If the value does not fit the binary format.

        """
        if isinstance(value, type(self)):
            return value.pack()

        if value is None:
            value = self._value

        if value.find('/') >= 0:
            value = value.split('/')[0]

        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 exception:
            raise exceptions.UnpackException('%s; %s: %s' % (exception,
                                                             offset, buff))

    def get_size(self, value=None):
        """Return the ip address size in bytes.

        Args:
            value: In structs, the user can assign other value instead of
                this class' instance. Here, in such cases, ``self`` is a class
                attribute of the struct.

        Returns:
            int: The address size in bytes.

        """
        return 4


class HWAddress(GenericType):
    """Defines a hardware address."""


    # pylint: disable=useless-super-delegation
    def __init__(self, hw_address='00:00:00:00:00:00'):
        """Create a HWAddress with the parameters below.

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

    def pack(self, value=None):
        """Pack the value as a binary representation.

        If the passed value (or the self._value) is zero (int), then the pack
        will assume that the value to be packed is '00:00:00:00:00:00'.

        Returns
            bytes: The binary representation.

        Raises:
            struct.error: If the value does not fit the binary format.

        """
        if isinstance(value, type(self)):
            return value.pack()

        if value is None:
            value = self._value

        if value == 0:
            value = '00:00:00:00:00:00'

        value = 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(number):
            return "{0:0{1}x}".format(number, 2)

        try:
            unpacked_data = struct.unpack('!6B', buff[offset:offset+6])
        except struct.error as exception:
            raise exceptions.UnpackException('%s; %s: %s' % (exception,
                                                             offset, buff))

        transformed_data = ':'.join([_int2hex(x) for x in unpacked_data])
        self._value = transformed_data

    def get_size(self, value=None):
        """Return the address size in bytes.

        Args:
            value: In structs, the user can assign other value instead of
                this class' instance. Here, in such cases, ``self`` is a class
                attribute of the struct.

        Returns:
            int: The address size in bytes.

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

        """
        value = self._pack(value)
        super().__init__(value)

    @staticmethod
    def _pack(value):
        if hasattr(value, 'pack') and callable(value.pack):
            value = value.pack()
        elif value is None:
            value = b''

        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, value=None):
        """Pack the value as a binary representation.

        Returns:
            bytes: The binary representation.

        Raises:
            :exc:`~.exceptions.NotBinaryData`: If value is not :class:`bytes`.

        """
        if value is None:
            return self._value
        return self._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.

        Args:
            buff (bytes): Binary data package to be unpacked.
            offset (int): Where to begin unpacking.
        """
        self._value = buff[offset:]

    def get_size(self, value=None):
        """Return the size in bytes.

        Args:
            value (bytes): In structs, the user can assign other value instead
                of this class' instance. Here, in such cases, ``self`` is a
                class attribute of the struct.

        Returns:
            int: The address size in bytes.

        """
        if value is None:
            return len(self._value)
        elif hasattr(value, 'get_size'):
            return value.get_size()

        return len(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, value=None):
        """Pack the value as a binary representation.

        Returns:
            bytes: The binary representation.

        """
        if isinstance(value, type(self)):
            return value.pack()

        if value is None:
            value = self
        else:
            container = type(self)(items=None)
            container.extend(value)
            value = container

        bin_message = b''
        try:
            for item in value:
                bin_message += item.pack()
            return bin_message
        except exceptions.PackException as err:
            msg = "{} pack error: {}".format(type(self).__name__, err)
            raise exceptions.PackException(msg)

    # pylint: disable=arguments-differ
    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()
    # pylint: enable=arguments-differ

    def get_size(self, value=None):
        """Return the size in bytes.

        Args:
            value: In structs, the user can assign other value instead of
                this class' instance. Here, in such cases, ``self`` is a class
                attribute of the struct.

        Returns:
            int: The size in bytes.

        """
        if value is None:
            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)

        return type(self)(value).get_size()

    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):
        """Create a FixedTypeList with the 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):  # pylint: disable=arguments-differ
        """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.
    """

    # pylint: disable=useless-super-delegation
    def __init__(self, items=None):
        """Create a ConstantTypeList that can contain itens 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__)


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
9
10			# Third-party imports
11
12	1		__all__ = ('BinaryData', 'Char', 'ConstantTypeList', 'FixedTypeList',
13			'IPAddress', 'DPID', 'HWAddress', 'Pad', 'UBInt8', 'UBInt16',
14			'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, value=None):
38			"""Return the type size in bytes.
39
40			Args:
41			value (int): In structs, the user can assign other value instead of
42			this class' instance. Here, in such cases, ``self`` is a class
43			attribute of the struct.
44
45			Returns:
46			int: Size in bytes.
47
48			"""
49	1		return self._length
50
51	1		def unpack(self, buff, offset=0):
52			"""Unpack buff into this object.
53
54			Do nothing, since the _length is already defined and it is just a Pad.
55			Keep buff and offset just for compability with other unpack methods.
56
57			Args:
58			buff: Buffer where data is located.
59			offset (int): Where data stream begins.
60			"""
61	1		pass
62
63	1		def pack(self, value=None):
64			"""Pack the object.
65
66			Args:
67			value (int): In structs, the user can assign other value instead of
68			this class' instance. Here, in such cases, ``self`` is a class
69			attribute of the struct.
70
71			Returns:
72			bytes: the byte 0 (zero) length times.
73
74			"""
75	1		return b'\x00' * self._length
76
77
78	1		class UBInt8(GenericType):
79			"""Format character for an Unsigned Char.
80
81			Class for an 8-bit (1-byte) Unsigned Integer.
82			"""
83
84	1		_fmt = "!B"
85
86
87	1		class UBInt16(GenericType):
88			"""Format character for an Unsigned Short.
89
90			Class for an 16-bit (2-byte) Unsigned Integer.
91			"""
92
93	1		_fmt = "!H"
94
95
96	1		class UBInt32(GenericType):
97			"""Format character for an Unsigned Int.
98
99			Class for an 32-bit (4-byte) Unsigned Integer.
100			"""
101
102	1		_fmt = "!I"
103
104
105	1		class UBInt64(GenericType):
106			"""Format character for an Unsigned Long Long.
107
108			Class for an 64-bit (8-byte) Unsigned Integer.
109			"""
110
111	1		_fmt = "!Q"
112
113
114	1		class DPID(GenericType):
115			"""DataPath ID. Identifies a switch."""
116
117	1		_fmt = "!8B"
118
119	1		def __init__(self, dpid=None):
120			"""Create an instance and optionally set its dpid value.
121
122			Args:
123			dpid (str): String with DPID value(e.g. `00:00:00:00:00:00:00:01`).
124			"""
125	1		super().__init__(value=dpid)
126
127	1		def __str__(self):
128			return self._value
129
130	1		@property
131			def value(self):
132			"""Return dpid value.
133
134			Returns:
135			str: DataPath ID stored by DPID class.
136
137			"""
138			return self._value
139
140	1		def pack(self, value=None):
141			"""Pack the value as a binary representation.
142
143			Returns:
144			bytes: The binary representation.
145
146			Raises:
147			struct.error: If the value does not fit the binary format.
148
149			"""
150	1		if isinstance(value, type(self)):
151	1		return value.pack()
152	1		if value is None:
153	1		value = self._value
154	1		return struct.pack('!8B', *[int(v, 16) for v in value.split(':')])
155
156	1		def unpack(self, buff, offset=0):
157			"""Unpack a binary message into this object's attributes.
158
159			Unpack the binary value buff and update this object attributes based
160			on the results.
161
162			Args:
163			buff (bytes): Binary data package to be unpacked.
164			offset (int): Where to begin unpacking.
165
166			Raises:
167			Exception: If there is a struct unpacking error.
168
169			"""
170	1		begin = offset
171	1		hexas = []
172	1		while begin < offset + 8:
173	1		number = struct.unpack("!B", buff[begin:begin+1])[0]
174	1		hexas.append("%.2x" % number)
175	1		begin += 1
176	1		self._value = ':'.join(hexas)
177
178
179	1		class Char(GenericType):
180			"""Build a double char type according to the length."""
181
182	1		def __init__(self, value=None, length=0):
183			"""Create a Char with the optional parameters below.
184
185			Args:
186			value: The character to be build.
187			length (int): Character size.
188		View Code Duplication	"""
			0 ignored issues – show Duplication introduced 2017-07-26 18:53 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
189	1		super().__init__(value)
190	1		self.length = length
191	1		self._fmt = '!{}{}'.format(self.length, 's')
192
193	1		def pack(self, value=None):
194			"""Pack the value as a binary representation.
195
196			Returns:
197			bytes: The binary representation.
198
199			Raises:
200			struct.error: If the value does not fit the binary format.
201
202			"""
203	1		if isinstance(value, type(self)):
204	1		return value.pack()
205
206	1		try:
207	1		if value is None:
208	1		value = self.value
209	1		packed = struct.pack(self._fmt, bytes(value, 'ascii'))
210	1		return packed[:-1] + b'\0' # null-terminated
211			except struct.error as err:
212			msg = "Char Pack error. "
213			msg += "Class: {}, struct error: {} ".format(type(value).__name__,
214			err)
215			raise exceptions.PackException(msg)
216
217	1		def unpack(self, buff, offset=0):
218			"""Unpack a binary message into this object's attributes.
219
220			Unpack the binary value buff and update this object attributes based
221			on the results.
222
223			Args:
224			buff (bytes): Binary data package to be unpacked.
225			offset (int): Where to begin unpacking.
226
227			Raises:
228			Exception: If there is a struct unpacking error.
229
230			"""
231	1		try:
232	1		begin = offset
233	1		end = begin + self.length
234	1		unpacked_data = struct.unpack(self._fmt, buff[begin:end])[0]
235			except struct.error:
236			raise Exception("%s: %s" % (offset, buff))
237
238	1		self._value = unpacked_data.decode('ascii').rstrip('\0')
239
240
241	1		class IPAddress(GenericType):
242			"""Defines a IP address."""
243
244	1		netmask = UBInt32()
245	1		max_prefix = UBInt32(32)
246
247	1		def __init__(self, address="0.0.0.0/32"):
248			"""Create an IPAddress with the parameters below.
249
250			Args:
251			address (str): IP Address using ipv4. Defaults to '0.0.0.0/32'
252			"""
253	1		if address.find('/') >= 0:
254	1	View Code Duplication	address, netmask = address.split('/')
			0 ignored issues – show Duplication introduced 2017-09-05 21:03 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
255			else:
256	1		netmask = 32
257
258	1		super().__init__(address)
259	1		self.netmask = int(netmask)
260
261	1		def pack(self, value=None):
262			"""Pack the value as a binary representation.
263
264			If the value is None the self._value will be used to pack.
265
266			Args:
267			value (str): IP Address with ipv4 format.
268
269			Returns:
270			bytes: The binary representation.
271
272			Raises:
273			struct.error: If the value does not fit the binary format.
274
275			"""
276	1		if isinstance(value, type(self)):
277	1		return value.pack()
278
279	1		if value is None:
280	1		value = self._value
281
282	1		if value.find('/') >= 0:
283			value = value.split('/')[0]
284
285	1		try:
286	1		value = value.split('.')
287	1		return struct.pack('!4B', *[int(x) for x in value])
288			except struct.error as err:
289			msg = "IPAddress error. "
290			msg += "Class: {}, struct error: {} ".format(type(value).__name__,
291			err)
292			raise exceptions.PackException(msg)
293
294	1		def unpack(self, buff, offset=0):
295			"""Unpack a binary message into this object's attributes.
296
297			Unpack the binary value buff and update this object attributes based
298			on the results.
299
300			Args:
301			buff (bytes): Binary data package to be unpacked.
302			offset (int): Where to begin unpacking.
303
304			Raises:
305			Exception: If there is a struct unpacking error.
306
307			"""
308	1		try:
309	1		unpacked_data = struct.unpack('!4B', buff[offset:offset+4])
310	1		self._value = '.'.join([str(x) for x in unpacked_data])
311			except struct.error as exception:
312			raise exceptions.UnpackException('%s; %s: %s' % (exception,
313			offset, buff))
314
315	1		def get_size(self, value=None):
316			"""Return the ip address size in bytes.
317
318			Args:
319			value: In structs, the user can assign other value instead of
320			this class' instance. Here, in such cases, ``self`` is a class
321			attribute of the struct.
322
323			Returns:
324			int: The address size in bytes.
325
326			"""
327	1		return 4
328
329
330	1		class HWAddress(GenericType):
331		View Code Duplication	"""Defines a hardware address."""
			0 ignored issues – show Duplication introduced 2017-09-05 21:03 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
332
333			# pylint: disable=useless-super-delegation
334	1		def __init__(self, hw_address='00:00:00:00:00:00'):
335			"""Create a HWAddress with the parameters below.
336
337			Args:
338			hw_address (bytes): Hardware address. Defaults to
339			'00:00:00:00:00:00'.
340			"""
341	1		super().__init__(hw_address)
342
343	1		def pack(self, value=None):
344			"""Pack the value as a binary representation.
345
346			If the passed value (or the self._value) is zero (int), then the pack
347			will assume that the value to be packed is '00:00:00:00:00:00'.
348
349			Returns
350			bytes: The binary representation.
351
352			Raises:
353			struct.error: If the value does not fit the binary format.
354
355			"""
356	1		if isinstance(value, type(self)):
357	1		return value.pack()
358
359	1		if value is None:
360	1		value = self._value
361
362	1		if value == 0:
363			value = '00:00:00:00:00:00'
364
365	1		value = value.split(':')
366
367	1		try:
368	1		return struct.pack('!6B', *[int(x, 16) for x in value])
369			except struct.error as err:
370			msg = "HWAddress error. "
371			msg += "Class: {}, struct error: {} ".format(type(value).__name__,
372			err)
373			raise exceptions.PackException(msg)
374
375	1		def unpack(self, buff, offset=0):
376			"""Unpack a binary message into this object's attributes.
377
378			Unpack the binary value buff and update this object attributes based
379			on the results.
380
381			Args:
382			buff (bytes): Binary data package to be unpacked.
383			offset (int): Where to begin unpacking.
384
385			Raises:
386			Exception: If there is a struct unpacking error.
387
388			"""
389	1		def _int2hex(number):
390	1		return "{0:0{1}x}".format(number, 2)
391
392	1		try:
393	1		unpacked_data = struct.unpack('!6B', buff[offset:offset+6])
394			except struct.error as exception:
395			raise exceptions.UnpackException('%s; %s: %s' % (exception,
396			offset, buff))
397
398	1		transformed_data = ':'.join([_int2hex(x) for x in unpacked_data])
399	1		self._value = transformed_data
400
401	1		def get_size(self, value=None):
402			"""Return the address size in bytes.
403
404			Args:
405			value: In structs, the user can assign other value instead of
406			this class' instance. Here, in such cases, ``self`` is a class
407			attribute of the struct.
408
409			Returns:
410			int: The address size in bytes.
411
412			"""
413	1		return 6
414
415	1		def is_broadcast(self):
416			"""Return true if the value is a broadcast address. False otherwise."""
417			return self.value == 'ff:ff:ff:ff:ff:ff'
418
419
420	1		class BinaryData(GenericType):
421			"""Class to create objects that represent binary data.
422
423			This is used in the ``data`` attribute from
424			:class:`~pyof.v0x01.asynchronous.packet_in.PacketIn` and
425			:class:`~pyof.v0x01.controller2switch.packet_out.PacketOut` messages.
426			Both the :meth:`pack` and :meth:`unpack` methods will return the
427			binary data itself. :meth:`get_size` method will
428			return the size of the instance using Python's :func:`len`.
429			"""
430
431	1		def __init__(self, value=None): # noqa
432			"""The constructor takes the parameter below.
433
434			Args:
435			value (bytes): The binary data. Defaults to an empty value.
436
437			Raises:
438			ValueError: If given value is not bytes.
439
440			"""
441	1		value = self._pack(value)
442	1		super().__init__(value)
443
444	1		@staticmethod
445			def _pack(value):
446	1		if hasattr(value, 'pack') and callable(value.pack):
447	1		value = value.pack()
448	1		elif value is None:
449	1		value = b''
450
451	1		if not isinstance(value, bytes):
452	1		msg = 'BinaryData value must contain bytes or have pack method. '
453	1		msg += 'Received type {} value: "{}"'.format(type(value), value)
454	1		raise ValueError(msg)
455
456	1		return value
457
458	1		def pack(self, value=None):
459			"""Pack the value as a binary representation.
460
461			Returns:
462			bytes: The binary representation.
463
464			Raises:
465			:exc:`~.exceptions.NotBinaryData`: If value is not :class:`bytes`.
466
467			"""
468	1		if value is None:
469	1		return self._value
470	1		return self._pack(value)
471
472	1		def unpack(self, buff, offset=0):
473			"""Unpack a binary message into this object's attributes.
474
475			Unpack the binary value buff and update this object attributes based
476			on the results. Since the buff is binary data, no conversion is done.
477
478			Args:
479			buff (bytes): Binary data package to be unpacked.
480			offset (int): Where to begin unpacking.
481			"""
482	1		self._value = buff[offset:]
483
484	1		def get_size(self, value=None):
485			"""Return the size in bytes.
486
487			Args:
488			value (bytes): In structs, the user can assign other value instead
489			of this class' instance. Here, in such cases, ``self`` is a
490			class attribute of the struct.
491
492			Returns:
493			int: The address size in bytes.
494
495			"""
496	1		if value is None:
497	1		return len(self._value)
498	1		elif hasattr(value, 'get_size'):
499	1		return value.get_size()
500
501	1		return len(value)
502
503
504	1		class TypeList(list, GenericStruct):
505			"""Base class for lists that store objects of one single type."""
506
507	1		def __init__(self, items):
508			"""Initialize the list with one item or a list of items.
509
510			Args:
511			items (iterable, ``pyof_class``): Items to be stored.
512			"""
513	1		super().__init__()
514	1		if isinstance(items, list):
515	1		self.extend(items)
516	1		elif items:
517			self.append(items)
518
519	1		def extend(self, items):
520			"""Extend the list by adding all items of ``items``.
521
522			Args:
523			items (iterable): Items to be added to the list.
524
525			Raises:
526			:exc:`~.exceptions.WrongListItemType`: If an item has an unexpected
527			type.
528
529			"""
530	1		for item in items:
531	1		self.append(item)
532
533	1		def pack(self, value=None):
534			"""Pack the value as a binary representation.
535
536			Returns:
537			bytes: The binary representation.
538
539			"""
540	1		if isinstance(value, type(self)):
541	1		return value.pack()
542
543	1		if value is None:
544	1		value = self
545			else:
546	1		container = type(self)(items=None)
547	1		container.extend(value)
548	1		value = container
549
550	1		bin_message = b''
551	1		try:
552	1		for item in value:
553	1		bin_message += item.pack()
554	1		return bin_message
555			except exceptions.PackException as err:
556			msg = "{} pack error: {}".format(type(self).__name__, err)
557			raise exceptions.PackException(msg)
558
559			# pylint: disable=arguments-differ
560	1		def unpack(self, buff, item_class, offset=0):
561			"""Unpack the elements of the list.
562
563			This unpack method considers that all elements have the same size.
564			To use this class with a pyof_class that accepts elements with
565			different sizes, you must reimplement the unpack method.
566
567			Args:
568			buff (bytes): The binary data to be unpacked.
569			item_class (:obj:`type`): Class of the expected items on this list.
570			offset (int): If we need to shift the beginning of the data.
571			"""
572	1		begin = offset
573	1		limit_buff = len(buff)
574
575	1		while begin < limit_buff:
576	1		item = item_class()
577	1		item.unpack(buff, begin)
578	1		self.append(item)
579	1		begin += item.get_size()
580			# pylint: enable=arguments-differ
581
582	1		def get_size(self, value=None):
583			"""Return the size in bytes.
584
585			Args:
586			value: In structs, the user can assign other value instead of
587			this class' instance. Here, in such cases, ``self`` is a class
588			attribute of the struct.
589
590			Returns:
591			int: The size in bytes.
592
593			"""
594	1		if value is None:
595	1		if not self:
596			# If this is a empty list, then returns zero
597	1		return 0
598	1		elif issubclass(type(self[0]), GenericType):
599			# If the type of the elements is GenericType, then returns the
600			# length of the list multiplied by the size of the GenericType.
601			return len(self) * self[0].get_size()
602
603			# Otherwise iter over the list accumulating the sizes.
604	1		return sum(item.get_size() for item in self)
605
606	1		return type(self)(value).get_size()
607
608	1		def __str__(self):
609			"""Human-readable object representantion."""
610			return "{}".format([str(item) for item in self])
611
612
613	1		class FixedTypeList(TypeList):
614			"""A list that stores instances of one pyof class."""
615
616	1		_pyof_class = None
617
618	1		def __init__(self, pyof_class, items=None):
619			"""Create a FixedTypeList with the parameters follows.
620
621			Args:
622			pyof_class (:obj:`type`): Class of the items to be stored.
623			items (iterable, ``pyof_class``): Items to be stored.
624			"""
625	1		self._pyof_class = pyof_class
626	1		super().__init__(items)
627
628	1		def append(self, item):
629			"""Append one item to the list.
630
631			Args:
632			item: Item to be appended. Its type must match the one defined in
633			the constructor.
634
635			Raises:
636			:exc:`~.exceptions.WrongListItemType`: If the item has a different
637			type than the one specified in the constructor.
638
639			"""
640	1		if isinstance(item, list):
641			self.extend(item)
642	1		elif issubclass(item.__class__, self._pyof_class):
643	1		list.append(self, item)
644			else:
645			raise exceptions.WrongListItemType(item.__class__.__name__,
646			self._pyof_class.__name__)
647
648	1		def insert(self, index, item):
649			"""Insert an item at the specified index.
650
651			Args:
652			index (int): Position to insert the item.
653			item: Item to be inserted. It must have the type specified in the
654			constructor.
655
656			Raises:
657			:exc:`~.exceptions.WrongListItemType`: If the item has a different
658			type than the one specified in the constructor.
659
660			"""
661			if issubclass(item.__class__, self._pyof_class):
662			list.insert(self, index, item)
663			else:
664			raise exceptions.WrongListItemType(item.__class__.__name__,
665			self._pyof_class.__name__)
666
667	1		def unpack(self, buff, offset=0): # pylint: disable=arguments-differ
668			"""Unpack the elements of the list.
669
670			This unpack method considers that all elements have the same size.
671			To use this class with a pyof_class that accepts elements with
672			different sizes, you must reimplement the unpack method.
673
674			Args:
675			buff (bytes): The binary data to be unpacked.
676			offset (int): If we need to shift the beginning of the data.
677			"""
678	1		super().unpack(buff, self._pyof_class, offset)
679
680
681	1		class ConstantTypeList(TypeList):
682			"""List that contains only objects of the same type (class).
683
684			The types of all items are expected to be the same as the first item's.
685			Otherwise, :exc:`~.exceptions.WrongListItemType` is raised in many
686			list operations.
687			"""
688
689			# pylint: disable=useless-super-delegation
690	1		def __init__(self, items=None):
691			"""Create a ConstantTypeList that can contain itens to be stored.
692
693			Args:
694			items (iterable, :class:`object`): Items to be stored.
695
696			Raises:
697			:exc:`~.exceptions.WrongListItemType`: If an item has a different
698			type than the first item to be stored.
699
700			"""
701			super().__init__(items)
702
703	1		def append(self, item):
704			"""Append one item to the list.
705
706			Args:
707			item: Item to be appended.
708
709			Raises:
710			:exc:`~.exceptions.WrongListItemType`: If an item has a different
711			type than the first item to be stored.
712
713			"""
714			if isinstance(item, list):
715			self.extend(item)
716			elif not self:
717			list.append(self, item)
718			elif item.__class__ == self[0].__class__:
719			list.append(self, item)
720			else:
721			raise exceptions.WrongListItemType(item.__class__.__name__,
722			self[0].__class__.__name__)
723
724	1		def insert(self, index, item):
725			"""Insert an item at the specified index.
726
727			Args:
728			index (int): Position to insert the item.
729			item: Item to be inserted.
730
731			Raises:
732			:exc:`~.exceptions.WrongListItemType`: If an item has a different
733			type than the first item to be stored.
734
735			"""
736			if not self:
737			list.append(self, item)
738			elif item.__class__ == self[0].__class__:
739			list.insert(self, index, item)
740			else:
741			raise exceptions.WrongListItemType(item.__class__.__name__,
742			self[0].__class__.__name__)
743

kytos / python-openflow

Pull Request — master (#445)

BinaryData.pack() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like