Gitiles
Code Review Sign In
gerrit.onosproject.org / onos-loxi / fd2fcd54ad0096a181b6ba7c94b3480c8fd75e05 / . / py_gen / tests.py
blob: 23d40b8643e11ad69ceac7ca3afd1493ecd26789 [file] [log] [blame]
Rich Lanea06d0c32013-03-25 08:52:03 -0700[diff] [blame]1#!/usr/bin/env python
2# Copyright 2013, Big Switch Networks, Inc.
3#
4# LoxiGen is licensed under the Eclipse Public License, version 1.0 (EPL), with
5# the following special exception:
6#
7# LOXI Exception
8#
9# As a special exception to the terms of the EPL, you may distribute libraries
10# generated by LoxiGen (LoxiGen Libraries) under the terms of your choice, provided
11# that copyright and licensing notices generated by LoxiGen are not altered or removed
12# from the LoxiGen Libraries and the notice provided below is (i) included in
13# the LoxiGen Libraries, if distributed in source code form and (ii) included in any
14# documentation for the LoxiGen Libraries, if distributed in binary form.
15#
16# Notice: "Copyright 2013, Big Switch Networks, Inc. This library was generated by the LoxiGen Compiler."
17#
18# You may not use this file except in compliance with the EPL or LOXI Exception. You may obtain
19# a copy of the EPL at:
20#
21# http://www.eclipse.org/legal/epl-v10.html
22#
23# Unless required by applicable law or agreed to in writing, software
24# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
25# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
26# EPL for the specific language governing permissions and limitations
27# under the EPL.
28import unittest
29
30try:
31 import loxi
32 del loxi
33except ImportError:
34 exit("loxi package not found. Try setting PYTHONPATH.")
35
36class TestImports(unittest.TestCase):
37 def test_toplevel(self):
38 import loxi
39 self.assertTrue(hasattr(loxi, "ProtocolError"))
40 ofp = loxi.protocol(1)
41 self.assertEquals(ofp.OFP_VERSION, 1)
42 self.assertTrue(hasattr(ofp, "action"))
43 self.assertTrue(hasattr(ofp, "common"))
44 self.assertTrue(hasattr(ofp, "const"))
45 self.assertTrue(hasattr(ofp, "message"))
46
47 def test_version(self):
48 import loxi.of10
49 self.assertTrue(hasattr(loxi.of10, "ProtocolError"))
50 self.assertTrue(hasattr(loxi.of10, "OFP_VERSION"))
51 self.assertEquals(loxi.of10.OFP_VERSION, 1)
52 self.assertTrue(hasattr(loxi.of10, "action"))
53 self.assertTrue(hasattr(loxi.of10, "common"))
54 self.assertTrue(hasattr(loxi.of10, "const"))
55 self.assertTrue(hasattr(loxi.of10, "message"))
56
57class TestActions(unittest.TestCase):
58 def test_output_pack(self):
59 import loxi.of10 as ofp
60 expected = "\x00\x00\x00\x08\xff\xf8\xff\xff"
61 action = ofp.action.output(port=ofp.OFPP_IN_PORT, max_len=0xffff)
62 self.assertEquals(expected, action.pack())
63
64 def test_output_unpack(self):
65 import loxi.of10 as ofp
66
67 # Normal case
68 buf = "\x00\x00\x00\x08\xff\xf8\xff\xff"
69 action = ofp.action.output.unpack(buf)
70 self.assertEqual(action.port, ofp.OFPP_IN_PORT)
71 self.assertEqual(action.max_len, 0xffff)
72
73 # Invalid length
74 buf = "\x00\x00\x00\x09\xff\xf8\xff\xff\x00"
75 with self.assertRaises(ofp.ProtocolError):
76 ofp.action.output.unpack(buf)
77
78 def test_output_equality(self):
79 import loxi.of10 as ofp
80 action = ofp.action.output(port=1, max_len=0x1234)
81 action2 = ofp.action.output(port=1, max_len=0x1234)
82 self.assertEquals(action, action2)
83
84 action2.port = 2
85 self.assertNotEquals(action, action2)
86 action2.port = 1
87
88 action2.max_len = 0xffff
89 self.assertNotEquals(action, action2)
90 action2.max_len = 0x1234
91
92 def test_output_show(self):
93 import loxi.of10 as ofp
94 action = ofp.action.output(port=1, max_len=0x1234)
95 expected = "output { port = 1, max_len = 0x1234 }"
96 self.assertEquals(expected, action.show())
97
98 def test_bsn_set_tunnel_dst_pack(self):
99 import loxi.of10 as ofp
100 expected = ''.join([
101 "\xff\xff", "\x00\x10", # type/length
102 "\x00\x5c\x16\xc7", # experimenter
103 "\x00\x00\x00\x02", # subtype
104 "\x12\x34\x56\x78" # dst
105 ])
106 action = ofp.action.bsn_set_tunnel_dst(dst=0x12345678)
107 self.assertEquals(expected, action.pack())
108
109 def test_bsn_set_tunnel_dst_unpack(self):
110 import loxi.of10 as ofp
111 buf = ''.join([
112 "\xff\xff", "\x00\x10", # type/length
113 "\x00\x5c\x16\xc7", # experimenter
114 "\x00\x00\x00\x02", # subtype
115 "\x12\x34\x56\x78" # dst
116 ])
117 action = ofp.action.bsn_set_tunnel_dst.unpack(buf)
118 self.assertEqual(action.subtype, 2)
119 self.assertEqual(action.dst, 0x12345678)
120
121# Assumes action serialization/deserialization works
122class TestActionList(unittest.TestCase):
123 def test_normal(self):
124 import loxi.of10 as ofp
125
126 expected = []
127 bufs = []
128
129 def add(action):
130 expected.append(action)
131 bufs.append(action.pack())
132
133 add(ofp.action.output(port=1, max_len=0xffff))
134 add(ofp.action.output(port=2, max_len=0xffff))
135 add(ofp.action.output(port=ofp.OFPP_IN_PORT, max_len=0xffff))
136 add(ofp.action.bsn_set_tunnel_dst(dst=0x12345678))
137 add(ofp.action.nicira_dec_ttl())
138
139 actions = ofp.action.unpack_list(''.join(bufs))
140 self.assertEquals(actions, expected)
141
142 def test_empty_list(self):
143 import loxi.of10 as ofp
144 self.assertEquals(ofp.action.unpack_list(''), [])
145
146 def test_invalid_list_length(self):
147 import loxi.of10 as ofp
148 buf = '\x00' * 9
149 with self.assertRaisesRegexp(ofp.ProtocolError, 'not a multiple of 8'):
150 ofp.action.unpack_list(buf)
151
152 def test_invalid_action_length(self):
153 import loxi.of10 as ofp
154
155 buf = '\x00' * 8
156 with self.assertRaisesRegexp(ofp.ProtocolError, 'is 0'):
157 ofp.action.unpack_list(buf)
158
159 buf = '\x00\x00\x00\x04'
160 with self.assertRaisesRegexp(ofp.ProtocolError, 'not a multiple of 8'):
161 ofp.action.unpack_list(buf)
162
163 buf = '\x00\x00\x00\x10\x00\x00\x00\x00'
164 with self.assertRaisesRegexp(ofp.ProtocolError, 'overrun'):
165 ofp.action.unpack_list(buf)
166
167 def test_invalid_action_type(self):
168 import loxi.of10 as ofp
169 buf = '\xff\xfe\x00\x08\x00\x00\x00\x00'
170 with self.assertRaisesRegexp(ofp.ProtocolError, 'unknown action type'):
171 ofp.action.unpack_list(buf)
172
173class TestConstants(unittest.TestCase):
174 def test_ports(self):
175 import loxi.of10 as ofp
176 self.assertEquals(0xffff, ofp.OFPP_NONE)
177
178 def test_wildcards(self):
179 import loxi.of10 as ofp
180 self.assertEquals(0xfc000, ofp.OFPFW_NW_DST_MASK)
181
182class TestCommon(unittest.TestCase):
183 def test_port_desc_pack(self):
184 import loxi.of10 as ofp
185 obj = ofp.port_desc(port_no=ofp.OFPP_CONTROLLER,
186 hw_addr=[1,2,3,4,5,6],
187 name="foo",
188 config=ofp.OFPPC_NO_FLOOD,
189 state=ofp.OFPPS_STP_FORWARD,
190 curr=ofp.OFPPF_10MB_HD,
191 advertised=ofp.OFPPF_1GB_FD,
192 supported=ofp.OFPPF_AUTONEG,
193 peer=ofp.OFPPF_PAUSE_ASYM)
194 expected = ''.join([
195 '\xff\xfd', # port_no
196 '\x01\x02\x03\x04\x05\x06', # hw_addr
197 'foo'.ljust(16, '\x00'), # name
198 '\x00\x00\x00\x10', # config
199 '\x00\x00\x02\x00', # state
200 '\x00\x00\x00\x01', # curr
201 '\x00\x00\x00\x20', # advertised
202 '\x00\x00\x02\x00', # supported
203 '\x00\x00\x08\x00', # peer
204 ])
205 self.assertEquals(expected, obj.pack())
206
207 def test_port_desc_unpack(self):
208 import loxi.of10 as ofp
209 buf = ''.join([
210 '\xff\xfd', # port_no
211 '\x01\x02\x03\x04\x05\x06', # hw_addr
212 'foo'.ljust(16, '\x00'), # name
213 '\x00\x00\x00\x10', # config
214 '\x00\x00\x02\x00', # state
215 '\x00\x00\x00\x01', # curr
216 '\x00\x00\x00\x20', # advertised
217 '\x00\x00\x02\x00', # supported
218 '\x00\x00\x08\x00', # peer
219 ])
220 obj = ofp.port_desc.unpack(buf)
221 self.assertEquals(ofp.OFPP_CONTROLLER, obj.port_no)
222 self.assertEquals('foo', obj.name)
223 self.assertEquals(ofp.OFPPF_PAUSE_ASYM, obj.peer)
224
225 def test_table_stats_entry_pack(self):
226 import loxi.of10 as ofp
227 obj = ofp.table_stats_entry(table_id=3,
228 name="foo",
229 wildcards=ofp.OFPFW_ALL,
230 max_entries=5,
231 active_count=2,
232 lookup_count=1099511627775,
233 matched_count=9300233470495232273L)
234 expected = ''.join([
235 '\x03', # table_id
236 '\x00\x00\x00', # pad
237 'foo'.ljust(32, '\x00'), # name
238 '\x00\x3f\xFF\xFF', # wildcards
239 '\x00\x00\x00\x05', # max_entries
240 '\x00\x00\x00\x02', # active_count
241 '\x00\x00\x00\xff\xff\xff\xff\xff', # lookup_count
242 '\x81\x11\x11\x11\x11\x11\x11\x11', # matched_count
243 ])
244 self.assertEquals(expected, obj.pack())
245
246 def test_table_stats_entry_unpack(self):
247 import loxi.of10 as ofp
248 buf = ''.join([
249 '\x03', # table_id
250 '\x00\x00\x00', # pad
251 'foo'.ljust(32, '\x00'), # name
252 '\x00\x3f\xFF\xFF', # wildcards
253 '\x00\x00\x00\x05', # max_entries
254 '\x00\x00\x00\x02', # active_count
255 '\x00\x00\x00\xff\xff\xff\xff\xff', # lookup_count
256 '\x81\x11\x11\x11\x11\x11\x11\x11', # matched_count
257 ])
258 obj = ofp.table_stats_entry.unpack(buf)
259 self.assertEquals(3, obj.table_id)
260 self.assertEquals('foo', obj.name)
261 self.assertEquals(9300233470495232273L, obj.matched_count)
262
263 def test_flow_stats_entry_pack(self):
264 import loxi.of10 as ofp
265 obj = ofp.flow_stats_entry(table_id=3,
266 match=ofp.match(),
267 duration_sec=1,
268 duration_nsec=2,
269 priority=100,
270 idle_timeout=5,
271 hard_timeout=10,
272 cookie=0x0123456789abcdef,
273 packet_count=10,
274 byte_count=1000,
275 actions=[ofp.action.output(port=1),
276 ofp.action.output(port=2)])
277 expected = ''.join([
278 '\x00\x68', # length
279 '\x03', # table_id
280 '\x00', # pad
281 '\x00\x3f\xff\xff', # match.wildcards
282 '\x00' * 36, # remaining match fields
283 '\x00\x00\x00\x01', # duration_sec
284 '\x00\x00\x00\x02', # duration_nsec
285 '\x00\x64', # priority
286 '\x00\x05', # idle_timeout
287 '\x00\x0a', # hard_timeout
288 '\x00' * 6, # pad2
289 '\x01\x23\x45\x67\x89\xab\xcd\xef', # cookie
290 '\x00\x00\x00\x00\x00\x00\x00\x0a', # packet_count
291 '\x00\x00\x00\x00\x00\x00\x03\xe8', # byte_count
292 '\x00\x00', # actions[0].type
293 '\x00\x08', # actions[0].len
294 '\x00\x01', # actions[0].port
295 '\x00\x00', # actions[0].max_len
296 '\x00\x00', # actions[1].type
297 '\x00\x08', # actions[1].len
298 '\x00\x02', # actions[1].port
299 '\x00\x00', # actions[1].max_len
300 ])
301 self.assertEquals(expected, obj.pack())
302
303 def test_flow_stats_entry_unpack(self):
304 import loxi.of10 as ofp
305 buf = ''.join([
306 '\x00\x68', # length
307 '\x03', # table_id
308 '\x00', # pad
309 '\x00\x3f\xff\xff', # match.wildcards
310 '\x00' * 36, # remaining match fields
311 '\x00\x00\x00\x01', # duration_sec
312 '\x00\x00\x00\x02', # duration_nsec
313 '\x00\x64', # priority
314 '\x00\x05', # idle_timeout
315 '\x00\x0a', # hard_timeout
316 '\x00' * 6, # pad2
317 '\x01\x23\x45\x67\x89\xab\xcd\xef', # cookie
318 '\x00\x00\x00\x00\x00\x00\x00\x0a', # packet_count
319 '\x00\x00\x00\x00\x00\x00\x03\xe8', # byte_count
320 '\x00\x00', # actions[0].type
321 '\x00\x08', # actions[0].len
322 '\x00\x01', # actions[0].port
323 '\x00\x00', # actions[0].max_len
324 '\x00\x00', # actions[1].type
325 '\x00\x08', # actions[1].len
326 '\x00\x02', # actions[1].port
327 '\x00\x00', # actions[1].max_len
328 ])
329 obj = ofp.flow_stats_entry.unpack(buf)
330 self.assertEquals(3, obj.table_id)
331 self.assertEquals(ofp.OFPFW_ALL, obj.match.wildcards)
332 self.assertEquals(2, len(obj.actions))
333 self.assertEquals(1, obj.actions[0].port)
334 self.assertEquals(2, obj.actions[1].port)
335
336 def test_match(self):
337 import loxi.of10 as ofp
338 match = ofp.match()
339 self.assertEquals(match.wildcards, ofp.OFPFW_ALL)
340 self.assertEquals(match.tcp_src, 0)
341 buf = match.pack()
342 match2 = ofp.match.unpack(buf)
343 self.assertEquals(match, match2)
344
345class TestMessages(unittest.TestCase):
346 def test_hello_construction(self):
347 import loxi.of10 as ofp
348
349 msg = ofp.message.hello()
350 self.assertEquals(msg.version, ofp.OFP_VERSION)
351 self.assertEquals(msg.type, ofp.OFPT_HELLO)
352 self.assertEquals(msg.xid, None)
353
354 msg = ofp.message.hello(xid=123)
355 self.assertEquals(msg.xid, 123)
356
357 # 0 is a valid xid distinct from None
358 msg = ofp.message.hello(xid=0)
359 self.assertEquals(msg.xid, 0)
360
361 def test_hello_unpack(self):
362 import loxi.of10 as ofp
363
364 # Normal case
365 buf = "\x01\x00\x00\x08\x12\x34\x56\x78"
366 msg = ofp.message.hello(xid=0x12345678)
367 self.assertEquals(buf, msg.pack())
368
369 # Invalid length
370 buf = "\x01\x00\x00\x09\x12\x34\x56\x78\x9a"
371 with self.assertRaisesRegexp(ofp.ProtocolError, "should be 8"):
372 ofp.message.hello.unpack(buf)
373
374 def test_echo_request_construction(self):
375 import loxi.of10 as ofp
376 msg = ofp.message.echo_request(data="abc")
377 self.assertEquals(msg.data, "abc")
378
379 def test_echo_request_pack(self):
380 import loxi.of10 as ofp
381
382 msg = ofp.message.echo_request(xid=0x12345678, data="abc")
383 buf = msg.pack()
384 self.assertEquals(buf, "\x01\x02\x00\x0b\x12\x34\x56\x78\x61\x62\x63")
385
386 msg2 = ofp.message.echo_request.unpack(buf)
387 self.assertEquals(msg, msg2)
388
389 def test_echo_request_unpack(self):
390 import loxi.of10 as ofp
391
392 # Normal case
393 buf = "\x01\x02\x00\x0b\x12\x34\x56\x78\x61\x62\x63"
394 msg = ofp.message.echo_request(xid=0x12345678, data="abc")
395 self.assertEquals(buf, msg.pack())
396
397 # Invalid length
398 buf = "\x01\x02\x00\x07\x12\x34\x56"
399 with self.assertRaisesRegexp(ofp.ProtocolError, "buffer too short"):
400 ofp.message.echo_request.unpack(buf)
401
402 def test_echo_request_equality(self):
403 import loxi.of10 as ofp
404
405 msg = ofp.message.echo_request(xid=0x12345678, data="abc")
406 msg2 = ofp.message.echo_request(xid=0x12345678, data="abc")
407 #msg2 = ofp.message.echo_request.unpack(msg.pack())
408 self.assertEquals(msg, msg2)
409
410 msg2.xid = 1
411 self.assertNotEquals(msg, msg2)
412 msg2.xid = msg.xid
413
414 msg2.data = "a"
415 self.assertNotEquals(msg, msg2)
416 msg2.data = msg.data
417
418 def test_echo_request_show(self):
419 import loxi.of10 as ofp
420 expected = "echo_request { xid = 0x12345678, data = 'ab\\x01' }"
421 msg = ofp.message.echo_request(xid=0x12345678, data="ab\x01")
422 self.assertEquals(msg.show(), expected)
423
424 def test_flow_add(self):
425 import loxi.of10 as ofp
426 match = ofp.match()
427 msg = ofp.message.flow_add(xid=1,
428 match=match,
429 cookie=1,
430 idle_timeout=5,
431 flags=ofp.OFPFF_CHECK_OVERLAP,
432 actions=[
433 ofp.action.output(port=1),
434 ofp.action.output(port=2),
435 ofp.action.output(port=ofp.OFPP_CONTROLLER,
436 max_len=1024)])
437 buf = msg.pack()
438 msg2 = ofp.message.flow_add.unpack(buf)
439 self.assertEquals(msg, msg2)
440
441 def test_port_mod_pack(self):
442 import loxi.of10 as ofp
443 msg = ofp.message.port_mod(xid=2,
444 port_no=ofp.OFPP_CONTROLLER,
445 hw_addr=[1,2,3,4,5,6],
446 config=0x90ABCDEF,
447 mask=0xFF11FF11,
448 advertise=0xCAFE6789)
449 expected = "\x01\x0f\x00\x20\x00\x00\x00\x02\xff\xfd\x01\x02\x03\x04\x05\x06\x90\xab\xcd\xef\xff\x11\xff\x11\xca\xfe\x67\x89\x00\x00\x00\x00"
450 self.assertEquals(expected, msg.pack())
451
452 def test_desc_stats_reply_pack(self):
453 import loxi.of10 as ofp
454 msg = ofp.message.desc_stats_reply(xid=3,
455 flags=ofp.OFPSF_REPLY_MORE,
456 mfr_desc="The Indigo-2 Community",
457 hw_desc="Unknown server",
458 sw_desc="Indigo-2 LRI pre-release",
459 serial_num="11235813213455",
460 dp_desc="Indigo-2 LRI forwarding module")
461 expected = ''.join([
462 '\x01', '\x11', # version/type
463 '\x04\x2c', # length
464 '\x00\x00\x00\x03', # xid
465 '\x00\x00', # stats_type
466 '\x00\x01', # flags
467 'The Indigo-2 Community'.ljust(256, '\x00'), # mfr_desc
468 'Unknown server'.ljust(256, '\x00'), # hw_desc
469 'Indigo-2 LRI pre-release'.ljust(256, '\x00'), # sw_desc
470 '11235813213455'.ljust(32, '\x00'), # serial_num
471 'Indigo-2 LRI forwarding module'.ljust(256, '\x00'), # dp_desc
472 ])
473 self.assertEquals(expected, msg.pack())
474
475 def test_desc_stats_reply_unpack(self):
476 import loxi.of10 as ofp
477 buf = ''.join([
478 '\x01', '\x11', # version/type
479 '\x04\x2c', # length
480 '\x00\x00\x00\x03', # xid
481 '\x00\x00', # stats_type
482 '\x00\x01', # flags
483 'The Indigo-2 Community'.ljust(256, '\x00'), # mfr_desc
484 'Unknown server'.ljust(256, '\x00'), # hw_desc
485 'Indigo-2 LRI pre-release'.ljust(256, '\x00'), # sw_desc
486 '11235813213455'.ljust(32, '\x00'), # serial_num
487 'Indigo-2 LRI forwarding module'.ljust(256, '\x00'), # dp_desc
488 ])
489 msg = ofp.message.desc_stats_reply.unpack(buf)
490 self.assertEquals('Indigo-2 LRI forwarding module', msg.dp_desc)
491 self.assertEquals('11235813213455', msg.serial_num)
492 self.assertEquals(ofp.OFPST_DESC, msg.stats_type)
493 self.assertEquals(ofp.OFPSF_REPLY_MORE, msg.flags)
494
495 def test_port_status_pack(self):
496 import loxi.of10 as ofp
497
498 desc = ofp.port_desc(port_no=ofp.OFPP_CONTROLLER,
499 hw_addr=[1,2,3,4,5,6],
500 name="foo",
501 config=ofp.OFPPC_NO_FLOOD,
502 state=ofp.OFPPS_STP_FORWARD,
503 curr=ofp.OFPPF_10MB_HD,
504 advertised=ofp.OFPPF_1GB_FD,
505 supported=ofp.OFPPF_AUTONEG,
506 peer=ofp.OFPPF_PAUSE_ASYM)
507
508 msg = ofp.message.port_status(xid=4,
509 reason=ofp.OFPPR_DELETE,
510 desc=desc)
511 expected = ''.join([
512 '\x01', '\x0c', # version/type
513 '\x00\x40', # length
514 '\x00\x00\x00\x04', # xid
515 '\x01', # reason
516 '\x00\x00\x00\x00\x00\x00\x00' # pad
517 '\xff\xfd', # desc.port_no
518 '\x01\x02\x03\x04\x05\x06', # desc.hw_addr
519 'foo'.ljust(16, '\x00'), # desc.name
520 '\x00\x00\x00\x10', # desc.config
521 '\x00\x00\x02\x00', # desc.state
522 '\x00\x00\x00\x01', # desc.curr
523 '\x00\x00\x00\x20', # desc.advertised
524 '\x00\x00\x02\x00', # desc.supported
525 '\x00\x00\x08\x00', # desc.peer
526 ])
527 self.assertEquals(expected, msg.pack())
528
529 def test_port_status_unpack(self):
530 import loxi.of10 as ofp
531 buf = ''.join([
532 '\x01', '\x0c', # version/type
533 '\x00\x40', # length
534 '\x00\x00\x00\x04', # xid
535 '\x01', # reason
536 '\x00\x00\x00\x00\x00\x00\x00' # pad
537 '\xff\xfd', # desc.port_no
538 '\x01\x02\x03\x04\x05\x06', # desc.hw_addr
539 'foo'.ljust(16, '\x00'), # desc.name
540 '\x00\x00\x00\x10', # desc.config
541 '\x00\x00\x02\x00', # desc.state
542 '\x00\x00\x00\x01', # desc.curr
543 '\x00\x00\x00\x20', # desc.advertised
544 '\x00\x00\x02\x00', # desc.supported
545 '\x00\x00\x08\x00', # desc.peer
546 ])
547 msg = ofp.message.port_status.unpack(buf)
548 self.assertEquals('foo', msg.desc.name)
549 self.assertEquals(ofp.OFPPF_PAUSE_ASYM, msg.desc.peer)
550
551 def test_port_stats_reply_pack(self):
552 import loxi.of10 as ofp
553 msg = ofp.message.port_stats_reply(xid=5, flags=0, entries=[
554 ofp.port_stats_entry(port_no=1, rx_packets=56, collisions=5),
555 ofp.port_stats_entry(port_no=ofp.OFPP_LOCAL, rx_packets=1, collisions=1)])
556 expected = ''.join([
557 '\x01', '\x11', # version/type
558 '\x00\xdc', # length
559 '\x00\x00\x00\x05', # xid
560 '\x00\x04', # stats_type
561 '\x00\x00', # flags
562 '\x00\x01', # entries[0].port_no
563 '\x00\x00\x00\x00\x00\x00' # entries[0].pad
564 '\x00\x00\x00\x00\x00\x00\x00\x38', # entries[0].rx_packets
565 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].tx_packets
566 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].rx_bytes
567 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].tx_bytes
568 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].rx_dropped
569 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].tx_dropped
570 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].rx_errors
571 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].tx_errors
572 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].rx_frame_err
573 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].rx_over_err
574 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].rx_crc_err
575 '\x00\x00\x00\x00\x00\x00\x00\x05', # entries[0].collisions
576 '\xff\xfe', # entries[1].port_no
577 '\x00\x00\x00\x00\x00\x00' # entries[1].pad
578 '\x00\x00\x00\x00\x00\x00\x00\x01', # entries[1].rx_packets
579 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].tx_packets
580 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].rx_bytes
581 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].tx_bytes
582 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].rx_dropped
583 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].tx_dropped
584 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].rx_errors
585 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].tx_errors
586 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].rx_frame_err
587 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].rx_over_err
588 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].rx_crc_err
589 '\x00\x00\x00\x00\x00\x00\x00\x01', # entries[1].collisions
590 ])
591 self.assertEquals(expected, msg.pack())
592
593 def test_port_stats_reply_unpack(self):
594 import loxi.of10 as ofp
595 buf = ''.join([
596 '\x01', '\x11', # version/type
597 '\x00\xdc', # length
598 '\x00\x00\x00\x05', # xid
599 '\x00\x04', # stats_type
600 '\x00\x00', # flags
601 '\x00\x01', # entries[0].port_no
602 '\x00\x00\x00\x00\x00\x00' # entries[0].pad
603 '\x00\x00\x00\x00\x00\x00\x00\x38', # entries[0].rx_packets
604 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].tx_packets
605 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].rx_bytes
606 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].tx_bytes
607 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].rx_dropped
608 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].tx_dropped
609 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].rx_errors
610 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].tx_errors
611 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].rx_frame_err
612 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].rx_over_err
613 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[0].rx_crc_err
614 '\x00\x00\x00\x00\x00\x00\x00\x05', # entries[0].collisions
615 '\xff\xfe', # entries[1].port_no
616 '\x00\x00\x00\x00\x00\x00' # entries[1].pad
617 '\x00\x00\x00\x00\x00\x00\x00\x01', # entries[1].rx_packets
618 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].tx_packets
619 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].rx_bytes
620 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].tx_bytes
621 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].rx_dropped
622 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].tx_dropped
623 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].rx_errors
624 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].tx_errors
625 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].rx_frame_err
626 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].rx_over_err
627 '\x00\x00\x00\x00\x00\x00\x00\x00', # entries[1].rx_crc_err
628 '\x00\x00\x00\x00\x00\x00\x00\x01', # entries[1].collisions
629 ])
630 msg = ofp.message.port_stats_reply.unpack(buf)
631 self.assertEquals(ofp.OFPST_PORT, msg.stats_type)
632 self.assertEquals(2, len(msg.entries))
633
634 sample_flow_stats_reply_buf = ''.join([
635 '\x01', '\x11', # version/type
636 '\x00\xe4', # length
637 '\x00\x00\x00\x06', # xid
638 '\x00\x01', # stats_type
639 '\x00\x00', # flags
640 '\x00\x68', # entries[0].length
641 '\x03', # entries[0].table_id
642 '\x00', # entries[0].pad
643 '\x00\x3f\xff\xff', # entries[0].match.wildcards
644 '\x00' * 36, # remaining match fields
645 '\x00\x00\x00\x01', # entries[0].duration_sec
646 '\x00\x00\x00\x02', # entries[0].duration_nsec
647 '\x00\x64', # entries[0].priority
648 '\x00\x05', # entries[0].idle_timeout
649 '\x00\x0a', # entries[0].hard_timeout
650 '\x00' * 6, # entries[0].pad2
651 '\x01\x23\x45\x67\x89\xab\xcd\xef', # entries[0].cookie
652 '\x00\x00\x00\x00\x00\x00\x00\x0a', # entries[0].packet_count
653 '\x00\x00\x00\x00\x00\x00\x03\xe8', # entries[0].byte_count
654 '\x00\x00', # entries[0].actions[0].type
655 '\x00\x08', # entries[0].actions[0].len
656 '\x00\x01', # entries[0].actions[0].port
657 '\x00\x00', # entries[0].actions[0].max_len
658 '\x00\x00', # entries[0].actions[1].type
659 '\x00\x08', # entries[0].actions[1].len
660 '\x00\x02', # entries[0].actions[1].port
661 '\x00\x00', # entries[0].actions[1].max_len
662 '\x00\x70', # entries[1].length
663 '\x04', # entries[1].table_id
664 '\x00', # entries[1].pad
665 '\x00\x3f\xff\xff', # entries[1].match.wildcards
666 '\x00' * 36, # remaining match fields
667 '\x00\x00\x00\x01', # entries[1].duration_sec
668 '\x00\x00\x00\x02', # entries[1].duration_nsec
669 '\x00\x64', # entries[1].priority
670 '\x00\x05', # entries[1].idle_timeout
671 '\x00\x0a', # entries[1].hard_timeout
672 '\x00' * 6, # entries[1].pad2
673 '\x01\x23\x45\x67\x89\xab\xcd\xef', # entries[1].cookie
674 '\x00\x00\x00\x00\x00\x00\x00\x0a', # entries[1].packet_count
675 '\x00\x00\x00\x00\x00\x00\x03\xe8', # entries[1].byte_count
676 '\x00\x00', # entries[1].actions[0].type
677 '\x00\x08', # entries[1].actions[0].len
678 '\x00\x01', # entries[1].actions[0].port
679 '\x00\x00', # entries[1].actions[0].max_len
680 '\x00\x00', # entries[1].actions[1].type
681 '\x00\x08', # entries[1].actions[1].len
682 '\x00\x02', # entries[1].actions[1].port
683 '\x00\x00', # entries[1].actions[1].max_len
684 '\x00\x00', # entries[1].actions[2].type
685 '\x00\x08', # entries[1].actions[2].len
686 '\x00\x03', # entries[1].actions[2].port
687 '\x00\x00', # entries[1].actions[2].max_len
688 ])
689
690 def test_flow_stats_reply_pack(self):
691 import loxi.of10 as ofp
692 msg = ofp.message.flow_stats_reply(xid=6, flags=0, entries=[
693 ofp.flow_stats_entry(table_id=3,
694 match=ofp.match(),
695 duration_sec=1,
696 duration_nsec=2,
697 priority=100,
698 idle_timeout=5,
699 hard_timeout=10,
700 cookie=0x0123456789abcdef,
701 packet_count=10,
702 byte_count=1000,
703 actions=[ofp.action.output(port=1),
704 ofp.action.output(port=2)]),
705 ofp.flow_stats_entry(table_id=4,
706 match=ofp.match(),
707 duration_sec=1,
708 duration_nsec=2,
709 priority=100,
710 idle_timeout=5,
711 hard_timeout=10,
712 cookie=0x0123456789abcdef,
713 packet_count=10,
714 byte_count=1000,
715 actions=[ofp.action.output(port=1),
716 ofp.action.output(port=2),
717 ofp.action.output(port=3)])])
718 self.assertEquals(self.sample_flow_stats_reply_buf, msg.pack())
719
720 def test_flow_stats_reply_unpack(self):
721 import loxi.of10 as ofp
722 msg = ofp.message.flow_stats_reply.unpack(self.sample_flow_stats_reply_buf)
723 self.assertEquals(ofp.OFPST_FLOW, msg.stats_type)
724 self.assertEquals(2, len(msg.entries))
725 self.assertEquals(2, len(msg.entries[0].actions))
726 self.assertEquals(3, len(msg.entries[1].actions))
727
728 def test_flow_add_show(self):
729 import loxi.of10 as ofp
730 expected = """\
731flow_add {
732 xid = None,
733 match = match_v1 {
734 wildcards = OFPFW_DL_SRC|OFPFW_DL_DST,
735 in_port = 3,
736 eth_src = 01:23:45:67:89:ab,
737 eth_dst = cd:ef:01:23:45:67,
738 vlan_vid = 0x0,
739 vlan_pcp = 0x0,
740 pad1 = 0x0,
741 eth_type = 0x0,
742 ip_dscp = 0x0,
743 ip_proto = 0x0,
744 pad2 = [ 0, 0 ],
745 ipv4_src = 192.168.3.127,
746 ipv4_dst = 255.255.255.255,
747 tcp_src = 0x0,
748 tcp_dst = 0x0
749 },
750 cookie = 0x0,
751 idle_timeout = 0x0,
752 hard_timeout = 0x0,
753 priority = 0x0,
754 buffer_id = 0x0,
755 out_port = 0,
756 flags = 0x0,
757 actions = [
758 output { port = OFPP_FLOOD, max_len = 0x0 },
759 nicira_dec_ttl { pad = 0x0, pad2 = 0x0 },
760 bsn_set_tunnel_dst { dst = 0x0 }
761 ]
762}"""
763 msg = ofp.message.flow_add(
764 match=ofp.match(
765 wildcards=ofp.OFPFW_DL_SRC|ofp.OFPFW_DL_DST,
766 in_port=3,
767 ipv4_src=0xc0a8037f,
768 ipv4_dst=0xffffffff,
769 eth_src=[0x01, 0x23, 0x45, 0x67, 0x89, 0xab],
770 eth_dst=[0xcd, 0xef, 0x01, 0x23, 0x45, 0x67]),
771 actions=[
772 ofp.action.output(port=ofp.OFPP_FLOOD),
773 ofp.action.nicira_dec_ttl(),
774 ofp.action.bsn_set_tunnel_dst()])
775 self.assertEquals(msg.show(), expected)
776
777 sample_packet_out_buf = ''.join([
778 '\x01', '\x0d', # version/type
779 '\x00\x23', # length
780 '\x12\x34\x56\x78', # xid
781 '\xab\xcd\xef\x01', # buffer_id
782 '\xff\xfe', # in_port
783 '\x00\x10', # actions_len
784 '\x00\x00', # actions[0].type
785 '\x00\x08', # actions[0].len
786 '\x00\x01', # actions[0].port
787 '\x00\x00', # actions[0].max_len
788 '\x00\x00', # actions[1].type
789 '\x00\x08', # actions[1].len
790 '\x00\x02', # actions[1].port
791 '\x00\x00', # actions[1].max_len
792 'abc' # data
793 ])
794
795 def test_packet_out_pack(self):
796 import loxi.of10 as ofp
797 msg = ofp.message.packet_out(
798 xid=0x12345678,
799 buffer_id=0xabcdef01,
800 in_port=ofp.OFPP_LOCAL,
801 actions=[
802 ofp.action.output(port=1),
803 ofp.action.output(port=2)],
804 data='abc')
805 self.assertEquals(self.sample_packet_out_buf, msg.pack())
806
807 def test_packet_out_unpack(self):
808 import loxi.of10 as ofp
809 msg = ofp.message.packet_out.unpack(self.sample_packet_out_buf)
810 self.assertEquals(0x12345678, msg.xid)
811 self.assertEquals(0xabcdef01, msg.buffer_id)
812 self.assertEquals(ofp.OFPP_LOCAL, msg.in_port)
813 self.assertEquals(2, len(msg.actions))
814 self.assertEquals(1, msg.actions[0].port)
815 self.assertEquals(2, msg.actions[1].port)
816 self.assertEquals('abc', msg.data)
817
818 sample_packet_in_buf = ''.join([
819 '\x01', '\x0a', # version/type
820 '\x00\x15', # length
821 '\x12\x34\x56\x78', # xid
822 '\xab\xcd\xef\x01', # buffer_id
823 '\x00\x09', # total_len
824 '\xff\xfe', # in_port
825 '\x01', # reason
826 '\x00', # pad
827 'abc', # data
828 ])
829
830 def test_packet_in_pack(self):
831 import loxi.of10 as ofp
832 msg = ofp.message.packet_in(
833 xid=0x12345678,
834 buffer_id=0xabcdef01,
835 total_len=9,
836 in_port=ofp.OFPP_LOCAL,
837 reason=ofp.OFPR_ACTION,
838 data='abc')
839 self.assertEquals(self.sample_packet_in_buf, msg.pack())
840
841 def test_packet_in_unpack(self):
842 import loxi.of10 as ofp
843 msg = ofp.message.packet_in.unpack(self.sample_packet_in_buf)
844 self.assertEquals(0x12345678, msg.xid)
845 self.assertEquals(0xabcdef01, msg.buffer_id)
846 self.assertEquals(9, msg.total_len)
847 self.assertEquals(ofp.OFPP_LOCAL, msg.in_port)
848 self.assertEquals(ofp.OFPR_ACTION, msg.reason)
849 self.assertEquals('abc', msg.data)
850
851 sample_queue_get_config_reply_buf = ''.join([
852 '\x01', '\x15', # version/type
853 '\x00\x50', # length
854 '\x12\x34\x56\x78', # xid
855 '\xff\xfe', # port
856 '\x00\x00\x00\x00\x00\x00', # pad
857 '\x00\x00\x00\x01', # queues[0].queue_id
858 '\x00\x18', # queues[0].len
859 '\x00\x00', # queues[0].pad
860 '\x00\x01', # queues[0].properties[0].type
861 '\x00\x10', # queues[0].properties[0].length
862 '\x00\x00\x00\x00', # queues[0].properties[0].pad
863 '\x00\x05', # queues[0].properties[0].rate
864 '\x00\x00\x00\x00\x00\x00', # queues[0].properties[0].pad2
865 '\x00\x00\x00\x02', # queues[1].queue_id
866 '\x00\x28', # queues[1].len
867 '\x00\x00', # queues[1].pad
868 '\x00\x01', # queues[1].properties[0].type
869 '\x00\x10', # queues[1].properties[0].length
870 '\x00\x00\x00\x00', # queues[1].properties[0].pad
871 '\x00\x06', # queues[1].properties[0].rate
872 '\x00\x00\x00\x00\x00\x00', # queues[1].properties[0].pad2
873 '\x00\x01', # queues[1].properties[1].type
874 '\x00\x10', # queues[1].properties[1].length
875 '\x00\x00\x00\x00', # queues[1].properties[1].pad
876 '\x00\x07', # queues[1].properties[1].rate
877 '\x00\x00\x00\x00\x00\x00', # queues[1].properties[1].pad2
878 ])
879
880 def test_queue_get_config_reply_pack(self):
881 import loxi.of10 as ofp
882 msg = ofp.message.queue_get_config_reply(
883 xid=0x12345678,
884 port=ofp.OFPP_LOCAL,
885 queues=[
886 ofp.packet_queue(queue_id=1, properties=[
887 ofp.queue_prop_min_rate(rate=5)]),
888 ofp.packet_queue(queue_id=2, properties=[
889 ofp.queue_prop_min_rate(rate=6),
890 ofp.queue_prop_min_rate(rate=7)])])
891 self.assertEquals(self.sample_queue_get_config_reply_buf, msg.pack())
892
893 def test_queue_get_config_reply_unpack(self):
894 import loxi.of10 as ofp
895 msg = ofp.message.queue_get_config_reply.unpack(self.sample_queue_get_config_reply_buf)
896 self.assertEquals(ofp.OFPP_LOCAL, msg.port)
897 self.assertEquals(msg.queues[0].queue_id, 1)
898 self.assertEquals(msg.queues[0].properties[0].rate, 5)
899 self.assertEquals(msg.queues[1].queue_id, 2)
900 self.assertEquals(msg.queues[1].properties[0].rate, 6)
901 self.assertEquals(msg.queues[1].properties[1].rate, 7)
902
903class TestParse(unittest.TestCase):
904 def test_parse_header(self):
905 import loxi
906 import loxi.of10 as ofp
907
908 msg_ver, msg_type, msg_len, msg_xid = ofp.message.parse_header("\x01\x04\xAF\xE8\x12\x34\x56\x78")
909 self.assertEquals(1, msg_ver)
910 self.assertEquals(4, msg_type)
911 self.assertEquals(45032, msg_len)
912 self.assertEquals(0x12345678, msg_xid)
913
914 with self.assertRaisesRegexp(loxi.ProtocolError, "too short"):
915 ofp.message.parse_header("\x01\x04\xAF\xE8\x12\x34\x56")
916
917 def test_parse_message(self):
918 import loxi
919 import loxi.of10 as ofp
920
921 buf = "\x01\x00\x00\x08\x12\x34\x56\x78"
922 msg = ofp.message.parse_message(buf)
923 assert(msg.xid == 0x12345678)
924
925 # Get a list of all message classes
926 test_klasses = [x for x in ofp.message.__dict__.values()
927 if type(x) == type
928 and issubclass(x, ofp.message.Message)
929 and x != ofp.message.Message]
930
931 for klass in test_klasses:
932 self.assertIsInstance(ofp.message.parse_message(klass(xid=1).pack()), klass)
933
934class TestUtils(unittest.TestCase):
935 def test_unpack_array(self):
936 import loxi
937 import loxi.of10.util as util
938
939 a = util.unpack_array(str, 3, "abcdefghi")
940 self.assertEquals(['abc', 'def', 'ghi'], a)
941
942 with self.assertRaisesRegexp(loxi.ProtocolError, "invalid array length"):
943 util.unpack_array(str, 3, "abcdefgh")
944
945 def test_pretty_wildcards(self):
946 import loxi.of10 as ofp
947 self.assertEquals("OFPFW_ALL", ofp.util.pretty_wildcards(ofp.OFPFW_ALL))
948 self.assertEquals("0", ofp.util.pretty_wildcards(0))
949 self.assertEquals("OFPFW_DL_SRC|OFPFW_DL_DST",
950 ofp.util.pretty_wildcards(ofp.OFPFW_DL_SRC|ofp.OFPFW_DL_DST))
951 self.assertEquals("OFPFW_NW_SRC_MASK&0x2000",
952 ofp.util.pretty_wildcards(ofp.OFPFW_NW_SRC_ALL))
953 self.assertEquals("OFPFW_NW_SRC_MASK&0x1a00",
954 ofp.util.pretty_wildcards(0x00001a00))
955 self.assertEquals("OFPFW_IN_PORT|0x80000000",
956 ofp.util.pretty_wildcards(ofp.OFPFW_IN_PORT|0x80000000))
957
958class TestAll(unittest.TestCase):
959 """
960 Round-trips every class through serialization/deserialization.
961 Not a replacement for handcoded tests because it only uses the
962 default member values.
963 """
964
965 def setUp(self):
966 import loxi.of10 as ofp
967 mods = [ofp.action,ofp.message,ofp.common]
968 self.klasses = [klass for mod in mods
969 for klass in mod.__dict__.values()
970 if hasattr(klass, 'show')]
971 self.klasses.sort(key=lambda x: str(x))
972
973 def test_serialization(self):
974 import loxi.of10 as ofp
975 expected_failures = []
976 for klass in self.klasses:
977 def fn():
978 obj = klass()
979 if hasattr(obj, "xid"): obj.xid = 42
980 buf = obj.pack()
981 obj2 = klass.unpack(buf)
982 self.assertEquals(obj, obj2)
983 if klass in expected_failures:
984 self.assertRaises(Exception, fn)
985 else:
986 fn()
987
988 def test_show(self):
989 import loxi.of10 as ofp
990 expected_failures = []
991 for klass in self.klasses:
992 def fn():
993 obj = klass()
994 if hasattr(obj, "xid"): obj.xid = 42
995 obj.show()
996 if klass in expected_failures:
997 self.assertRaises(Exception, fn)
998 else:
999 fn()
1000
1001if __name__ == '__main__':
1002 unittest.main()