Index: pyyaml/trunk/tests/test_structure.py =================================================================== --- pyyaml/trunk/tests/test_structure.py (revision 222) +++ pyyaml/trunk/tests/test_structure.py (revision 322) @@ -1,228 +1,187 @@ -import test_appliance +import yaml, canonical +import pprint -from yaml import * +def _convert_structure(loader): + if loader.check_event(yaml.ScalarEvent): + event = loader.get_event() + if event.tag or event.anchor or event.value: + return True + else: + return None + elif loader.check_event(yaml.SequenceStartEvent): + loader.get_event() + sequence = [] + while not loader.check_event(yaml.SequenceEndEvent): + sequence.append(_convert_structure(loader)) + loader.get_event() + return sequence + elif loader.check_event(yaml.MappingStartEvent): + loader.get_event() + mapping = [] + while not loader.check_event(yaml.MappingEndEvent): + key = _convert_structure(loader) + value = _convert_structure(loader) + mapping.append((key, value)) + loader.get_event() + return mapping + elif loader.check_event(yaml.AliasEvent): + loader.get_event() + return '*' + else: + loader.get_event() + return '?' -class TestStructure(test_appliance.TestAppliance): +def test_structure(data_filename, structure_filename, verbose=False): + nodes1 = [] + nodes2 = eval(open(structure_filename, 'rb').read()) + try: + loader = yaml.Loader(open(data_filename, 'rb')) + while loader.check_event(): + if loader.check_event(yaml.StreamStartEvent, yaml.StreamEndEvent, + yaml.DocumentStartEvent, yaml.DocumentEndEvent): + loader.get_event() + continue + nodes1.append(_convert_structure(loader)) + if len(nodes1) == 1: + nodes1 = nodes1[0] + assert nodes1 == nodes2, (nodes1, nodes2) + finally: + if verbose: + print "NODES1:" + pprint.pprint(nodes1) + print "NODES2:" + pprint.pprint(nodes2) - def _testStructure(self, test_name, data_filename, structure_filename): - node1 = None - node2 = eval(file(structure_filename, 'rb').read()) - try: - loader = Loader(file(data_filename, 'rb')) - node1 = [] - while not loader.check_event(StreamEndEvent): - if not loader.check_event(StreamStartEvent, DocumentStartEvent, DocumentEndEvent): - node1.append(self._convert(loader)) - else: - loader.get_event() - loader.get_event() - if len(node1) == 1: - node1 = node1[0] - self.failUnlessEqual(node1, node2) - except: - print - print "DATA:" - print file(data_filename, 'rb').read() - print "NODE1:", node1 - print "NODE2:", node2 - raise +test_structure.unittest = ['.data', '.structure'] - def _convert(self, loader): - if loader.check_event(ScalarEvent): - event = loader.get_event() - if event.tag or event.anchor or event.value: - return True - else: - return None - elif loader.check_event(SequenceStartEvent): - loader.get_event() - sequence = [] - while not loader.check_event(SequenceEndEvent): - sequence.append(self._convert(loader)) - loader.get_event() - return sequence - elif loader.check_event(MappingStartEvent): - loader.get_event() - mapping = [] - while not loader.check_event(MappingEndEvent): - key = self._convert(loader) - value = self._convert(loader) - mapping.append((key, value)) - loader.get_event() - return mapping - elif loader.check_event(AliasEvent): - loader.get_event() - return '*' - else: - loader.get_event() - return '?' +def _compare_events(events1, events2, full=False): + assert len(events1) == len(events2), (len(events1), len(events2)) + for event1, event2 in zip(events1, events2): + assert event1.__class__ == event2.__class__, (event1, event2) + if isinstance(event1, yaml.AliasEvent) and full: + assert event1.anchor == event2.anchor, (event1, event2) + if isinstance(event1, (yaml.ScalarEvent, yaml.CollectionStartEvent)): + if (event1.tag not in [None, u'!'] and event2.tag not in [None, u'!']) or full: + assert event1.tag == event2.tag, (event1, event2) + if isinstance(event1, yaml.ScalarEvent): + assert event1.value == event2.value, (event1, event2) -TestStructure.add_tests('testStructure', '.data', '.structure') +def test_parser(data_filename, canonical_filename, verbose=False): + events1 = None + events2 = None + try: + events1 = list(yaml.parse(open(data_filename, 'rb'))) + events2 = list(yaml.canonical_parse(open(canonical_filename, 'rb'))) + _compare_events(events1, events2) + finally: + if verbose: + print "EVENTS1:" + pprint.pprint(events1) + print "EVENTS2:" + pprint.pprint(events2) -class TestParser(test_appliance.TestAppliance): +test_parser.unittest = ['.data', '.canonical'] - def _testParser(self, test_name, data_filename, canonical_filename): - events1 = None - events2 = None - try: - events1 = list(parse(file(data_filename, 'rb'))) - events2 = list(test_appliance.canonical_parse(file(canonical_filename, 'rb'))) - self._compare(events1, events2) - except: - print - print "DATA1:" - print file(data_filename, 'rb').read() - print "DATA2:" - print file(canonical_filename, 'rb').read() - print "EVENTS1:", events1 - print "EVENTS2:", events2 - raise +def test_parser_on_canonical(canonical_filename, verbose=False): + events1 = None + events2 = None + try: + events1 = list(yaml.parse(open(canonical_filename, 'rb'))) + events2 = list(yaml.canonical_parse(open(canonical_filename, 'rb'))) + _compare_events(events1, events2, full=True) + finally: + if verbose: + print "EVENTS1:" + pprint.pprint(events1) + print "EVENTS2:" + pprint.pprint(events2) - def _compare(self, events1, events2): - self.failUnlessEqual(len(events1), len(events2)) - for event1, event2 in zip(events1, events2): - self.failUnlessEqual(event1.__class__, event2.__class__) - if isinstance(event1, AliasEvent): - #self.failUnlessEqual(event1.name, event2.name) - pass - elif isinstance(event1, ScalarEvent): - #self.failUnlessEqual(event1.anchor, event2.anchor) - #self.failUnlessEqual(event1.tag, event2.tag) - self.failUnlessEqual(event1.value, event2.value) - if isinstance(event1, CollectionStartEvent): - #self.failUnlessEqual(event1.anchor, event2.anchor) - #self.failUnlessEqual(event1.tag, event2.tag) - pass +test_parser_on_canonical.unittest = ['.canonical'] -TestParser.add_tests('testParser', '.data', '.canonical') +def _compare_nodes(node1, node2): + assert node1.__class__ == node2.__class__, (node1, node2) + assert node1.tag == node2.tag, (node1, node2) + if isinstance(node1, yaml.ScalarNode): + assert node1.value == node2.value, (node1, node2) + else: + assert len(node1.value) == len(node2.value), (node1, node2) + for item1, item2 in zip(node1.value, node2.value): + if not isinstance(item1, tuple): + item1 = (item1,) + item2 = (item2,) + for subnode1, subnode2 in zip(item1, item2): + _compare_nodes(subnode1, subnode2) -class TestResolver(test_appliance.TestAppliance): +def test_composer(data_filename, canonical_filename, verbose=False): + nodes1 = None + nodes2 = None + try: + nodes1 = list(yaml.compose_all(open(data_filename, 'rb'))) + nodes2 = list(yaml.canonical_compose_all(open(canonical_filename, 'rb'))) + assert len(nodes1) == len(nodes2), (len(nodes1), len(nodes2)) + for node1, node2 in zip(nodes1, nodes2): + _compare_nodes(node1, node2) + finally: + if verbose: + print "NODES1:" + pprint.pprint(nodes1) + print "NODES2:" + pprint.pprint(nodes2) - def _testResolver(self, test_name, data_filename, canonical_filename): - nodes1 = None - nodes2 = None - try: - nodes1 = list(compose_all(file(data_filename, 'rb'))) - nodes2 = list(test_appliance.canonical_compose_all(file(canonical_filename, 'rb'))) - self.failUnlessEqual(len(nodes1), len(nodes2)) - for node1, node2 in zip(nodes1, nodes2): - self._compare(node1, node2) - except: - print - print "DATA1:" - print file(data_filename, 'rb').read() - print "DATA2:" - print file(canonical_filename, 'rb').read() - print "NODES1:", nodes1 - print "NODES2:", nodes2 - raise +test_composer.unittest = ['.data', '.canonical'] - def _compare(self, node1, node2): - self.failUnlessEqual(node1.__class__, node2.__class__) - if isinstance(node1, ScalarNode): - #self.failUnlessEqual(node1.tag, node2.tag) - self.failUnlessEqual(node1.value, node2.value) - elif isinstance(node1, SequenceNode): - self.failUnlessEqual(len(node1.value), len(node2.value)) - for item1, item2 in zip(node1.value, node2.value): - self._compare(item1, item2) - elif isinstance(node1, MappingNode): - self.failUnlessEqual(len(node1.value), len(node2.value)) - items1 = node1.value.items() - items1.sort(lambda (k1,v1), (k2,v2): cmp((k1.tag,k1.value,v1.tag,v1.value), - (k2.tag,k2.value,v2.tag,v2.value))) - items2 = node2.value.items() - items2.sort(lambda (k1,v1), (k2,v2): cmp((k1.tag,k1.value,v1.tag,v1.value), - (k2.tag,k2.value,v2.tag,v2.value))) - for (key1, value1), (key2, value2) in zip(items1, items2): - self._compare(key1, key2) - self._compare(value1, value2) +def _make_loader(): + global MyLoader -TestResolver.add_tests('testResolver', '.data', '.canonical') + class MyLoader(yaml.Loader): + def construct_sequence(self, node): + return tuple(yaml.Loader.construct_sequence(self, node)) + def construct_mapping(self, node): + pairs = self.construct_pairs(node) + pairs.sort() + return pairs + def construct_undefined(self, node): + return self.construct_scalar(node) -class MyLoader(Loader): - def construct_sequence(self, node): - return tuple(Loader.construct_sequence(self, node)) + MyLoader.add_constructor(u'tag:yaml.org,2002:map', MyLoader.construct_mapping) + MyLoader.add_constructor(None, MyLoader.construct_undefined) - def construct_mapping(self, node): - pairs = self.construct_pairs(node) - pairs.sort() - return pairs +def _make_canonical_loader(): + global MyCanonicalLoader - def construct_undefined(self, node): - return self.construct_scalar(node) + class MyCanonicalLoader(yaml.CanonicalLoader): + def construct_sequence(self, node): + return tuple(yaml.CanonicalLoader.construct_sequence(self, node)) + def construct_mapping(self, node): + pairs = self.construct_pairs(node) + pairs.sort() + return pairs + def construct_undefined(self, node): + return self.construct_scalar(node) -MyLoader.add_constructor(u'tag:yaml.org,2002:map', MyLoader.construct_mapping) -MyLoader.add_constructor(None, MyLoader.construct_undefined) + MyCanonicalLoader.add_constructor(u'tag:yaml.org,2002:map', MyCanonicalLoader.construct_mapping) + MyCanonicalLoader.add_constructor(None, MyCanonicalLoader.construct_undefined) -class MyCanonicalLoader(test_appliance.CanonicalLoader): +def test_constructor(data_filename, canonical_filename, verbose=False): + _make_loader() + _make_canonical_loader() + native1 = None + native2 = None + try: + native1 = list(yaml.load_all(open(data_filename, 'rb'), Loader=MyLoader)) + native2 = list(yaml.load_all(open(canonical_filename, 'rb'), Loader=MyCanonicalLoader)) + assert native1 == native2, (native1, native2) + finally: + if verbose: + print "NATIVE1:" + pprint.pprint(native1) + print "NATIVE2:" + pprint.pprint(native2) - def construct_sequence(self, node): - return tuple(test_appliance.CanonicalLoader.construct_sequence(self, node)) +test_constructor.unittest = ['.data', '.canonical'] - def construct_mapping(self, node): - pairs = self.construct_pairs(node) - pairs.sort() - return pairs +if __name__ == '__main__': + import test_appliance + test_appliance.run(globals()) - def construct_undefined(self, node): - return self.construct_scalar(node) - -MyCanonicalLoader.add_constructor(u'tag:yaml.org,2002:map', MyCanonicalLoader.construct_mapping) -MyCanonicalLoader.add_constructor(None, MyCanonicalLoader.construct_undefined) - -class TestConstructor(test_appliance.TestAppliance): - - def _testConstructor(self, test_name, data_filename, canonical_filename): - data1 = None - data2 = None - try: - data1 = list(load_all(file(data_filename, 'rb'), Loader=MyLoader)) - data2 = list(load_all(file(canonical_filename, 'rb'), Loader=MyCanonicalLoader)) - self.failUnlessEqual(data1, data2) - except: - print - print "DATA1:" - print file(data_filename, 'rb').read() - print "DATA2:" - print file(canonical_filename, 'rb').read() - print "NATIVES1:", data1 - print "NATIVES2:", data2 - raise - -TestConstructor.add_tests('testConstructor', '.data', '.canonical') - -class TestParserOnCanonical(test_appliance.TestAppliance): - - def _testParserOnCanonical(self, test_name, canonical_filename): - events1 = None - events2 = None - try: - events1 = list(parse(file(canonical_filename, 'rb'))) - events2 = list(test_appliance.canonical_parse(file(canonical_filename, 'rb'))) - self._compare(events1, events2) - except: - print - print "DATA:" - print file(canonical_filename, 'rb').read() - print "EVENTS1:", events1 - print "EVENTS2:", events2 - raise - - def _compare(self, events1, events2): - self.failUnlessEqual(len(events1), len(events2)) - for event1, event2 in zip(events1, events2): - self.failUnlessEqual(event1.__class__, event2.__class__) - if isinstance(event1, AliasEvent): - self.failUnlessEqual(event1.anchor, event2.anchor) - elif isinstance(event1, ScalarEvent): - self.failUnlessEqual(event1.anchor, event2.anchor) - self.failUnlessEqual(event1.tag, event2.tag) - self.failUnlessEqual(event1.value, event2.value) - if isinstance(event1, CollectionStartEvent): - self.failUnlessEqual(event1.anchor, event2.anchor) - self.failUnlessEqual(event1.tag, event2.tag) - -TestParserOnCanonical.add_tests('testParserOnCanonical', '.canonical') -