Changeset 322 for pyyaml/trunk/tests/test_yaml_ext.py

Timestamp:

12/28/08 15:16:50 (4 years ago)

Author:

xi

Message:

Refactored the test suite; updated include and library paths in setup.cfg.

File:

• pyyaml/trunk/tests/test_yaml_ext.py (modified) (1 diff)

pyyaml/trunk/tests/test_yaml_ext.py

@@ -1 @@
-
-import unittest, test_appliance
 
 import _yaml, yaml
-
-test_appliance.TestAppliance.SKIP_EXT = '.skip-ext'
-
-class TestCVersion(unittest.TestCase):
-
-    def testCVersion(self):
-        self.failUnlessEqual("%s.%s.%s" % _yaml.get_version(), _yaml.get_version_string())
-
-class TestCLoader(test_appliance.TestAppliance):
-
-    def _testCScannerFileInput(self, test_name, data_filename, canonical_filename):
-        self._testCScanner(test_name, data_filename, canonical_filename, True)
-
-    def _testCScanner(self, test_name, data_filename, canonical_filename, file_input=False, Loader=yaml.Loader):
-        if file_input:
-            data = file(data_filename, 'r')
-        else:
-            data = file(data_filename, 'r').read()
-        tokens = list(yaml.scan(data, Loader=Loader))
-        ext_tokens = []
+import types, pprint
+
+yaml.PyBaseLoader = yaml.BaseLoader
+yaml.PySafeLoader = yaml.SafeLoader
+yaml.PyLoader = yaml.Loader
+yaml.PyBaseDumper = yaml.BaseDumper
+yaml.PySafeDumper = yaml.SafeDumper
+yaml.PyDumper = yaml.Dumper
+
+old_scan = yaml.scan
+def new_scan(stream, Loader=yaml.CLoader):
+    return old_scan(stream, Loader)
+
+old_parse = yaml.parse
+def new_parse(stream, Loader=yaml.CLoader):
+    return old_parse(stream, Loader)
+
+old_compose = yaml.compose
+def new_compose(stream, Loader=yaml.CLoader):
+    return old_compose(stream, Loader)
+
+old_compose_all = yaml.compose_all
+def new_compose_all(stream, Loader=yaml.CLoader):
+    return old_compose_all(stream, Loader)
+
+old_load = yaml.load
+def new_load(stream, Loader=yaml.CLoader):
+    return old_load(stream, Loader)
+
+old_load_all = yaml.load_all
+def new_load_all(stream, Loader=yaml.CLoader):
+    return old_load_all(stream, Loader)
+
+old_safe_load = yaml.safe_load
+def new_safe_load(stream):
+    return old_load(stream, yaml.CSafeLoader)
+
+old_safe_load_all = yaml.safe_load_all
+def new_safe_load_all(stream):
+    return old_load_all(stream, yaml.CSafeLoader)
+
+old_emit = yaml.emit
+def new_emit(events, stream=None, Dumper=yaml.CDumper, **kwds):
+    return old_emit(events, stream, Dumper, **kwds)
+
+old_serialize = yaml.serialize
+def new_serialize(node, stream, Dumper=yaml.CDumper, **kwds):
+    return old_serialize(node, stream, Dumper, **kwds)
+
+old_serialize_all = yaml.serialize_all
+def new_serialize_all(nodes, stream=None, Dumper=yaml.CDumper, **kwds):
+    return old_serialize_all(nodes, stream, Dumper, **kwds)
+
+old_dump = yaml.dump
+def new_dump(data, stream=None, Dumper=yaml.CDumper, **kwds):
+    return old_dump(data, stream, Dumper, **kwds)
+
+old_dump_all = yaml.dump_all
+def new_dump_all(documents, stream=None, Dumper=yaml.CDumper, **kwds):
+    return old_dump_all(documents, stream, Dumper, **kwds)
+
+old_safe_dump = yaml.safe_dump
+def new_safe_dump(data, stream=None, **kwds):
+    return old_dump(data, stream, yaml.CSafeDumper, **kwds)
+
+old_safe_dump_all = yaml.safe_dump_all
+def new_safe_dump_all(documents, stream=None, **kwds):
+    return old_dump_all(documents, stream, yaml.CSafeDumper, **kwds)
+
+def _set_up():
+    yaml.BaseLoader = yaml.CBaseLoader
+    yaml.SafeLoader = yaml.CSafeLoader
+    yaml.Loader = yaml.CLoader
+    yaml.BaseDumper = yaml.CBaseDumper
+    yaml.SafeDumper = yaml.CSafeDumper
+    yaml.Dumper = yaml.CDumper
+    yaml.scan = new_scan
+    yaml.parse = new_parse
+    yaml.compose = new_compose
+    yaml.compose_all = new_compose_all
+    yaml.load = new_load
+    yaml.load_all = new_load_all
+    yaml.safe_load = new_safe_load
+    yaml.safe_load_all = new_safe_load_all
+    yaml.emit = new_emit
+    yaml.serialize = new_serialize
+    yaml.serialize_all = new_serialize_all
+    yaml.dump = new_dump
+    yaml.dump_all = new_dump_all
+    yaml.safe_dump = new_safe_dump
+    yaml.safe_dump_all = new_safe_dump_all
+
+def _tear_down():
+    yaml.BaseLoader = yaml.PyBaseLoader
+    yaml.SafeLoader = yaml.PySafeLoader
+    yaml.Loader = yaml.PyLoader
+    yaml.BaseDumper = yaml.PyBaseDumper
+    yaml.SafeDumper = yaml.PySafeDumper
+    yaml.Dumper = yaml.PyDumper
+    yaml.scan = old_scan
+    yaml.parse = old_parse
+    yaml.compose = old_compose
+    yaml.compose_all = old_compose_all
+    yaml.load = old_load
+    yaml.load_all = old_load_all
+    yaml.safe_load = old_safe_load
+    yaml.safe_load_all = old_safe_load_all
+    yaml.emit = old_emit
+    yaml.serialize = old_serialize
+    yaml.serialize_all = old_serialize_all
+    yaml.dump = old_dump
+    yaml.dump_all = old_dump_all
+    yaml.safe_dump = old_safe_dump
+    yaml.safe_dump_all = old_safe_dump_all
+
+def test_c_version(verbose=False):
+    if verbose:
+        print _yaml.get_version()
+        print _yaml.get_version_string()
+    assert ("%s.%s.%s" % _yaml.get_version()) == _yaml.get_version_string(),    \
+            (_yaml.get_version(), _yaml.get_version_string())
+
+def _compare_scanners(py_data, c_data, verbose):
+    py_tokens = list(yaml.scan(py_data, Loader=yaml.PyLoader))
+    c_tokens = []
+    try:
+        for token in yaml.scan(c_data, Loader=yaml.CLoader):
+            c_tokens.append(token)
+        assert len(py_tokens) == len(c_tokens), (len(py_tokens), len(c_tokens))
+        for py_token, c_token in zip(py_tokens, c_tokens):
+            assert py_token.__class__ == c_token.__class__, (py_token, c_token)
+            if hasattr(py_token, 'value'):
+                assert py_token.value == c_token.value, (py_token, c_token)
+            if isinstance(py_token, yaml.StreamEndToken):
+                continue
+            py_start = (py_token.start_mark.index, py_token.start_mark.line, py_token.start_mark.column)
+            py_end = (py_token.end_mark.index, py_token.end_mark.line, py_token.end_mark.column)
+            c_start = (c_token.start_mark.index, c_token.start_mark.line, c_token.start_mark.column)
+            c_end = (c_token.end_mark.index, c_token.end_mark.line, c_token.end_mark.column)
+            assert py_start == c_start, (py_start, c_start)
+            assert py_end == c_end, (py_end, c_end)
+    finally:
+        if verbose:
+            print "PY_TOKENS:"
+            pprint.pprint(py_tokens)
+            print "C_TOKENS:"
+            pprint.pprint(c_tokens)
+
+def test_c_scanner(data_filename, canonical_filename, verbose=False):
+    _compare_scanners(open(data_filename, 'rb'),
+            open(data_filename, 'rb'), verbose)
+    _compare_scanners(open(data_filename, 'rb').read(),
+            open(data_filename, 'rb').read(), verbose)
+    _compare_scanners(open(canonical_filename, 'rb'),
+            open(canonical_filename, 'rb'), verbose)
+    _compare_scanners(open(canonical_filename, 'rb').read(),
+            open(canonical_filename, 'rb').read(), verbose)
+
+test_c_scanner.unittest = ['.data', '.canonical']
+test_c_scanner.skip = ['.skip-ext']
+
+def _compare_parsers(py_data, c_data, verbose):
+    py_events = list(yaml.parse(py_data, Loader=yaml.PyLoader))
+    c_events = []
+    try:
+        for event in yaml.parse(c_data, Loader=yaml.CLoader):
+            c_events.append(event)
+        assert len(py_events) == len(c_events), (len(py_events), len(c_events))
+        for py_event, c_event in zip(py_events, c_events):
+            for attribute in ['__class__', 'anchor', 'tag', 'implicit',
+                                'value', 'explicit', 'version', 'tags']:
+                py_value = getattr(py_event, attribute, None)
+                c_value = getattr(c_event, attribute, None)
+                assert py_value == c_value, (py_event, c_event, attribute)
+    finally:
+        if verbose:
+            print "PY_EVENTS:"
+            pprint.pprint(py_events)
+            print "C_EVENTS:"
+            pprint.pprint(c_events)
+
+def test_c_parser(data_filename, canonical_filename, verbose=False):
+    _compare_parsers(open(data_filename, 'rb'),
+            open(data_filename, 'rb'), verbose)
+    _compare_parsers(open(data_filename, 'rb').read(),
+            open(data_filename, 'rb').read(), verbose)
+    _compare_parsers(open(canonical_filename, 'rb'),
+            open(canonical_filename, 'rb'), verbose)
+    _compare_parsers(open(canonical_filename, 'rb').read(),
+            open(canonical_filename, 'rb').read(), verbose)
+
+test_c_parser.unittest = ['.data', '.canonical']
+test_c_parser.skip = ['.skip-ext']
+
+def _compare_emitters(data, verbose):
+    events = list(yaml.parse(data, Loader=yaml.PyLoader))
+    c_data = yaml.emit(events, Dumper=yaml.CDumper)
+    if verbose:
+        print c_data
+    py_events = list(yaml.parse(c_data, Loader=yaml.PyLoader))
+    c_events = list(yaml.parse(c_data, Loader=yaml.CLoader))
+    try:
+        assert len(events) == len(py_events), (len(events), len(py_events))
+        assert len(events) == len(c_events), (len(events), len(c_events))
+        for event, py_event, c_event in zip(events, py_events, c_events):
+            for attribute in ['__class__', 'anchor', 'tag', 'implicit',
+                                'value', 'explicit', 'version', 'tags']:
+                value = getattr(event, attribute, None)
+                py_value = getattr(py_event, attribute, None)
+                c_value = getattr(c_event, attribute, None)
+                if attribute == 'tag' and value in [None, u'!'] \
+                        and py_value in [None, u'!'] and c_value in [None, u'!']:
+                    continue
+                if attribute == 'explicit' and (py_value or c_value):
+                    continue
+                assert value == py_value, (event, py_event, attribute)
+                assert value == c_value, (event, c_event, attribute)
+    finally:
+        if verbose:
+            print "EVENTS:"
+            pprint.pprint(events)
+            print "PY_EVENTS:"
+            pprint.pprint(py_events)
+            print "C_EVENTS:"
+            pprint.pprint(c_events)
+
+def test_c_emitter(data_filename, canonical_filename, verbose=False):
+    _compare_emitters(open(data_filename, 'rb').read(), verbose)
+    _compare_emitters(open(canonical_filename, 'rb').read(), verbose)
+
+test_c_emitter.unittest = ['.data', '.canonical']
+test_c_emitter.skip = ['.skip-ext']
+
+def wrap_ext_function(function):
+    def wrapper(*args, **kwds):
+        _set_up()
         try:
