third_party_pyyaml/tests/test_yaml_ext.py

import unittest, test_appliance

import _yaml, yaml

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 = []
        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'):
                    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)

if __name__ == '__main__':
    main()