tests.unit.test_struct.TestStruct._test_unpack() - Code Metrics - Inspection of "Merge pull request #606 from gleybersonandrade/tag..." - kytos/python-openflow - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( b50e97...feedcb )

by Humberto

created 2020-06-18 18:27 UTC

tests.unit.test_struct.TestStruct._test_unpack() A

↳ Parent: tests.unit.test_struct

Complexity

Conditions

Size

Total Lines	20
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	10
dl	0
loc	20
rs	9.9
c	0
b	0
f	0
cc	2
nop	3

"""Automate struct tests."""
import unittest

from pyof.foundation.base import GenericMessage
from tests.unit.raw_dump import RawDump


class TestStruct(unittest.TestCase):
    """Run tests related to struct packing and unpacking.

    Test the lib with raw dump files from an OpenFlow switch. We assume the
    raw files are valid according to the OF specs to check whether our pack and
    unpack implementations are correct.

    Also, check the minimum size of the struct by instantiating an object with
    no parameters.

    To run these tests, just extends this class and call 2 methods in the
    ``setUp`` method like the example.

    Example:
        .. code-block:: python3

            class MyTest(TestDump):
                @classmethod
                def setUpClass(cls):
                    super().setUpClass()
                    super().set_raw_dump_file('v0x01', 'ofpt_barrier_reply')
                    # Create BarrierReply(xid=5) when needed
                    super().set_raw_dump_object(BarrierReply, xid=5)
                    # As in spec: ``OFP_ASSERT(sizeof(struct ...) == ...);``
                    super().set_minimum_size(8)

        To only test the minimum size and skip packing/unpacking:

        .. code-block:: python3
            class MyTest(TestDump):
                @classmethod
                def setUpClass(cls):
                    super().set_minimum_size(8, BarrierReply)
    """

    def __init__(self, *args, **kwargs):
        """Avoid that this class tests are executed.

        The tests in this class are executed through the child, so there's no
        no need for them to be executed once more through the parent.
        """
        super().__init__(*args, **kwargs)
        # Override the run method, so it does nothing instead of running the
        # tests (again).
        if self.__class__ == TestStruct:
            self.run = lambda *args, **kwargs: None

    _new_raw_dump = None
    _new_raw_object = None
    _msg_cls = None
    _min_size = None

    @classmethod
    def set_raw_dump_file(cls, version, basename):
        """Set which raw dump the tests will use.

        Args:
            protocol_version (str): OpenFlow protocol version,
                e.g. ``v0x01``.
            basename (str): The raw filename without extension.
                E.g. ``ofpt_echo_reply``.
        """
        cls._new_raw_dump = lambda: RawDump(version, basename)

    @classmethod
    def get_raw_dump(cls):
        """Return a new instance of :class:`.RawDump`.

        Use the parameters set in :meth:`set_raw_dump_file`.

        Returns:
            RawDump: with parameters previously set using
                :meth:`set_raw_dump_file`.

        """
        if cls._new_raw_dump is None:
            raise FileNotFoundError()
        return cls._new_raw_dump()

    @classmethod
    def set_raw_dump_object(cls, msg_cls, *args, **kwargs):
        """Set how to create the object that is dumped in a raw file.

        Args:
            msg_class (:obj:`type`): The message class that is packed as a
                raw file, followed by its parameters to instantiate an
                object.

        Example:
            ``super().__init__(BarrierReply, xid=5)`` will create
            ``BarrierReply(xid=5)``.
        """
        TestStruct._msg_cls = msg_cls
        cls._new_raw_object = lambda: msg_cls(*args, **kwargs)

    @classmethod
    def get_raw_object(cls):
        """Create a new object of the dumped message.

        Use the class and parameters set in :meth:`set_raw_dump_object`.

        Returns:
            A new object using class and parameters priviously set through
                :meth:`set_raw_dump_object`.

        """
        pyof_obj = cls._new_raw_object()
        if isinstance(pyof_obj, GenericMessage):
            pyof_obj.update_header_length()
        return pyof_obj

    @classmethod
    def set_minimum_size(cls, size, msg_cls=None):
        """Set the struct minimum size.

        The minimum size can be found in OF spec. For example,
        :class:`.PhyPort` minimum size is 48 because of
        ``OFP_ASSERT(sizeof(struct ofp_phy_port) == 48);`` (spec 1.0.0).

        Args:
            size (int): The minimum size of the struct, in bytes.
            msg_cls (class): The class (or function) to have its size checked.
                If None, use the same class set in :meth:`set_raw_dump_object`.
        """
        cls._min_size = size
        if msg_cls is not None:
            TestStruct._msg_cls = msg_cls

    def _test_pack(self, obj, expected_bytes):
        """Check whether packed objects equals to dump file."""
        actual_bytes = obj.pack()
        self.assertSequenceEqual(expected_bytes, actual_bytes)

    def test_raw_dump_file(self):
        """Object pack should equal file; file unpack should equal object.

        The object to be packed is set with :method:`set_raw_object` and the
        file, with :method:`set_raw_dump_file`.
        """
        try:
            file_bytes = self.get_raw_dump().read()
        except FileNotFoundError:
            raise self.skipTest('No raw dump file found.')

        pyof_obj = self.get_raw_object()
        self._test_pack(pyof_obj, file_bytes)
        self._test_unpack(pyof_obj, file_bytes)

    def _test_unpack(self, pyof_obj, bytes2unpack=None):
        """Check whether unpacking ``bytes2unpack`` equals ``pyof_obj``.

        Args:
            pyof_obj (GenericStruct, GenericType): Object supporting (un)pack
                operations.
            bytes2unpack (bytes): If not supplied, use ``pyof_obj.pack()``.
        """
        bytes2unpack = bytes2unpack or pyof_obj.pack()

        unpacked = type(pyof_obj)()
        # If it's a GenericMessage, unpack the Header first
        if isinstance(pyof_obj, GenericMessage):
            header_bytes = bytes2unpack[:8]
            unpacked.header.unpack(header_bytes)
            bytes2unpack = bytes2unpack[8:unpacked.header.length.value]
        unpacked.unpack(bytes2unpack)

        self.assertEqual(pyof_obj, unpacked)
        self.assertEqual(pyof_obj.get_size(), unpacked.get_size())

    def test_minimum_size(self):
        """Test struct minimum size."""
        if self._min_size is None:
            raise self.skipTest('minimum size was not set.')
        obj = TestStruct._msg_cls()
        self.assertEqual(obj.get_size(), self._min_size)