-            if file_input:
-                data = file(data_filename, 'r')
-            for token in yaml.scan(data, Loader=yaml.CLoader):
-                ext_tokens.append(token)
-            self.failUnlessEqual(len(tokens), len(ext_tokens))
-            for token, ext_token in zip(tokens, ext_tokens):
-                self.failUnlessEqual(token.__class__, ext_token.__class__)
-                if not isinstance(token, yaml.StreamEndToken):
-                    self.failUnlessEqual((token.start_mark.index, token.start_mark.line, token.start_mark.column),
-                            (ext_token.start_mark.index, ext_token.start_mark.line, ext_token.start_mark.column))
-                    self.failUnlessEqual((token.end_mark.index, token.end_mark.line, token.end_mark.column),
-                            (ext_token.end_mark.index, ext_token.end_mark.line, ext_token.end_mark.column))
-                if hasattr(token, 'value'):
-                    self.failUnlessEqual(token.value, ext_token.value)
-        except:
-            print
-            print "DATA:"
-            print file(data_filename, 'rb').read()
-            print "TOKENS:", tokens
-            print "EXT_TOKENS:", ext_tokens
-            raise
-
-    def _testCParser(self, test_name, data_filename, canonical_filename, Loader=yaml.Loader):
-        data = file(data_filename, 'r').read()
-        events = list(yaml.parse(data, Loader=Loader))
-        ext_events = []
-        try:
-            for event in yaml.parse(data, Loader=yaml.CLoader):
-                ext_events.append(event)
-                #print "EVENT:", event
-            self.failUnlessEqual(len(events), len(ext_events))
-            for event, ext_event in zip(events, ext_events):
-                self.failUnlessEqual(event.__class__, ext_event.__class__)
-                if hasattr(event, 'anchor'):
-                    self.failUnlessEqual(event.anchor, ext_event.anchor)
-                if hasattr(event, 'tag'):
-                    self.failUnlessEqual(event.tag, ext_event.tag)
-                if hasattr(event, 'implicit'):
-                    self.failUnlessEqual(event.implicit, ext_event.implicit)
-                if hasattr(event, 'value'):
-                    self.failUnlessEqual(event.value, ext_event.value)
-                if hasattr(event, 'explicit'):
-                    self.failUnlessEqual(event.explicit, ext_event.explicit)
-                if hasattr(event, 'version'):
-                    self.failUnlessEqual(event.version, ext_event.version)
-                if hasattr(event, 'tags'):
-                    self.failUnlessEqual(event.tags, ext_event.tags)
-        except:
-            print
-            print "DATA:"
-            print file(data_filename, 'rb').read()
-            print "EVENTS:", events
-            print "EXT_EVENTS:", ext_events
-            raise
-
-TestCLoader.add_tests('testCScanner', '.data', '.canonical')
-TestCLoader.add_tests('testCScannerFileInput', '.data', '.canonical')
-TestCLoader.add_tests('testCParser', '.data', '.canonical')
-
-class TestCEmitter(test_appliance.TestAppliance):
-
-    def _testCEmitter(self, test_name, data_filename, canonical_filename, Loader=yaml.Loader):
-        data1 = file(data_filename, 'r').read()
-        events = list(yaml.parse(data1, Loader=Loader))
-        data2 = yaml.emit(events, Dumper=yaml.CDumper)
-        ext_events = []
-        try:
-            for event in yaml.parse(data2):
-                ext_events.append(event)
-            self.failUnlessEqual(len(events), len(ext_events))
-            for event, ext_event in zip(events, ext_events):
-                self.failUnlessEqual(event.__class__, ext_event.__class__)
-                if hasattr(event, 'anchor'):
-                    self.failUnlessEqual(event.anchor, ext_event.anchor)
-                if hasattr(event, 'tag'):
-                    if not (event.tag in ['!', None] and ext_event.tag in ['!', None]):
-                        self.failUnlessEqual(event.tag, ext_event.tag)
-                if hasattr(event, 'implicit'):
-                    self.failUnlessEqual(event.implicit, ext_event.implicit)
-                if hasattr(event, 'value'):
-                    self.failUnlessEqual(event.value, ext_event.value)
-                if hasattr(event, 'explicit') and event.explicit:
-                    self.failUnlessEqual(event.explicit, ext_event.explicit)
-                if hasattr(event, 'version'):
-                    self.failUnlessEqual(event.version, ext_event.version)
-                if hasattr(event, 'tags'):
-                    self.failUnlessEqual(event.tags, ext_event.tags)
-        except:
-            print
-            print "DATA1:"
-            print data1
-            print "DATA2:"
-            print data2
-            print "EVENTS:", events
-            print "EXT_EVENTS:", ext_events
-            raise
-
-TestCEmitter.add_tests('testCEmitter', '.data', '.canonical')
-
-yaml.BaseLoader = yaml.CBaseLoader
-yaml.SafeLoader = yaml.CSafeLoader
-yaml.Loader = yaml.CLoader
-yaml.BaseDumper = yaml.CBaseDumper
-yaml.SafeDumper = yaml.CSafeDumper
-yaml.Dumper = yaml.CDumper
-old_scan = yaml.scan
-def scan(stream, Loader=yaml.CLoader):
-    return old_scan(stream, Loader)
-yaml.scan = scan
-old_parse = yaml.parse
-def parse(stream, Loader=yaml.CLoader):
-    return old_parse(stream, Loader)
-yaml.parse = parse
-old_compose = yaml.compose
-def compose(stream, Loader=yaml.CLoader):
-    return old_compose(stream, Loader)
-yaml.compose = compose
-old_compose_all = yaml.compose_all
-def compose_all(stream, Loader=yaml.CLoader):
-    return old_compose_all(stream, Loader)
-yaml.compose_all = compose_all
-old_load_all = yaml.load_all
-def load_all(stream, Loader=yaml.CLoader):
-    return old_load_all(stream, Loader)
-yaml.load_all = load_all
-old_load = yaml.load
-def load(stream, Loader=yaml.CLoader):
-    return old_load(stream, Loader)
-yaml.load = load
-def safe_load_all(stream):
-    return yaml.load_all(stream, yaml.CSafeLoader)
-yaml.safe_load_all = safe_load_all
-def safe_load(stream):
-    return yaml.load(stream, yaml.CSafeLoader)
-yaml.safe_load = safe_load
-old_emit = yaml.emit
-def emit(events, stream=None, Dumper=yaml.CDumper, **kwds):
-    return old_emit(events, stream, Dumper, **kwds)
-yaml.emit = emit
-old_serialize_all = yaml.serialize_all
-def serialize_all(nodes, stream=None, Dumper=yaml.CDumper, **kwds):
-    return old_serialize_all(nodes, stream, Dumper, **kwds)
-yaml.serialize_all = serialize_all
-old_serialize = yaml.serialize
-def serialize(node, stream, Dumper=yaml.CDumper, **kwds):
-    return old_serialize(node, stream, Dumper, **kwds)
-yaml.serialize = serialize
-old_dump_all = yaml.dump_all
-def dump_all(documents, stream=None, Dumper=yaml.CDumper, **kwds):
-    return old_dump_all(documents, stream, Dumper, **kwds)
-yaml.dump_all = dump_all
-old_dump = yaml.dump
-def dump(data, stream=None, Dumper=yaml.CDumper, **kwds):
-    return old_dump(data, stream, Dumper, **kwds)
-yaml.dump = dump
-def safe_dump_all(documents, stream=None, **kwds):
-    return yaml.dump_all(documents, stream, yaml.CSafeDumper, **kwds)
-yaml.safe_dump_all = safe_dump_all
-def safe_dump(data, stream=None, **kwds):
-    return yaml.dump(data, stream, yaml.CSafeDumper, **kwds)
-yaml.safe_dump = safe_dump
-
-from test_yaml import *
-
-def main(module='__main__'):
-    unittest.main(module)
+            function(*args, **kwds)
+        finally:
+            _tear_down()
+    wrapper.func_name = '%s_ext' % function.func_name
+    wrapper.unittest = function.unittest
+    wrapper.skip = getattr(function, 'skip', [])+['.skip-ext']
+    return wrapper
+
+def wrap_ext(collections):
+    functions = []
+    if not isinstance(collections, list):
+        collections = [collections]
+    for collection in collections:
+        if not isinstance(collection, dict):
+            collection = vars(collection)
+        keys = collection.keys()
+        keys.sort()
+        for key in keys:
+            value = collection[key]
+            if isinstance(value, types.FunctionType) and hasattr(value, 'unittest'):
+                functions.append(wrap_ext_function(value))
+    for function in functions:
+        assert function.func_name not in globals()
+        globals()[function.func_name] = function
+
+import test_tokens, test_structure, test_errors, test_resolver, test_constructor,   \
+        test_emitter, test_representer, test_recursive
+wrap_ext([test_tokens, test_structure, test_errors, test_resolver, test_constructor,
+        test_emitter, test_representer, test_recursive])
 
 if __name__ == '__main__':
-    main()
-
+    import test_appliance
+    test_appliance.run(globals())
+