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-19 13:38 UTC

BinaryData.pack() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	10
CRAP Score	5

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	22
ccs	10
cts	10
cp	1
rs	8.3411
cc	5
crap	5

"""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):  # pylint: disable=useless-super-delegation
        """The constructor takes the parameter below.

        Args:
            value (bytes): The binary data. Defaults to an empty value.
        """
        super().__init__(value)

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

        Returns:
            bytes: The binary representation.

        Raises:
            ValueError: If value can't be represented with bytes

        """
        if value is None:
            value = self._value

        if hasattr(value, 'pack') and callable(value.pack):
            return value.pack()
        elif isinstance(value, bytes):
            return value
        elif value is None:
            return b''
        else:
            raise ValueError(f"BinaryData can't be {type(value)} = '{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:
            value = self._value
        elif hasattr(value, 'get_size'):
            return value.get_size()

        return len(self.pack(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): # pylint: disable=useless-super-delegation
432			"""The constructor takes the parameter below.
433
434			Args:
435			value (bytes): The binary data. Defaults to an empty value.
436			"""
437	1		super().__init__(value)
438
439	1		def pack(self, value=None):
440			"""Pack the value as a binary representation.
441
442			Returns:
443			bytes: The binary representation.
444
445			Raises:
446			ValueError: If value can't be represented with bytes
447
448			"""
449	1		if value is None:
450	1		value = self._value
451
452	1		if hasattr(value, 'pack') and callable(value.pack):
453	1		return value.pack()
454	1		elif isinstance(value, bytes):
455	1		return value
456	1		elif value is None:
457	1		return b''
458			else:
459	1		raise ValueError(f"BinaryData can't be {type(value)} = '{value}'")
460
461	1		def unpack(self, buff, offset=0):
462			"""Unpack a binary message into this object's attributes.
463
464			Unpack the binary value buff and update this object attributes based
465			on the results. Since the buff is binary data, no conversion is done.
466
467			Args:
468			buff (bytes): Binary data package to be unpacked.
469			offset (int): Where to begin unpacking.
470			"""
471	1		self._value = buff[offset:]
472
473	1		def get_size(self, value=None):
474			"""Return the size in bytes.
475
476			Args:
477			value (bytes): In structs, the user can assign other value instead
478			of this class' instance. Here, in such cases, ``self`` is a
479			class attribute of the struct.
480
481			Returns:
482			int: The address size in bytes.
483
484			"""
485	1		if value is None:
486	1		value = self._value
487	1		elif hasattr(value, 'get_size'):
488	1		return value.get_size()
489
490	1		return len(self.pack(value))
491
492
493	1		class TypeList(list, GenericStruct):
494			"""Base class for lists that store objects of one single type."""
495
496	1		def __init__(self, items):
497			"""Initialize the list with one item or a list of items.
498
499			Args:
500			items (iterable, ``pyof_class``): Items to be stored.
501			"""
502	1		super().__init__()
503	1		if isinstance(items, list):
504	1		self.extend(items)
505	1		elif items:
506			self.append(items)
507
508	1		def extend(self, items):
509			"""Extend the list by adding all items of ``items``.
510
511			Args:
512			items (iterable): Items to be added to the list.
513
514			Raises:
515			:exc:`~.exceptions.WrongListItemType`: If an item has an unexpected
516			type.
517
518			"""
519	1		for item in items:
520	1		self.append(item)
521
522	1		def pack(self, value=None):
523			"""Pack the value as a binary representation.
524
525			Returns:
526			bytes: The binary representation.
527
528			"""
529	1		if isinstance(value, type(self)):
530	1		return value.pack()
531
532	1		if value is None:
533	1		value = self
534			else:
535	1		container = type(self)(items=None)
536	1		container.extend(value)
537	1		value = container
538
539	1		bin_message = b''
540	1		try:
541	1		for item in value:
542	1		bin_message += item.pack()
543	1		return bin_message
544			except exceptions.PackException as err:
545			msg = "{} pack error: {}".format(type(self).__name__, err)
546			raise exceptions.PackException(msg)
547
548			# pylint: disable=arguments-differ
549	1		def unpack(self, buff, item_class, offset=0):
550			"""Unpack the elements of the list.
551
552			This unpack method considers that all elements have the same size.
553			To use this class with a pyof_class that accepts elements with
554			different sizes, you must reimplement the unpack method.
555
556			Args:
557			buff (bytes): The binary data to be unpacked.
558			item_class (:obj:`type`): Class of the expected items on this list.
559			offset (int): If we need to shift the beginning of the data.
560			"""
561	1		begin = offset
562	1		limit_buff = len(buff)
563
564	1		while begin < limit_buff:
565	1		item = item_class()
566	1		item.unpack(buff, begin)
567	1		self.append(item)
568	1		begin += item.get_size()
569			# pylint: enable=arguments-differ
570
571	1		def get_size(self, value=None):
572			"""Return the size in bytes.
573
574			Args:
575			value: In structs, the user can assign other value instead of
576			this class' instance. Here, in such cases, ``self`` is a class
577			attribute of the struct.
578
579			Returns:
580			int: The size in bytes.
581
582			"""
583	1		if value is None:
584	1		if not self:
585			# If this is a empty list, then returns zero
586	1		return 0
587	1		elif issubclass(type(self[0]), GenericType):
588			# If the type of the elements is GenericType, then returns the
589			# length of the list multiplied by the size of the GenericType.
590			return len(self) * self[0].get_size()
591
592			# Otherwise iter over the list accumulating the sizes.
593	1		return sum(item.get_size() for item in self)
594
595	1		return type(self)(value).get_size()
596
597	1		def __str__(self):
598			"""Human-readable object representantion."""
599			return "{}".format([str(item) for item in self])
600
601
602	1		class FixedTypeList(TypeList):
603			"""A list that stores instances of one pyof class."""
604
605	1		_pyof_class = None
606
607	1		def __init__(self, pyof_class, items=None):
608			"""Create a FixedTypeList with the parameters follows.
609
610			Args:
611			pyof_class (:obj:`type`): Class of the items to be stored.
612			items (iterable, ``pyof_class``): Items to be stored.
613			"""
614	1		self._pyof_class = pyof_class
615	1		super().__init__(items)
616
617	1		def append(self, item):
618			"""Append one item to the list.
619
620			Args:
621			item: Item to be appended. Its type must match the one defined in
622			the constructor.
623
624			Raises:
625			:exc:`~.exceptions.WrongListItemType`: If the item has a different
626			type than the one specified in the constructor.
627
628			"""
629	1		if isinstance(item, list):
630			self.extend(item)
631	1		elif issubclass(item.__class__, self._pyof_class):
632	1		list.append(self, item)
633			else:
634			raise exceptions.WrongListItemType(item.__class__.__name__,
635			self._pyof_class.__name__)
636
637	1		def insert(self, index, item):
638			"""Insert an item at the specified index.
639
640			Args:
641			index (int): Position to insert the item.
642			item: Item to be inserted. It must have the type specified in the
643			constructor.
644
645			Raises:
646			:exc:`~.exceptions.WrongListItemType`: If the item has a different
647			type than the one specified in the constructor.
648
649			"""
650			if issubclass(item.__class__, self._pyof_class):
651			list.insert(self, index, item)
652			else:
653			raise exceptions.WrongListItemType(item.__class__.__name__,
654			self._pyof_class.__name__)
655
656	1		def unpack(self, buff, offset=0): # pylint: disable=arguments-differ
657			"""Unpack the elements of the list.
658
659			This unpack method considers that all elements have the same size.
660			To use this class with a pyof_class that accepts elements with
661			different sizes, you must reimplement the unpack method.
662
663			Args:
664			buff (bytes): The binary data to be unpacked.
665			offset (int): If we need to shift the beginning of the data.
666			"""
667	1		super().unpack(buff, self._pyof_class, offset)
668
669
670	1		class ConstantTypeList(TypeList):
671			"""List that contains only objects of the same type (class).
672
673			The types of all items are expected to be the same as the first item's.
674			Otherwise, :exc:`~.exceptions.WrongListItemType` is raised in many
675			list operations.
676			"""
677
678			# pylint: disable=useless-super-delegation
679	1		def __init__(self, items=None):
680			"""Create a ConstantTypeList that can contain itens to be stored.
681
682			Args:
683			items (iterable, :class:`object`): Items to be stored.
684
685			Raises:
686			:exc:`~.exceptions.WrongListItemType`: If an item has a different
687			type than the first item to be stored.
688
689			"""
690			super().__init__(items)
691
692	1		def append(self, item):
693			"""Append one item to the list.
694
695			Args:
696			item: Item to be appended.
697
698			Raises:
699			:exc:`~.exceptions.WrongListItemType`: If an item has a different
700			type than the first item to be stored.
701
702			"""
703			if isinstance(item, list):
704			self.extend(item)
705			elif not self:
706			list.append(self, item)
707			elif item.__class__ == self[0].__class__:
708			list.append(self, item)
709			else:
710			raise exceptions.WrongListItemType(item.__class__.__name__,
711			self[0].__class__.__name__)
712
713	1		def insert(self, index, item):
714			"""Insert an item at the specified index.
715
716			Args:
717			index (int): Position to insert the item.
718			item: Item to be inserted.
719
720			Raises:
721			:exc:`~.exceptions.WrongListItemType`: If an item has a different
722			type than the first item to be stored.
723
724			"""
725			if not self:
726			list.append(self, item)
727			elif item.__class__ == self[0].__class__:
728			list.insert(self, index, item)
729			else:
730			raise exceptions.WrongListItemType(item.__class__.__name__,
731			self[0].__class__.__name__)
732

kytos / python-openflow

Pull Request — master (#445)

BinaryData.pack() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like