BinaryData.pack() - Code Metrics - Inspection of "Merge pull request #445 from macartur/binarydata_f..." - kytos/python-openflow - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 61e2ba...fcdfda )

by Beraldo

created 2017-09-19 20:54 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

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

kytos / python-openflow

Push — master ( 61e2ba...fcdfda )

BinaryData.pack() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like