py_gen/tests/of13.py

#!/usr/bin/env python
# Copyright 2013, Big Switch Networks, Inc.
#
# LoxiGen is licensed under the Eclipse Public License, version 1.0 (EPL), with
# the following special exception:
#
# LOXI Exception
#
# As a special exception to the terms of the EPL, you may distribute libraries
# generated by LoxiGen (LoxiGen Libraries) under the terms of your choice, provided
# that copyright and licensing notices generated by LoxiGen are not altered or removed
# from the LoxiGen Libraries and the notice provided below is (i) included in
# the LoxiGen Libraries, if distributed in source code form and (ii) included in any
# documentation for the LoxiGen Libraries, if distributed in binary form.
#
# Notice: "Copyright 2013, Big Switch Networks, Inc. This library was generated by the LoxiGen Compiler."
#
# You may not use this file except in compliance with the EPL or LOXI Exception. You may obtain
# a copy of the EPL at:
#
# http://www.eclipse.org/legal/epl-v10.html
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# EPL for the specific language governing permissions and limitations
# under the EPL.
import unittest
from testutil import test_serialization
from testutil import add_datafiles_tests

try:
    import loxi.of13 as ofp
    from loxi.generic_util import OFReader
except ImportError:
    exit("loxi package not found. Try setting PYTHONPATH.")

class TestImports(unittest.TestCase):
    def test_toplevel(self):
        import loxi
        self.assertTrue(hasattr(loxi, "ProtocolError"))
        self.assertEquals(loxi.version_names[4], "1.3")
        ofp = loxi.protocol(4)
        self.assertEquals(ofp.OFP_VERSION, 4)
        self.assertTrue(hasattr(ofp, "action"))
        self.assertTrue(hasattr(ofp, "common"))
        self.assertTrue(hasattr(ofp, "const"))
        self.assertTrue(hasattr(ofp, "message"))
        self.assertTrue(hasattr(ofp, "oxm"))

    def test_version(self):
        import loxi
        self.assertTrue(hasattr(loxi.of13, "ProtocolError"))
        self.assertTrue(hasattr(loxi.of13, "OFP_VERSION"))
        self.assertEquals(loxi.of13.OFP_VERSION, 4)
        self.assertTrue(hasattr(loxi.of13, "action"))
        self.assertTrue(hasattr(loxi.of13, "common"))
        self.assertTrue(hasattr(loxi.of13, "const"))
        self.assertTrue(hasattr(loxi.of13, "message"))
        self.assertTrue(hasattr(loxi.of13, "oxm"))

class TestCommon(unittest.TestCase):
    def test_list_hello_elem_unpack(self):
        buf = ''.join([
            '\x00\x01\x00\x04', # versionbitmap
            '\x00\x00\x00\x04', # unknown type
            '\x00\x01\x00\x04', # versionbitmap
        ])
        l = ofp.unpack_list_hello_elem(OFReader(buf))
        self.assertEquals(len(l), 2)
        self.assertTrue(isinstance(l[0], ofp.hello_elem_versionbitmap))
        self.assertTrue(isinstance(l[1], ofp.hello_elem_versionbitmap))

# The majority of the serialization tests are created here using the files in
# the test_data directory.
class TestDataFiles(unittest.TestCase):
    pass
add_datafiles_tests(TestDataFiles, 'of13/', ofp)

class TestAllOF13(unittest.TestCase):
    """
    Round-trips every class through serialization/deserialization.
    Not a replacement for handcoded tests because it only uses the
    default member values.
    """

    def setUp(self):
        mods = [ofp.action,ofp.message,ofp.common,ofp.oxm]
        self.klasses = [klass for mod in mods
                              for klass in mod.__dict__.values()
                              if hasattr(klass, 'show')]
        self.klasses.sort(key=lambda x: str(x))

    def test_serialization(self):
        expected_failures = [
            ofp.common.table_feature_prop_apply_actions,
            ofp.common.table_feature_prop_apply_actions_miss,
            ofp.common.table_feature_prop_write_actions,
            ofp.common.table_feature_prop_write_actions_miss,
            ofp.common.table_features,
            ofp.message.table_features_stats_reply,
            ofp.message.table_features_stats_request,
        ]
        for klass in self.klasses:
            def fn():
                obj = klass()
                if hasattr(obj, "xid"): obj.xid = 42
                buf = obj.pack()
                obj2 = klass.unpack(buf)
                self.assertEquals(obj, obj2)
            if klass in expected_failures:
                self.assertRaises(Exception, fn)
            else:
                fn()

    def test_parse_message(self):
        expected_failures = [
            ofp.message.aggregate_stats_reply,
            ofp.message.aggregate_stats_request,
            ofp.message.bsn_bw_clear_data_reply,
            ofp.message.bsn_bw_clear_data_request,
            ofp.message.bsn_bw_enable_get_reply,
            ofp.message.bsn_bw_enable_get_request,
            ofp.message.bsn_bw_enable_set_reply,
            ofp.message.bsn_bw_enable_set_request,
            ofp.message.bsn_get_interfaces_reply,
            ofp.message.bsn_get_interfaces_request,
            ofp.message.bsn_get_mirroring_reply,
            ofp.message.bsn_get_mirroring_request,
            ofp.message.bsn_set_mirroring,
            ofp.message.bsn_set_pktin_suppression_reply,
            ofp.message.bsn_set_pktin_suppression_request,
            ofp.message.bsn_virtual_port_create_reply,
            ofp.message.bsn_virtual_port_create_request,
            ofp.message.bsn_virtual_port_remove_reply,
            ofp.message.bsn_virtual_port_remove_request,
            ofp.message.desc_stats_reply,
            ofp.message.desc_stats_request,
            ofp.message.flow_add,
            ofp.message.flow_delete,
            ofp.message.flow_delete_strict,
            ofp.message.flow_modify,
            ofp.message.flow_modify_strict,
            ofp.message.flow_stats_reply,
            ofp.message.flow_stats_request,
            ofp.message.group_desc_stats_reply,
            ofp.message.group_desc_stats_request,
            ofp.message.group_features_stats_reply,
            ofp.message.group_features_stats_request,
            ofp.message.group_stats_reply,
            ofp.message.group_stats_request,
            ofp.message.meter_config_stats_reply,
            ofp.message.meter_config_stats_request,
            ofp.message.meter_features_stats_reply,
            ofp.message.meter_features_stats_request,
            ofp.message.meter_stats_reply,
            ofp.message.meter_stats_request,
            ofp.message.port_desc_stats_reply,
            ofp.message.port_desc_stats_request,
            ofp.message.port_stats_reply,
            ofp.message.port_stats_request,
            ofp.message.queue_stats_reply,
            ofp.message.queue_stats_request,
            ofp.message.table_features_stats_reply,
            ofp.message.table_features_stats_request,
            ofp.message.table_stats_reply,
            ofp.message.table_stats_request,
        ]
        for klass in self.klasses:
            if not issubclass(klass, ofp.message.Message):
                continue
            def fn():
                obj = klass(xid=42)
                buf = obj.pack()
                obj2 = ofp.message.parse_message(buf)
                self.assertEquals(obj, obj2)
            if klass in expected_failures:
                self.assertRaises(Exception, fn)
            else:
                fn()

    def test_show(self):
        expected_failures = []
        for klass in self.klasses:
            def fn():
                obj = klass()
                if hasattr(obj, "xid"): obj.xid = 42
                obj.show()
            if klass in expected_failures:
                self.assertRaises(Exception, fn)
            else:
                fn()

if __name__ == '__main__':
    unittest.main()