source: pyyaml/trunk/tests/lib3/canonical.py @ 330

Revision 330, 12.0 KB checked in by xi, 4 years ago (diff)

Share data files between Py2 and Py3 test suites.
Line 
1
2import yaml, yaml.composer, yaml.constructor, yaml.resolver
3
4class CanonicalError(yaml.YAMLError):
5    pass
6
7class CanonicalScanner:
8
9    def __init__(self, data):
10        if isinstance(data, bytes):
11            try:
12                data = data.decode('utf-8')
13            except UnicodeDecodeError:
14                raise CanonicalError("utf-8 stream is expected")
15        self.data = data+'\0'
16        self.index = 0
17        self.tokens = []
18        self.scanned = False
19
20    def check_token(self, *choices):
21        if not self.scanned:
22            self.scan()
23        if self.tokens:
24            if not choices:
25                return True
26            for choice in choices:
27                if isinstance(self.tokens[0], choice):
28                    return True
29        return False
30
31    def peek_token(self):
32        if not self.scanned:
33            self.scan()
34        if self.tokens:
35            return self.tokens[0]
36
37    def get_token(self, choice=None):
38        if not self.scanned:
39            self.scan()
40        token = self.tokens.pop(0)
41        if choice and not isinstance(token, choice):
42            raise CanonicalError("unexpected token "+repr(token))
43        return token
44
45    def get_token_value(self):
46        token = self.get_token()
47        return token.value
48
49    def scan(self):
50        self.tokens.append(yaml.StreamStartToken(None, None))
51        while True:
52            self.find_token()
53            ch = self.data[self.index]
54            if ch == '\0':
55                self.tokens.append(yaml.StreamEndToken(None, None))
56                break
57            elif ch == '%':
58                self.tokens.append(self.scan_directive())
59            elif ch == '-' and self.data[self.index:self.index+3] == '---':
60                self.index += 3
61                self.tokens.append(yaml.DocumentStartToken(None, None))
62            elif ch == '[':
63                self.index += 1
64                self.tokens.append(yaml.FlowSequenceStartToken(None, None))
65            elif ch == '{':
66                self.index += 1
67                self.tokens.append(yaml.FlowMappingStartToken(None, None))
68            elif ch == ']':
69                self.index += 1
70                self.tokens.append(yaml.FlowSequenceEndToken(None, None))
71            elif ch == '}':
72                self.index += 1
73                self.tokens.append(yaml.FlowMappingEndToken(None, None))
74            elif ch == '?':
75                self.index += 1
76                self.tokens.append(yaml.KeyToken(None, None))
77            elif ch == ':':
78                self.index += 1
79                self.tokens.append(yaml.ValueToken(None, None))
80            elif ch == ',':
81                self.index += 1
82                self.tokens.append(yaml.FlowEntryToken(None, None))
83            elif ch == '*' or ch == '&':
84                self.tokens.append(self.scan_alias())
85            elif ch == '!':
86                self.tokens.append(self.scan_tag())
87            elif ch == '"':
88                self.tokens.append(self.scan_scalar())
89            else:
90                raise CanonicalError("invalid token")
91        self.scanned = True
92
93    DIRECTIVE = '%YAML 1.1'
94
95    def scan_directive(self):
96        if self.data[self.index:self.index+len(self.DIRECTIVE)] == self.DIRECTIVE and \
97                self.data[self.index+len(self.DIRECTIVE)] in ' \n\0':
98            self.index += len(self.DIRECTIVE)
99            return yaml.DirectiveToken('YAML', (1, 1), None, None)
100        else:
101            raise CanonicalError("invalid directive")
102
103    def scan_alias(self):
104        if self.data[self.index] == '*':
105            TokenClass = yaml.AliasToken
106        else:
107            TokenClass = yaml.AnchorToken
108        self.index += 1
109        start = self.index
110        while self.data[self.index] not in ', \n\0':
111            self.index += 1
112        value = self.data[start:self.index]
113        return TokenClass(value, None, None)
114
115    def scan_tag(self):
116        self.index += 1
117        start = self.index
118        while self.data[self.index] not in ' \n\0':
119            self.index += 1
120        value = self.data[start:self.index]
121        if not value:
122            value = '!'
123        elif value[0] == '!':
124            value = 'tag:yaml.org,2002:'+value[1:]
125        elif value[0] == '<' and value[-1] == '>':
126            value = value[1:-1]
127        else:
128            value = '!'+value
129        return yaml.TagToken(value, None, None)
130
131    QUOTE_CODES = {
132        'x': 2,
133        'u': 4,
134        'U': 8,
135    }
136
137    QUOTE_REPLACES = {
138        '\\': '\\',
139        '\"': '\"',
140        ' ': ' ',
141        'a': '\x07',
142        'b': '\x08',
143        'e': '\x1B',
144        'f': '\x0C',
145        'n': '\x0A',
146        'r': '\x0D',
147        't': '\x09',
148        'v': '\x0B',
149        'N': '\u0085',
150        'L': '\u2028',
151        'P': '\u2029',
152        '_': '_',
153        '0': '\x00',
154    }
155
156    def scan_scalar(self):
157        self.index += 1
158        chunks = []
159        start = self.index
160        ignore_spaces = False
161        while self.data[self.index] != '"':
162            if self.data[self.index] == '\\':
163                ignore_spaces = False
164                chunks.append(self.data[start:self.index])
165                self.index += 1
166                ch = self.data[self.index]
167                self.index += 1
168                if ch == '\n':
169                    ignore_spaces = True
170                elif ch in self.QUOTE_CODES:
171                    length = self.QUOTE_CODES[ch]
172                    code = int(self.data[self.index:self.index+length], 16)
173                    chunks.append(chr(code))
174                    self.index += length
175                else:
176                    if ch not in self.QUOTE_REPLACES:
177                        raise CanonicalError("invalid escape code")
178                    chunks.append(self.QUOTE_REPLACES[ch])
179                start = self.index
180            elif self.data[self.index] == '\n':
181                chunks.append(self.data[start:self.index])
182                chunks.append(' ')
183                self.index += 1
184                start = self.index
185                ignore_spaces = True
186            elif ignore_spaces and self.data[self.index] == ' ':
187                self.index += 1
188                start = self.index
189            else:
190                ignore_spaces = False
191                self.index += 1
192        chunks.append(self.data[start:self.index])
193        self.index += 1
194        return yaml.ScalarToken(''.join(chunks), False, None, None)
195
196    def find_token(self):
197        found = False
198        while not found:
199            while self.data[self.index] in ' \t':
200                self.index += 1
201            if self.data[self.index] == '#':
202                while self.data[self.index] != '\n':
203                    self.index += 1
204            if self.data[self.index] == '\n':
205                self.index += 1
206            else:
207                found = True
208
209class CanonicalParser:
210
211    def __init__(self):
212        self.events = []
213        self.parsed = False
214
215    # stream: STREAM-START document* STREAM-END
216    def parse_stream(self):
217        self.get_token(yaml.StreamStartToken)
218        self.events.append(yaml.StreamStartEvent(None, None))
219        while not self.check_token(yaml.StreamEndToken):
220            if self.check_token(yaml.DirectiveToken, yaml.DocumentStartToken):
221                self.parse_document()
222            else:
223                raise CanonicalError("document is expected, got "+repr(self.tokens[0]))
224        self.get_token(yaml.StreamEndToken)
225        self.events.append(yaml.StreamEndEvent(None, None))
226
227    # document: DIRECTIVE? DOCUMENT-START node
228    def parse_document(self):
229        node = None
230        if self.check_token(yaml.DirectiveToken):
231            self.get_token(yaml.DirectiveToken)
232        self.get_token(yaml.DocumentStartToken)
233        self.events.append(yaml.DocumentStartEvent(None, None))
234        self.parse_node()
235        self.events.append(yaml.DocumentEndEvent(None, None))
236
237    # node: ALIAS | ANCHOR? TAG? (SCALAR|sequence|mapping)
238    def parse_node(self):
239        if self.check_token(yaml.AliasToken):
240            self.events.append(yaml.AliasEvent(self.get_token_value(), None, None))
241        else:
242            anchor = None
243            if self.check_token(yaml.AnchorToken):
244                anchor = self.get_token_value()
245            tag = None
246            if self.check_token(yaml.TagToken):
247                tag = self.get_token_value()
248            if self.check_token(yaml.ScalarToken):
249                self.events.append(yaml.ScalarEvent(anchor, tag, (False, False), self.get_token_value(), None, None))
250            elif self.check_token(yaml.FlowSequenceStartToken):
251                self.events.append(yaml.SequenceStartEvent(anchor, tag, None, None))
252                self.parse_sequence()
253            elif self.check_token(yaml.FlowMappingStartToken):
254                self.events.append(yaml.MappingStartEvent(anchor, tag, None, None))
255                self.parse_mapping()
256            else:
257                raise CanonicalError("SCALAR, '[', or '{' is expected, got "+repr(self.tokens[0]))
258
259    # sequence: SEQUENCE-START (node (ENTRY node)*)? ENTRY? SEQUENCE-END
260    def parse_sequence(self):
261        self.get_token(yaml.FlowSequenceStartToken)
262        if not self.check_token(yaml.FlowSequenceEndToken):
263            self.parse_node()
264            while not self.check_token(yaml.FlowSequenceEndToken):
265                self.get_token(yaml.FlowEntryToken)
266                if not self.check_token(yaml.FlowSequenceEndToken):
267                    self.parse_node()
268        self.get_token(yaml.FlowSequenceEndToken)
269        self.events.append(yaml.SequenceEndEvent(None, None))
270
271    # mapping: MAPPING-START (map_entry (ENTRY map_entry)*)? ENTRY? MAPPING-END
272    def parse_mapping(self):
273        self.get_token(yaml.FlowMappingStartToken)
274        if not self.check_token(yaml.FlowMappingEndToken):
275            self.parse_map_entry()
276            while not self.check_token(yaml.FlowMappingEndToken):
277                self.get_token(yaml.FlowEntryToken)
278                if not self.check_token(yaml.FlowMappingEndToken):
279                    self.parse_map_entry()
280        self.get_token(yaml.FlowMappingEndToken)
281        self.events.append(yaml.MappingEndEvent(None, None))
282
283    # map_entry: KEY node VALUE node
284    def parse_map_entry(self):
285        self.get_token(yaml.KeyToken)
286        self.parse_node()
287        self.get_token(yaml.ValueToken)
288        self.parse_node()
289
290    def parse(self):
291        self.parse_stream()
292        self.parsed = True
293
294    def get_event(self):
295        if not self.parsed:
296            self.parse()
297        return self.events.pop(0)
298
299    def check_event(self, *choices):
300        if not self.parsed:
301            self.parse()
302        if self.events:
303            if not choices:
304                return True
305            for choice in choices:
306                if isinstance(self.events[0], choice):
307                    return True
308        return False
309
310    def peek_event(self):
311        if not self.parsed:
312            self.parse()
313        return self.events[0]
314
315class CanonicalLoader(CanonicalScanner, CanonicalParser,
316        yaml.composer.Composer, yaml.constructor.Constructor, yaml.resolver.Resolver):
317
318    def __init__(self, stream):
319        if hasattr(stream, 'read'):
320            stream = stream.read()
321        CanonicalScanner.__init__(self, stream)
322        CanonicalParser.__init__(self)
323        yaml.composer.Composer.__init__(self)
324        yaml.constructor.Constructor.__init__(self)
325        yaml.resolver.Resolver.__init__(self)
326
327yaml.CanonicalLoader = CanonicalLoader
328
329def canonical_scan(stream):
330    return yaml.scan(stream, Loader=CanonicalLoader)
331
332yaml.canonical_scan = canonical_scan
333
334def canonical_parse(stream):
335    return yaml.parse(stream, Loader=CanonicalLoader)
336
337yaml.canonical_parse = canonical_parse
338
339def canonical_compose(stream):
340    return yaml.compose(stream, Loader=CanonicalLoader)
341
342yaml.canonical_compose = canonical_compose
343
344def canonical_compose_all(stream):
345    return yaml.compose_all(stream, Loader=CanonicalLoader)
346
347yaml.canonical_compose_all = canonical_compose_all
348
349def canonical_load(stream):
350    return yaml.load(stream, Loader=CanonicalLoader)
351
352yaml.canonical_load = canonical_load
353
354def canonical_load_all(stream):
355    return yaml.load_all(stream, Loader=CanonicalLoader)
356
357yaml.canonical_load_all = canonical_load_all
358
Note: See TracBrowser for help on using the repository browser.