1			"""Automate struct tests."""
2			import unittest
3
4			from pyof.foundation.base import GenericMessage
5			from tests.unit.raw_dump import RawDump
6
7
8			class TestStruct(unittest.TestCase):
9			"""Run tests related to struct packing and unpacking.
10
11			Test the lib with raw dump files from an OpenFlow switch. We assume the
12			raw files are valid according to the OF specs to check whether our pack and
13			unpack implementations are correct.
14
15			Also, check the minimum size of the struct by instantiating an object with
16			no parameters.
17
18			To run these tests, just extends this class and call 2 methods in the
19			``setUp`` method like the example.
20
21			Example:
22			.. code-block:: python3
23
24			class MyTest(TestDump):
25			@classmethod
26			def setUpClass(cls):
27			super().setUpClass()
28			super().set_raw_dump_file('v0x01', 'ofpt_barrier_reply')
29			# Create BarrierReply(xid=5) when needed
30			super().set_raw_dump_object(BarrierReply, xid=5)
31			# As in spec: ``OFP_ASSERT(sizeof(struct ...) == ...);``
32			super().set_minimum_size(8)
33
34			To only test the minimum size and skip packing/unpacking:
35
36			.. code-block:: python3
37			class MyTest(TestDump):
38			@classmethod
39			def setUpClass(cls):
40			super().set_minimum_size(8, BarrierReply)
41			"""
42
43			def __init__(self, args, *kwargs):
44			"""Avoid that this class tests are executed.
45
46			The tests in this class are executed through the child, so there's no
47			no need for them to be executed once more through the parent.
48			"""
49			super().__init__(args, *kwargs)
50			# Override the run method, so it does nothing instead of running the
51			# tests (again).
52			if self.__class__ == TestStruct:
53			self.run = lambda args, *kwargs: None
54
55			_new_raw_dump = None
56			_new_raw_object = None
57			_msg_cls = None
58			_min_size = None
59
60			@classmethod
61			def set_raw_dump_file(cls, version, basename):
62			"""Set which raw dump the tests will use.
63
64			Args:
65			protocol_version (str): OpenFlow protocol version,
66			e.g. ``v0x01``.
67			basename (str): The raw filename without extension.
68			E.g. ``ofpt_echo_reply``.
69			"""
70			cls._new_raw_dump = lambda: RawDump(version, basename)
71
72			@classmethod
73			def get_raw_dump(cls):
74			"""Return a new instance of :class:`.RawDump`.
75
76			Use the parameters set in :meth:`set_raw_dump_file`.
77
78			Returns:
79			RawDump: with parameters previously set using
80			:meth:`set_raw_dump_file`.
81
82			"""
83			if cls._new_raw_dump is None:
84			raise FileNotFoundError()
85			return cls._new_raw_dump()
86
87			@classmethod
88			def set_raw_dump_object(cls, msg_cls, args, *kwargs):
89			"""Set how to create the object that is dumped in a raw file.
90
91			Args:
92			msg_class (:obj:`type`): The message class that is packed as a
93			raw file, followed by its parameters to instantiate an
94			object.
95
96			Example:
97			``super().__init__(BarrierReply, xid=5)`` will create
98			``BarrierReply(xid=5)``.
99			"""
100			TestStruct._msg_cls = msg_cls
101			cls._new_raw_object = lambda: msg_cls(args, *kwargs)
102
103			@classmethod
104			def get_raw_object(cls):
105			"""Create a new object of the dumped message.
106
107			Use the class and parameters set in :meth:`set_raw_dump_object`.
108
109			Returns:
110			A new object using class and parameters priviously set through
111			:meth:`set_raw_dump_object`.
112
113			"""
114			pyof_obj = cls._new_raw_object()
115			if isinstance(pyof_obj, GenericMessage):
116			pyof_obj.update_header_length()
117			return pyof_obj
118
119			@classmethod
120			def set_minimum_size(cls, size, msg_cls=None):
121			"""Set the struct minimum size.
122
123			The minimum size can be found in OF spec. For example,
124			:class:`.PhyPort` minimum size is 48 because of
125			``OFP_ASSERT(sizeof(struct ofp_phy_port) == 48);`` (spec 1.0.0).
126
127			Args:
128			size (int): The minimum size of the struct, in bytes.
129			msg_cls (class): The class (or function) to have its size checked.
130			If None, use the same class set in :meth:`set_raw_dump_object`.
131			"""
132			cls._min_size = size
133			if msg_cls is not None:
134			TestStruct._msg_cls = msg_cls
135
136			def _test_pack(self, obj, expected_bytes):
137			"""Check whether packed objects equals to dump file."""
138			actual_bytes = obj.pack()
139			self.assertSequenceEqual(expected_bytes, actual_bytes)
140
141			def test_raw_dump_file(self):
142			"""Object pack should equal file; file unpack should equal object.
143
144			The object to be packed is set with :method:`set_raw_object` and the
145			file, with :method:`set_raw_dump_file`.
146			"""
147			try:
148			file_bytes = self.get_raw_dump().read()
149			except FileNotFoundError:
150			raise self.skipTest('No raw dump file found.')
151
152			pyof_obj = self.get_raw_object()
153			self._test_pack(pyof_obj, file_bytes)
154			self._test_unpack(pyof_obj, file_bytes)
155
156			def _test_unpack(self, pyof_obj, bytes2unpack=None):
157			"""Check whether unpacking ``bytes2unpack`` equals ``pyof_obj``.
158
159			Args:
160			pyof_obj (GenericStruct, GenericType): Object supporting (un)pack
161			operations.
162			bytes2unpack (bytes): If not supplied, use ``pyof_obj.pack()``.
163			"""
164			bytes2unpack = bytes2unpack or pyof_obj.pack()
165
166			unpacked = type(pyof_obj)()
167			# If it's a GenericMessage, unpack the Header first
168			if isinstance(pyof_obj, GenericMessage):
169			header_bytes = bytes2unpack[:8]
170			unpacked.header.unpack(header_bytes)
171			bytes2unpack = bytes2unpack[8:unpacked.header.length.value]
172			unpacked.unpack(bytes2unpack)
173
174			self.assertEqual(pyof_obj, unpacked)
175			self.assertEqual(pyof_obj.get_size(), unpacked.get_size())
176
177			def test_minimum_size(self):
178			"""Test struct minimum size."""
179			if self._min_size is None:
180			raise self.skipTest('minimum size was not set.')
181			obj = TestStruct._msg_cls()
182			self.assertEqual(obj.get_size(), self._min_size)
183

kytos / python-openflow

Push — master ( b50e97...feedcb )

tests.unit.test_struct.TestStruct._test_unpack() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like