|
1
|
|
|
"""Automate struct tests.""" |
|
2
|
|
|
import unittest |
|
3
|
|
|
|
|
4
|
|
|
from tests.raw_dump import RawDump |
|
5
|
|
|
|
|
6
|
|
|
|
|
7
|
|
|
class TestStruct(unittest.TestCase): |
|
8
|
|
|
"""Run tests related to struct packing and unpacking. |
|
9
|
|
|
|
|
10
|
|
|
Test the lib with raw dump files from an OpenFlow switch. We assume the |
|
11
|
|
|
raw files are valid according to the OF specs to check whether our pack and |
|
12
|
|
|
unpack implementations are correct. |
|
13
|
|
|
|
|
14
|
|
|
Also, check the minimum size of the struct by instantiating an object with |
|
15
|
|
|
no parameters. |
|
16
|
|
|
|
|
17
|
|
|
To run these tests, just extends this class and call 2 methods in the |
|
18
|
|
|
``setUp`` method like the example. |
|
19
|
|
|
|
|
20
|
|
|
Example: |
|
21
|
|
|
.. code-block:: python3 |
|
22
|
|
|
|
|
23
|
|
|
class MyTest(TestDump): |
|
24
|
|
|
@classmethod |
|
25
|
|
|
def setUpClass(cls): |
|
26
|
|
|
super().setUpClass() |
|
27
|
|
|
super().set_raw_dump_file('v0x01', 'ofpt_barrier_reply') |
|
28
|
|
|
# Create BarrierReply(xid=5) when needed |
|
29
|
|
|
super().set_raw_dump_object(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(TestDump): |
|
37
|
|
|
@classmethod |
|
38
|
|
|
def setUpClass(cls): |
|
39
|
|
|
super().set_minimum_size(8, BarrierReply) |
|
40
|
|
|
""" |
|
41
|
|
|
|
|
42
|
|
|
IS_MSG = True |
|
43
|
|
|
|
|
44
|
|
|
def __init__(self, *args, **kwargs): |
|
45
|
|
|
"""The constructor will avoid that this class tests are executed. |
|
46
|
|
|
|
|
47
|
|
|
The tests in this class are executed through the child, so there's no |
|
48
|
|
|
no need for them to be executed once more through the parent. |
|
49
|
|
|
""" |
|
50
|
|
|
super().__init__(*args, **kwargs) |
|
51
|
|
|
# Override the run method, so it does nothing instead of running the |
|
52
|
|
|
# tests (again). |
|
53
|
|
|
if self.__class__ == TestStruct: |
|
54
|
|
|
self.run = lambda self, *args, **kwargs: None |
|
55
|
|
|
|
|
56
|
|
|
_new_raw_dump = None |
|
57
|
|
|
_new_raw_object = None |
|
58
|
|
|
_msg_cls = None |
|
59
|
|
|
_min_size = None |
|
60
|
|
|
|
|
61
|
|
|
# @classmethod |
|
62
|
|
|
def set_raw_dump_file(self, version, basename): |
|
63
|
|
|
"""Set which raw dump the tests will use. |
|
64
|
|
|
|
|
65
|
|
|
Args: |
|
66
|
|
|
protocol_version (str): OpenFlow protocol version, |
|
67
|
|
|
e.g. ``v0x01``. |
|
68
|
|
|
basename (str): The raw filename without extension. |
|
69
|
|
|
E.g. ``ofpt_echo_reply``. |
|
70
|
|
|
""" |
|
71
|
|
|
self._new_raw_dump = lambda: RawDump(version, basename) |
|
72
|
|
|
|
|
73
|
|
|
# @classmethod |
|
74
|
|
|
def get_raw_dump(self): |
|
75
|
|
|
"""Return a new instance of :class:`.RawDump`. |
|
76
|
|
|
|
|
77
|
|
|
Use the parameters set in :meth:`set_raw_dump_file`. |
|
78
|
|
|
|
|
79
|
|
|
Returns: |
|
80
|
|
|
RawDump: with parameters previously set using |
|
81
|
|
|
:meth:`set_raw_dump_file`. |
|
82
|
|
|
""" |
|
83
|
|
|
if self._new_raw_dump is None: |
|
84
|
|
|
raise FileNotFoundError() |
|
85
|
|
|
return self._new_raw_dump().read() |
|
86
|
|
|
|
|
87
|
|
|
# @classmethod |
|
88
|
|
|
def set_raw_dump_object(self, 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
|
|
|
self._msg_cls = msg_cls |
|
101
|
|
|
self._new_raw_object = lambda: msg_cls(*args, **kwargs) |
|
102
|
|
|
|
|
103
|
|
|
# @classmethod |
|
104
|
|
|
def get_raw_object(self): |
|
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
|
|
|
return self._new_raw_object() |
|
114
|
|
|
|
|
115
|
|
|
# @classmethod |
|
116
|
|
|
def set_minimum_size(self, size, msg_cls=None): |
|
117
|
|
|
"""Set the struct minimum size. |
|
118
|
|
|
|
|
119
|
|
|
The minimum size can be found in OF spec. For example, |
|
120
|
|
|
:class:`.PhyPort` minimum size is 48 because of |
|
121
|
|
|
``OFP_ASSERT(sizeof(struct ofp_phy_port) == 48);`` (spec 1.0.0). |
|
122
|
|
|
|
|
123
|
|
|
Args: |
|
124
|
|
|
size (int): The minimum size of the struct, in bytes. |
|
125
|
|
|
msg_cls (class): The class (or function) to have its size checked. |
|
126
|
|
|
If None, use the same class set in :meth:`set_raw_dump_object`. |
|
127
|
|
|
""" |
|
128
|
|
|
self._min_size = size |
|
129
|
|
|
if msg_cls is not None: |
|
130
|
|
|
TestStruct._msg_cls = msg_cls |
|
131
|
|
|
|
|
132
|
|
|
def unpack_dump(self): |
|
133
|
|
|
if self.IS_MSG: |
|
134
|
|
|
return self._msg_cls().unpack(self.get_raw_dump()[8:]) |
|
135
|
|
|
else: |
|
136
|
|
|
return self._msg_cls().unpack(self.get_raw_dump()) |
|
137
|
|
|
|
|
138
|
|
|
def test_pack(self): |
|
139
|
|
|
"""Check whether packed objects equals to dump file.""" |
|
140
|
|
|
try: |
|
141
|
|
|
raw_file = self.get_raw_dump() |
|
142
|
|
|
msg = self.get_raw_object() |
|
143
|
|
|
packed_obj = msg.pack() |
|
144
|
|
|
self.assertEqual(packed_obj, raw_file) |
|
145
|
|
|
except FileNotFoundError: |
|
146
|
|
|
raise self.skipTest('No raw dump file found.') |
|
147
|
|
|
|
|
148
|
|
|
def test_unpack(self): |
|
149
|
|
|
"""Check whether the unpacked dump equals to expected object.""" |
|
150
|
|
|
try: |
|
151
|
|
|
unpacked = self.unpack_dump() |
|
152
|
|
|
obj = self.get_raw_object() |
|
153
|
|
|
self.assertEqual(unpacked, obj) |
|
154
|
|
|
except FileNotFoundError: |
|
155
|
|
|
raise self.skipTest('No raw dump file found.') |
|
156
|
|
|
|
|
157
|
|
|
def test_minimum_size(self): |
|
158
|
|
|
"""Test struct minimum size.""" |
|
159
|
|
|
if self._min_size is None: |
|
160
|
|
|
raise self.skipTest('minimum size was not set.') |
|
161
|
|
|
obj = TestStruct._msg_cls() |
|
162
|
|
|
self.assertEqual(obj.get_size(), self._min_size) |
|
163
|
|
|
|
|
164
|
|
|
def test_obj_get_size(self): |
|
165
|
|
|
"""Check whether the unpacked dump has the expected size.""" |
|
166
|
|
|
try: |
|
167
|
|
|
unpacked = self.unpack_dump() |
|
168
|
|
|
obj = self.get_raw_object() |
|
169
|
|
|
self.assertEqual(obj.get_size(), unpacked.get_size()) |
|
170
|
|
|
except FileNotFoundError: |
|
171
|
|
|
raise self.skipTest('No raw dump file found.') |
|
172
|
|
|
|
|
173
|
|
|
def test_raw_dump_size(self): |
|
174
|
|
|
"""Check whether the unpacked dump has the expected size.""" |
|
175
|
|
|
try: |
|
176
|
|
|
dump = self.get_raw_dump() |
|
177
|
|
|
obj = self.get_raw_object() |
|
178
|
|
|
self.assertEqual(obj.get_size(), len(dump)) |
|
179
|
|
|
except FileNotFoundError: |
|
180
|
|
|
raise self.skipTest('No raw dump file found.') |
|
181
|
|
|
|
kytos /
python-openflow
