Source

pyyaml / tests / test_structure.py

Full commit
Kirill Simonov ed172c1 










Kirill Simonov 6932ece 
Kirill Simonov ed172c1 
Kirill Simonov 6932ece 


Kirill Simonov a4348b1 
Kirill Simonov 6932ece 

Kirill Simonov ed172c1 










Kirill Simonov 6932ece 


Kirill Simonov ed172c1 



Kirill Simonov 6932ece 

Kirill Simonov ed172c1 
Kirill Simonov 6932ece 


Kirill Simonov ed172c1 
Kirill Simonov 6932ece 

Kirill Simonov ed172c1 
Kirill Simonov 6932ece 


Kirill Simonov ed172c1 
Kirill Simonov 6932ece 
Kirill Simonov ed172c1 
Kirill Simonov 6932ece 

Kirill Simonov ed172c1 

Kirill Simonov 6932ece 
Kirill Simonov ed172c1 









Kirill Simonov 6932ece 

Kirill Simonov ed172c1 





















Kirill Simonov ea86bf3 
Kirill Simonov ed172c1 











Kirill Simonov 6932ece 

Kirill Simonov ed172c1 



































Kirill Simonov 2d77d67 
Kirill Simonov ed172c1 
Kirill Simonov 2d77d67 
Kirill Simonov ed172c1 





Kirill Simonov 6932ece 


Kirill Simonov 988f014 
Kirill Simonov 6932ece 

Kirill Simonov 2d77d67 












Kirill Simonov 988f014 
Kirill Simonov 6932ece 
Kirill Simonov ed172c1 



Kirill Simonov 6932ece 

Kirill Simonov ed172c1 
Kirill Simonov 6932ece 


Kirill Simonov ed172c1 





Kirill Simonov 6932ece 

Kirill Simonov ed172c1 









Kirill Simonov 6932ece 

Kirill Simonov ed172c1 


















Kirill Simonov ea86bf3 
Kirill Simonov ed172c1 




import test_appliance

from yaml import *

class TestStructure(test_appliance.TestAppliance):

    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

    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 '?'

TestStructure.add_tests('testStructure', '.data', '.structure')

class TestParser(test_appliance.TestAppliance):

    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 _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

TestParser.add_tests('testParser', '.data', '.canonical')

class TestResolver(test_appliance.TestAppliance):

    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

    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)

TestResolver.add_tests('testResolver', '.data', '.canonical')

class MyLoader(Loader):
    def construct_sequence(self, node):
        return tuple(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)

MyLoader.add_constructor(u'tag:yaml.org,2002:map', MyLoader.construct_mapping)
MyLoader.add_constructor(None, MyLoader.construct_undefined)

class MyCanonicalLoader(test_appliance.CanonicalLoader):

    def construct_sequence(self, node):
        return tuple(test_appliance.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)

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')