pyloxi: add (incomplete) of13 message test cases
diff --git a/py_gen/tests/of13.py b/py_gen/tests/of13.py
index 5689a94..55cd20f 100644
--- a/py_gen/tests/of13.py
+++ b/py_gen/tests/of13.py
@@ -26,6 +26,7 @@
# EPL for the specific language governing permissions and limitations
# under the EPL.
import unittest
+import difflib
try:
import loxi.of13 as ofp
@@ -33,6 +34,35 @@
except ImportError:
exit("loxi package not found. Try setting PYTHONPATH.")
+# Human-friendly format for binary strings. 8 bytes per line.
+def format_binary(buf):
+ byts = map(ord, buf)
+ lines = [[]]
+ for byt in byts:
+ if len(lines[-1]) == 8:
+ lines.append([])
+ lines[-1].append(byt)
+ return '\n'.join([' '.join(['%02x' % y for y in x]) for x in lines])
+
+def diff(a, b):
+ return '\n'.join(difflib.ndiff(a.splitlines(), b.splitlines()))
+
+# Test serialization / deserialization of a sample object.
+# Depends on the __eq__ method being correct.
+def test_serialization(obj, buf):
+ packed = obj.pack()
+ if packed != buf:
+ a = format_binary(buf)
+ b = format_binary(packed)
+ raise AssertionError("Serialization of %s failed\nExpected:\n%s\nActual:\n%s\nDiff:\n%s" % \
+ (type(obj).__name__, a, b, diff(a, b)))
+ unpacked = type(obj).unpack(buf)
+ if obj != unpacked:
+ a = obj.show()
+ b = unpacked.show()
+ raise AssertionError("Deserialization of %s failed\nExpected:\n%s\nActual:\n%s\nDiff:\n%s" % \
+ (type(obj).__name__, a, b, diff(a, b)))
+
class TestImports(unittest.TestCase):
def test_toplevel(self):
import loxi
@@ -86,6 +116,443 @@
self.assertTrue(isinstance(l[0], ofp.hello_elem_versionbitmap))
self.assertTrue(isinstance(l[1], ofp.hello_elem_versionbitmap))
+class TestMessages(unittest.TestCase):
+ def test_hello(self):
+ obj = ofp.message.hello(
+ xid=0x12345678,
+ elements=[
+ ofp.hello_elem_versionbitmap(
+ bitmaps=[ofp.uint32(1), ofp.uint32(2)]),
+ ofp.hello_elem_versionbitmap(
+ bitmaps=[ofp.uint32(3), ofp.uint32(4)])])
+ buf = ''.join([
+ '\x04', '\x00', # version, type
+ '\x00\x20', # length
+ '\x12\x34\x56\x78', # xid
+ '\x00\x01', # elements[0].type
+ '\x00\x0c', # elements[0].length
+ '\x00\x00\x00\x01', # elements[0].bitmaps[0]
+ '\x00\x00\x00\x02', # elements[0].bitmaps[1]
+ '\x00\x01', # elements[1].type
+ '\x00\x0c', # elements[1].length
+ '\x00\x00\x00\x03', # elements[1].bitmaps[0]
+ '\x00\x00\x00\x04', # elements[1].bitmaps[1]
+ ])
+ test_serialization(obj, buf)
+
+ def test_error(self):
+ obj = ofp.message.error_msg(
+ xid=0x12345678,
+ err_type=ofp.OFPET_BAD_MATCH,
+ code=ofp.OFPBMC_BAD_MASK,
+ data="abc")
+ buf = ''.join([
+ '\x04', '\x01', # version, type
+ '\x00\x0f', # length
+ '\x12\x34\x56\x78', # xid
+ '\x00\x04', # err_type
+ '\x00\x08', # code
+ 'abc', # data
+ ])
+ test_serialization(obj, buf)
+
+ def test_echo_request(self):
+ obj = ofp.message.echo_request(
+ xid=0x12345678,
+ data="abc")
+ buf = ''.join([
+ '\x04', '\x02', # version, type
+ '\x00\x0b', # length
+ '\x12\x34\x56\x78', # xid
+ 'abc', # data
+ ])
+ test_serialization(obj, buf)
+
+ def test_echo_reply(self):
+ obj = ofp.message.echo_reply(
+ xid=0x12345678,
+ data="abc")
+ buf = ''.join([
+ '\x04', '\x03', # version, type
+ '\x00\x0b', # length
+ '\x12\x34\x56\x78', # xid
+ 'abc', # data
+ ])
+ test_serialization(obj, buf)
+
+ def test_features_request(self):
+ obj = ofp.message.features_request(xid=0x12345678)
+ buf = ''.join([
+ '\x04', '\x05', # version, type
+ '\x00\x08', # length
+ '\x12\x34\x56\x78', # xid
+ ])
+ test_serialization(obj, buf)
+
+ def test_features_reply(self):
+ obj = ofp.message.features_reply(
+ xid=0x12345678,
+ datapath_id=0xFEDCBA9876543210,
+ n_buffers=64,
+ n_tables=200,
+ auxiliary_id=5,
+ capabilities=ofp.OFPC_FLOW_STATS|ofp.OFPC_PORT_BLOCKED,
+ reserved=0)
+ buf = ''.join([
+ '\x04', '\x06', # version, type
+ '\x00\x20', # length
+ '\x12\x34\x56\x78', # xid
+ '\xfe\xdc\xba\x98\x76\x54\x32\x10', # datapath_id
+ '\x00\x00\x00\x40', # n_buffers
+ '\xc8', # n_tables
+ '\x05', # auxiliary_id
+ '\x00\x00', # pad
+ '\x00\x00\x01\x01', # capabilities
+ '\x00\x00\x00\x00', # reserved
+ ])
+ test_serialization(obj, buf)
+
+ def test_get_config_request(self):
+ obj = ofp.message.get_config_request(xid=0x12345678)
+ buf = ''.join([
+ '\x04', '\x07', # version, type
+ '\x00\x08', # length
+ '\x12\x34\x56\x78', # xid
+ ])
+ test_serialization(obj, buf)
+
+ def test_get_config_reply(self):
+ obj = ofp.message.get_config_reply(
+ xid=0x12345678,
+ flags=ofp.OFPC_FRAG_REASM,
+ miss_send_len=0xffff)
+ buf = ''.join([
+ '\x04', '\x08', # version, type
+ '\x00\x0c', # length
+ '\x12\x34\x56\x78', # xid
+ '\x00\x02', # flags
+ '\xff\xff', # miss_send_len
+ ])
+ test_serialization(obj, buf)
+
+ def test_set_config(self):
+ obj = ofp.message.set_config(
+ xid=0x12345678,
+ flags=ofp.OFPC_FRAG_REASM,
+ miss_send_len=0xffff)
+ buf = ''.join([
+ '\x04', '\x09', # version, type
+ '\x00\x0c', # length
+ '\x12\x34\x56\x78', # xid
+ '\x00\x02', # flags
+ '\xff\xff', # miss_send_len
+ ])
+ test_serialization(obj, buf)
+
+ def test_packet_in(self):
+ obj = ofp.message.packet_in(
+ xid=0x12345678,
+ buffer_id=100,
+ total_len=17000,
+ reason=ofp.OFPR_ACTION,
+ table_id=20,
+ cookie=0xFEDCBA9876543210,
+ match=ofp.match(oxm_list=[
+ ofp.oxm.arp_op(value=1),
+ ofp.oxm.in_port_masked(value=4, value_mask=5)]),
+ data="abc")
+ buf = ''.join([
+ '\x04', '\x0a', # version, type
+ '\x00\x35', # length
+ '\x12\x34\x56\x78', # xid
+ '\x00\x00\x00\x64', # buffer_id
+ '\x42\x68', # total_len
+ '\x01', # reason
+ '\x14', # table_id
+ '\xfe\xdc\xba\x98\x76\x54\x32\x10', # cookie
+ '\x00\x01', # match.type
+ '\x00\x16', # match.length
+ '\x80\x00\x2A\x02', # match.oxm_list[0].type_len
+ '\x00\x01', # match.oxm_list[0].value
+ '\x80\x00\x01\x08', # match.oxm_list[1].type_len
+ '\x00\x00\x00\x04', # match.oxm_list[1].value
+ '\x00\x00\x00\x05', # match.oxm_list[1].mask
+ '\x00\x00', # match.pad
+ '\x00\x00', # pad
+ 'abc', # data
+ ])
+ test_serialization(obj, buf)
+
+ def test_flow_removed(self):
+ obj = ofp.message.flow_removed(
+ xid=0x12345678,
+ cookie=0xFEDCBA9876543210,
+ priority=17000,
+ reason=ofp.OFPRR_DELETE,
+ table_id=20,
+ duration_sec=10,
+ duration_nsec=1000,
+ idle_timeout=5,
+ hard_timeout=30,
+ packet_count=1,
+ byte_count=2,
+ match=ofp.match(oxm_list=[
+ ofp.oxm.arp_op(value=1),
+ ofp.oxm.in_port_masked(value=4, value_mask=5)]))
+ buf = ''.join([
+ '\x04', '\x0b', # version, type
+ '\x00\x48', # length
+ '\x12\x34\x56\x78', # xid
+ '\xfe\xdc\xba\x98\x76\x54\x32\x10', # cookie
+ '\x42\x68', # priority
+ '\x02', # reason
+ '\x14', # table_id
+ '\x00\x00\x00\x0a', # duration_sec
+ '\x00\x00\x03\xe8', # duration_nsec
+ '\x00\x05', # idle_timeout
+ '\x00\x1e', # hard_timeout
+ '\x00\x00\x00\x00\x00\x00\x00\x01', # packet_count
+ '\x00\x00\x00\x00\x00\x00\x00\x02', # byte_count
+ '\x00\x01', # match.type
+ '\x00\x16', # match.length
+ '\x80\x00\x2A\x02', # match.oxm_list[0].type_len
+ '\x00\x01', # match.oxm_list[0].value
+ '\x80\x00\x01\x08', # match.oxm_list[1].type_len
+ '\x00\x00\x00\x04', # match.oxm_list[1].value
+ '\x00\x00\x00\x05', # match.oxm_list[1].mask
+ '\x00\x00', # match.pad
+ ])
+ test_serialization(obj, buf)
+
+ def test_port_status(self):
+ obj = ofp.message.port_status(
+ xid=0x12345678,
+ reason=ofp.OFPPR_MODIFY,
+ desc=ofp.port_desc(
+ port_no=4,
+ hw_addr=[1,2,3,4,5,6],
+ name="foo",
+ config=ofp.OFPPC_NO_FWD|ofp.OFPPC_NO_RECV,
+ state=ofp.OFPPS_BLOCKED,
+ curr=ofp.OFPPF_10MB_HD,
+ advertised=ofp.OFPPF_10MB_FD,
+ supported=ofp.OFPPF_100MB_HD,
+ peer=ofp.OFPPF_100MB_FD,
+ curr_speed=10,
+ max_speed=20))
+ buf = ''.join([
+ '\x04', '\x0c', # version, type
+ '\x00\x50', # length
+ '\x12\x34\x56\x78', # xid
+ '\x02', # reason
+ '\x00' * 7, # pad
+ '\x00\x00\x00\x04', # port_no
+ '\x00' * 4, # pad
+ '\x01\x02\x03\x04\x05\x06', # hw_addr
+ '\x00' * 2, # pad
+ 'foo' + '\x00' * 13, # name
+ '\x00\x00\x00\x24', # config
+ '\x00\x00\x00\x02', # state
+ '\x00\x00\x00\x01', # curr
+ '\x00\x00\x00\x02', # advertised
+ '\x00\x00\x00\x04', # supported
+ '\x00\x00\x00\x08', # peer
+ '\x00\x00\x00\x0a', # curr_speed
+ '\x00\x00\x00\x14', # max_speed
+ ])
+ test_serialization(obj, buf)
+
+ def test_packet_out(self):
+ # TODO
+ pass
+
+
+ ## Flow-mods
+
+ def test_flow_add(self):
+ # TODO
+ pass
+
+ def test_flow_modify(self):
+ # TODO
+ pass
+
+ def test_flow_modify_strict(self):
+ # TODO
+ pass
+
+ def test_flow_delete(self):
+ # TODO
+ pass
+
+ def test_flow_delete_strict(self):
+ # TODO
+ pass
+
+
+ def test_group_mod(self):
+ # TODO
+ pass
+
+ def test_port_mod(self):
+ # TODO
+ pass
+
+ def test_table_mod(self):
+ # TODO
+ pass
+
+
+ ## Multipart messages
+
+ def test_desc_stats_request(self):
+ # TODO
+ pass
+
+ def test_desc_stats_reply(self):
+ # TODO
+ pass
+
+ def test_flow_stats_request(self):
+ # TODO
+ pass
+
+ def test_flow_stats_reply(self):
+ # TODO
+ pass
+
+ def test_aggregate_stats_request(self):
+ # TODO
+ pass
+
+ def test_aggregate_stats_reply(self):
+ # TODO
+ pass
+
+ def test_port_stats_request(self):
+ # TODO
+ pass
+
+ def test_port_stats_reply(self):
+ # TODO
+ pass
+
+ def test_queue_stats_request(self):
+ # TODO
+ pass
+
+ def test_queue_stats_reply(self):
+ # TODO
+ pass
+
+ def test_group_stats_request(self):
+ # TODO
+ pass
+
+ def test_group_stats_reply(self):
+ # TODO
+ pass
+
+ def test_group_desc_stats_request(self):
+ # TODO
+ pass
+
+ def test_group_desc_stats_reply(self):
+ # TODO
+ pass
+
+ def test_group_features_stats_request(self):
+ # TODO
+ pass
+
+ def test_group_features_stats_reply(self):
+ # TODO
+ pass
+
+ def test_meter_stats_request(self):
+ # TODO
+ pass
+
+ def test_meter_stats_reply(self):
+ # TODO
+ pass
+
+ def test_meter_config_stats_request(self):
+ # TODO
+ pass
+
+ def test_meter_config_stats_reply(self):
+ # TODO
+ pass
+
+ def test_meter_features_stats_request(self):
+ # TODO
+ pass
+
+ def test_meter_features_stats_reply(self):
+ # TODO
+ pass
+
+ def test_table_features_stats_request(self):
+ # TODO
+ pass
+
+ def test_table_features_stats_reply(self):
+ # TODO
+ pass
+
+ def test_port_desc_stats_request(self):
+ # TODO
+ pass
+
+ def test_port_desc_stats_reply(self):
+ # TODO
+ pass
+
+
+ def test_barrier_request(self):
+ # TODO
+ pass
+
+ def test_barrier_reply(self):
+ # TODO
+ pass
+
+ def test_queue_get_config_request(self):
+ # TODO
+ pass
+
+ def test_queue_get_config_reply(self):
+ # TODO
+ pass
+
+ def test_role_request(self):
+ # TODO
+ pass
+
+ def test_role_reply(self):
+ # TODO
+ pass
+
+ def test_get_async_request(self):
+ # TODO
+ pass
+
+ def test_get_async_reply(self):
+ # TODO
+ pass
+
+ def test_set_async(self):
+ # TODO
+ pass
+
+ def test_meter_mod(self):
+ # TODO
+ pass
+
+ # TODO test experimenter messages
+
+
class TestOXM(unittest.TestCase):
def test_oxm_in_phy_port_pack(self):
import loxi.of13 as ofp