summaryrefslogtreecommitdiff
path: root/_test/lib/test_emitter.py
blob: b1991e3b327f7af453b097e9ffa07b3008f1193b (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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
from __future__ import absolute_import
from __future__ import print_function

import ruamel.yaml
from ruamel.yaml import YAML


def _compare_events(events1, events2):
    assert len(events1) == len(events2), (events1, events2)
    for event1, event2 in zip(events1, events2):
        assert event1.__class__ == event2.__class__, (event1, event2)
        if isinstance(event1, yaml.NodeEvent):
            assert event1.anchor == event2.anchor, (event1, event2)
        if isinstance(event1, yaml.CollectionStartEvent):
            assert event1.tag == event2.tag, (event1, event2)
        if isinstance(event1, yaml.ScalarEvent):
            if True not in event1.implicit + event2.implicit:
                assert event1.tag == event2.tag, (event1, event2)
            assert event1.value == event2.value, (event1, event2)


def test_emitter_on_data(data_filename, canonical_filename, verbose=False):
    with open(data_filename, 'rb') as fp0:
        events = list(YAML().parse(fp0))
    output = YAML().emit(events)
    if verbose:
        print('OUTPUT:')
        print(output)
    new_events = list(yaml.parse(output))
    _compare_events(events, new_events)


test_emitter_on_data.unittest = ['.data', '.canonical']


def test_emitter_on_canonical(canonical_filename, verbose=False):
    with open(canonical_filename, 'rb') as fp0:
        events = list(YAML().parse(fp0))
    for canonical in [False, True]:
        output = YAML().emit(events, canonical=canonical)
        if verbose:
            print('OUTPUT (canonical=%s):' % canonical)
            print(output)
        new_events = list(yaml.parse(output))
        _compare_events(events, new_events)


test_emitter_on_canonical.unittest = ['.canonical']


def test_emitter_styles(data_filename, canonical_filename, verbose=False):
    for filename in [data_filename, canonical_filename]:
        with open(filename, 'rb') as fp0:
            events = list(YAML().parse(fp0))
        for flow_style in [False, True]:
            for style in ['|', '>', '"', "'", ""]:
                styled_events = []
                for event in events:
                    if isinstance(event, yaml.ScalarEvent):
                        event = yaml.ScalarEvent(
                            event.anchor, event.tag, event.implicit, event.value, style=style
                        )
                    elif isinstance(event, yaml.SequenceStartEvent):
                        event = yaml.SequenceStartEvent(
                            event.anchor, event.tag, event.implicit, flow_style=flow_style
                        )
                    elif isinstance(event, yaml.MappingStartEvent):
                        event = yaml.MappingStartEvent(
                            event.anchor, event.tag, event.implicit, flow_style=flow_style
                        )
                    styled_events.append(event)
                output = YAML().emit(styled_events)
                if verbose:
                    print(
                        'OUTPUT (filename=%r, flow_style=%r, style=%r)'
                        % (filename, flow_style, style)
                    )
                    print(output)
                new_events = list(YAML().parse(output))
                _compare_events(events, new_events)


test_emitter_styles.unittest = ['.data', '.canonical']


class EventsLoader(ruamel.yaml.Loader):
    def construct_event(self, node):
        if isinstance(node, ruamel.yaml.ScalarNode):
            mapping = {}
        else:
            mapping = self.construct_mapping(node)
        class_name = str(node.tag[1:]) + 'Event'
        if class_name in [
            'AliasEvent',
            'ScalarEvent',
            'SequenceStartEvent',
            'MappingStartEvent',
        ]:
            mapping.setdefault('anchor', None)
        if class_name in ['ScalarEvent', 'SequenceStartEvent', 'MappingStartEvent']:
            mapping.setdefault('tag', None)
        if class_name in ['SequenceStartEvent', 'MappingStartEvent']:
            mapping.setdefault('implicit', True)
        if class_name == 'ScalarEvent':
            mapping.setdefault('implicit', (False, True))
            mapping.setdefault('value', "")
        value = getattr(yaml, class_name)(**mapping)
        return value


# if Loader is not a composite, add this function
# EventsLoader.add_constructor = yaml.constructor.Constructor.add_constructor


EventsLoader.add_constructor(None, EventsLoader.construct_event)


def test_emitter_events(events_filename, verbose=False):
    with open(events_filename, 'rb') as fp0:
        events = list(YAML().load(fp0, Loader=EventsLoader))
    output = YAML().emit(events)
    if verbose:
        print('OUTPUT:')
        print(output)
    new_events = list(YAML().parse(output))
    _compare_events(events, new_events)


if __name__ == '__main__':
    import test_appliance

    test_appliance.run(globals())