summaryrefslogtreecommitdiff
path: root/tests/test_resolver.py
blob: 4e140104bd9eb99ca172a3d3277442df8e4c0dc5 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

import test_appliance

from yaml import *

class MyLoader(Loader):
    pass

class MyDumper(Dumper):
    pass

add_path_resolver(u'!root', [],
        Loader=MyLoader, Dumper=MyDumper)

add_path_resolver(u'!root/scalar', [], str,
        Loader=MyLoader, Dumper=MyDumper)

add_path_resolver(u'!root/key11/key12/*', ['key11', 'key12'],
        Loader=MyLoader, Dumper=MyDumper)

add_path_resolver(u'!root/key21/1/*', ['key21', 1],
        Loader=MyLoader, Dumper=MyDumper)

add_path_resolver(u'!root/key31/*/*/key14/map', ['key31', None, None, 'key14'], dict,
        Loader=MyLoader, Dumper=MyDumper)

class TestResolver(test_appliance.TestAppliance):

    def _testImplicitResolver(self, test_name, data_filename, detect_filename):
        node = None
        correct_tag = None
        try:
            correct_tag = file(detect_filename, 'rb').read().strip()
            node = compose(file(data_filename, 'rb'))
            self.failUnless(isinstance(node, SequenceNode))
            for scalar in node.value:
                self.failUnless(isinstance(scalar, ScalarNode))
                self.failUnlessEqual(scalar.tag, correct_tag)
        except:
            print
            print "DATA:"
            print file(data_filename, 'rb').read()
            print "CORRECT_TAG:"
            print file(detect_filename, 'rb').read()
            print "ROOT NODE:", node
            print "SCALAR NODES:", node.value
            raise

    def _testPathResolverLoader(self, test_name, data_filename, path_filename):
        #print serialize_all(compose_all(file(data_filename, 'rb').read(), Loader=MyLoader))
        nodes1 = compose_all(file(data_filename, 'rb').read(), Loader=MyLoader)
        nodes2 = compose_all(file(path_filename, 'rb').read())
        for node1, node2 in zip(nodes1, nodes2):
            self.failUnlessEqual(self._convert(node1), self._convert(node2))

    def _testPathResolverDumper(self, test_name, data_filename, path_filename):
        for filename in [data_filename, path_filename]:
            output = serialize_all(compose_all(file(filename, 'rb').read()), Dumper=MyDumper)
            #print output
            nodes1 = compose_all(output)
            nodes2 = compose_all(file(data_filename, 'rb').read())
            for node1, node2 in zip(nodes1, nodes2):
                self.failUnlessEqual(self._convert(node1), self._convert(node2))

    def _convert(self, node):
        if isinstance(node, ScalarNode):
            return node.tag, node.value
        elif isinstance(node, SequenceNode):
            value = []
            for item in node.value:
                value.append(self._convert(item))
            return node.tag, value
        elif isinstance(node, MappingNode):
            value = []
            for key in node.value:
                item = node.value[key]
                value.append((self._convert(key), self._convert(item)))
            value.sort()
            return node.tag, value

TestResolver.add_tests('testImplicitResolver', '.data', '.detect')
TestResolver.add_tests('testPathResolverLoader', '.data', '.path')
TestResolver.add_tests('testPathResolverDumper', '.data', '.path')