Char() - Code Metrics - Inspection of "Teststruct improvement application" - kytos/python-openflow - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#404)

unknown

created 2017-07-14 15:26 UTC

Char() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	1.064

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
dl	0
loc	8
ccs	3
cts	5
cp	0.6
rs	9.4285
c	1
b	0
f	0
cc	1
crap	1.064

"""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 CharStringBase(GenericType):
    """A null terminated ascii char string."""

    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')


def Char(value=None, length=0):
    string_length = length

    class CharString(CharStringBase):
        length = string_length
        _fmt = '!{}{}'.format(length, 's')

    return CharString(value)


class IPAddress(GenericType):
    """Defines a IP address."""

    netmask = UBInt32()
    max_prefix = UBInt32(32)

    def __init__(self, address="0.0.0.0/32"):
        """The constructor takes the parameters below.

        Args:
            address (str): IP Address using ipv4. Defaults to '0.0.0.0/32'
        """
        if address.find('/') >= 0:
            address, netmask = address.split('/')
        else:
            netmask = 32

        super().__init__(address)
        self.netmask = int(netmask)

    def pack(self, value=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 e:
            raise exceptions.UnpackException('%s; %s: %s' % (e, 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."""

    def __init__(self, hw_address='00:00:00:00:00:00'):  # noqa
        """The constructor takes the parameters below.

        Args:
            hw_address (bytes): Hardware address. Defaults to
                '00:00:00:00:00:00'.
        """
        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(n):
            return "{0:0{1}x}".format(n, 2)

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

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

    def get_size(self, 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=b''):  # noqa
        """The constructor takes the parameter below.

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

        Raises:
            ValueError: If given value is not bytes.
        """
        if not isinstance(value, bytes):
            raise ValueError('BinaryData must contain bytes.')
        super().__init__(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 isinstance(value, type(self)):
            return value.pack()

        if value is None:
            value = self._value

        if value:
            if isinstance(value, bytes):
                return value
            raise ValueError('BinaryData must contain bytes.')

        return b''

    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)

    def unpack(self, buff, item_class, offset=0):
        """Unpack the elements of the list.

        This unpack method considers that all elements have the same size.
        To use this class with a pyof_class that accepts elements with
        different sizes, you must reimplement the unpack method.

        Args:
            buff (bytes): The binary data to be unpacked.
            item_class (:obj:`type`): Class of the expected items on this list.
            offset (int): If we need to shift the beginning of the data.
        """
        begin = offset
        limit_buff = len(buff)

        while begin < limit_buff:
            item = item_class()
            item.unpack(buff, begin)
            self.append(item)
            begin += item.get_size()

    def get_size(self, 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):
        """The constructor parameters follows.

        Args:
            pyof_class (:obj:`type`): Class of the items to be stored.
            items (iterable, ``pyof_class``): Items to be stored.
        """
        self._pyof_class = pyof_class
        super().__init__(items)

    def append(self, item):
        """Append one item to the list.

        Args:
            item: Item to be appended. Its type must match the one defined in
                the constructor.

        Raises:
            :exc:`~.exceptions.WrongListItemType`: If the item has a different
                type than the one specified in the constructor.
        """
        if isinstance(item, list):
            self.extend(item)
        elif issubclass(item.__class__, self._pyof_class):
            list.append(self, item)
        else:
            raise exceptions.WrongListItemType(item.__class__.__name__,
                                               self._pyof_class.__name__)

    def insert(self, index, item):
        """Insert an item at the specified index.

        Args:
            index (int): Position to insert the item.
            item: Item to be inserted. It must have the type specified in the
                constructor.

        Raises:
            :exc:`~.exceptions.WrongListItemType`: If the item has a different
                type than the one specified in the constructor.
        """
        if issubclass(item.__class__, self._pyof_class):
            list.insert(self, index, item)
        else:
            raise exceptions.WrongListItemType(item.__class__.__name__,
                                               self._pyof_class.__name__)

    def unpack(self, buff, offset=0):
        """Unpack the elements of the list.

        This unpack method considers that all elements have the same size.
        To use this class with a pyof_class that accepts elements with
        different sizes, you must reimplement the unpack method.

        Args:
            buff (bytes): The binary data to be unpacked.
            offset (int): If we need to shift the beginning of the data.
        """
        super().unpack(buff, self._pyof_class, offset)


class ConstantTypeList(TypeList):
    """List that contains only objects of the same type (class).

    The types of all items are expected to be the same as the first item's.
    Otherwise, :exc:`~.exceptions.WrongListItemType` is raised in many
    list operations.
    """

    def __init__(self, items=None):  # noqa
        """The contructor can contain the items to be stored.

        Args:
            items (iterable, :class:`object`): Items to be stored.

        Raises:
            :exc:`~.exceptions.WrongListItemType`: If an item has a different
                type than the first item to be stored.
        """
        super().__init__(items)

    def append(self, item):
        """Append one item to the list.

        Args:
            item: Item to be appended.

        Raises:
            :exc:`~.exceptions.WrongListItemType`: If an item has a different
                type than the first item to be stored.
        """
        if isinstance(item, list):
            self.extend(item)
        elif not self:
            list.append(self, item)
        elif item.__class__ == self[0].__class__:
            list.append(self, item)
        else:
            raise exceptions.WrongListItemType(item.__class__.__name__,
                                               self[0].__class__.__name__)

    def insert(self, index, item):
        """Insert an item at the specified index.

        Args:
            index (int): Position to insert the item.
            item: Item to be inserted.

        Raises:
            :exc:`~.exceptions.WrongListItemType`: If an item has a different
                type than the first item to be stored.
        """
        if not self:
            list.append(self, item)
        elif item.__class__ == self[0].__class__:
            list.insert(self, index, item)
        else:
            raise exceptions.WrongListItemType(item.__class__.__name__,
                                               self[0].__class__.__name__)


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

kytos / python-openflow

Pull Request — master (#404)

Char() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like