|
1
|
|
|
"""Automate struct tests.""" |
|
2
|
|
|
import unittest |
|
3
|
|
|
|
|
4
|
|
|
from pyof.v0x04.common.header import Header |
|
5
|
|
|
from pyof.v0x04.common.utils import new_message_from_header |
|
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(TestStruct): |
|
25
|
|
|
@classmethod |
|
26
|
|
|
def setUpClass(cls): |
|
27
|
|
|
super().setUpClass() |
|
28
|
|
|
# Create BarrierReply(xid=5) |
|
29
|
|
|
super().set_message(BarrierReply, xid=5) |
|
30
|
|
|
# As in spec: ``OFP_ASSERT(sizeof(struct ...) == ...);`` |
|
31
|
|
|
super().set_minimum_size(8) |
|
32
|
|
|
|
|
33
|
|
|
To only test the minimum size and skip packing/unpacking: |
|
34
|
|
|
|
|
35
|
|
|
.. code-block:: python3 |
|
36
|
|
|
class MyTest(TestStruct): |
|
37
|
|
|
@classmethod |
|
38
|
|
|
def setUpClass(cls): |
|
39
|
|
|
super().set_message(BarrierReply) |
|
40
|
|
|
super().set_minimum_size(8) |
|
41
|
|
|
""" |
|
42
|
|
|
|
|
43
|
|
|
def __init__(self, *args, **kwargs): |
|
44
|
|
|
"""The constructor will 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
|
|
|
_msg_cls = None |
|
56
|
|
|
_msg_params = None |
|
57
|
|
|
_min_size = None |
|
58
|
|
|
|
|
59
|
|
|
@classmethod |
|
60
|
|
|
def set_message(cls, msg_cls, *args, **kwargs): |
|
61
|
|
|
"""Set how to create the message object. |
|
62
|
|
|
|
|
63
|
|
|
Args: |
|
64
|
|
|
msg_class (:obj:`type`): The message class followed by its |
|
65
|
|
|
parameters to instantiate an object. |
|
66
|
|
|
|
|
67
|
|
|
Example: |
|
68
|
|
|
``super().__init__(BarrierReply, xid=5)`` will create |
|
69
|
|
|
``BarrierReply(xid=5)``. |
|
70
|
|
|
""" |
|
71
|
|
|
TestStruct._msg_cls = msg_cls |
|
72
|
|
|
cls._msg_params = (args, kwargs) |
|
73
|
|
|
|
|
74
|
|
|
@classmethod |
|
75
|
|
|
def set_minimum_size(cls, size): |
|
76
|
|
|
"""Set the struct minimum size (from spec). |
|
77
|
|
|
|
|
78
|
|
|
The minimum size can be found in OF spec. For example, |
|
79
|
|
|
:class:`.PhyPort` minimum size is 48 because of |
|
80
|
|
|
``OFP_ASSERT(sizeof(struct ofp_phy_port) == 48);`` (spec 1.0.0). |
|
81
|
|
|
|
|
82
|
|
|
Args: |
|
83
|
|
|
size (int): The minimum size of the struct, in bytes. |
|
84
|
|
|
""" |
|
85
|
|
|
cls._min_size = size |
|
86
|
|
|
|
|
87
|
|
|
def test_pack_unpack(self): |
|
88
|
|
|
"""Pack the message, unpack and check whether they are the same.""" |
|
89
|
|
|
if self._msg_cls: |
|
90
|
|
|
args, kwargs = self._msg_params |
|
91
|
|
|
self._test_pack_unpack(*args, **kwargs) |
|
92
|
|
|
|
|
93
|
|
|
def _test_pack_unpack(self, *args, **kwargs): |
|
94
|
|
|
"""Pack the message, unpack and check whether they are the same. |
|
95
|
|
|
|
|
96
|
|
|
Call this method multiple times if you want to test more than one |
|
97
|
|
|
object. |
|
98
|
|
|
""" |
|
99
|
|
|
obj = self._msg_cls(*args, **kwargs) |
|
100
|
|
|
packed = obj.pack() |
|
101
|
|
|
header = Header() |
|
102
|
|
|
header_size = header.get_size() |
|
103
|
|
|
header.unpack(packed[:header_size]) |
|
104
|
|
|
unpacked = new_message_from_header(header) |
|
105
|
|
|
unpacked.unpack(packed[header_size:]) |
|
106
|
|
|
|
|
107
|
|
|
self.assertEqual(packed, unpacked.pack()) |
|
108
|
|
|
|
|
109
|
|
|
def test_minimum_size(self): |
|
110
|
|
|
"""Test struct minimum size.""" |
|
111
|
|
|
if self._min_size is None: |
|
112
|
|
|
raise self.skipTest('minimum size was not set.') |
|
113
|
|
|
obj = TestStruct._msg_cls() |
|
114
|
|
|
self.assertEqual(obj.get_size(), self._min_size) |
|
115
|
|
|
|
kytos /
python-openflow
