diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-14 20:19:53 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-14 20:19:53 +0000 |
commit | e7ee850d46d54789979bf0c5244bae1825fb7149 (patch) | |
tree | 6e94ed55df9ec749682a3c792ce752d07892b968 /_test/lib/canonical.py | |
parent | Initial commit. (diff) | |
download | python-ruyaml-upstream.tar.xz python-ruyaml-upstream.zip |
Adding upstream version 0.91.0.upstream/0.91.0upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to '_test/lib/canonical.py')
-rw-r--r-- | _test/lib/canonical.py | 387 |
1 files changed, 387 insertions, 0 deletions
diff --git a/_test/lib/canonical.py b/_test/lib/canonical.py new file mode 100644 index 0000000..56fa297 --- /dev/null +++ b/_test/lib/canonical.py @@ -0,0 +1,387 @@ +import ruyaml +from ruyaml.composer import Composer +from ruyaml.constructor import Constructor +from ruyaml.resolver import Resolver + + +class CanonicalError(ruyaml.YAMLError): + pass + + +class CanonicalScanner: + def __init__(self, data): + try: + if isinstance(data, bytes): + data = data.decode('utf-8') + except UnicodeDecodeError: + raise CanonicalError('utf-8 stream is expected') + self.data = data + '\0' + self.index = 0 + self.tokens = [] + self.scanned = False + + def check_token(self, *choices): + if not self.scanned: + self.scan() + if self.tokens: + if not choices: + return True + for choice in choices: + if isinstance(self.tokens[0], choice): + return True + return False + + def peek_token(self): + if not self.scanned: + self.scan() + if self.tokens: + return self.tokens[0] + + def get_token(self, choice=None): + if not self.scanned: + self.scan() + token = self.tokens.pop(0) + if choice and not isinstance(token, choice): + raise CanonicalError('unexpected token ' + repr(token)) + return token + + def get_token_value(self): + token = self.get_token() + return token.value + + def scan(self): + self.tokens.append(ruyaml.StreamStartToken(None, None)) + while True: + self.find_token() + ch = self.data[self.index] + if ch == '\0': + self.tokens.append(ruyaml.StreamEndToken(None, None)) + break + elif ch == '%': + self.tokens.append(self.scan_directive()) + elif ch == '-' and self.data[self.index : self.index + 3] == '---': + self.index += 3 + self.tokens.append(ruyaml.DocumentStartToken(None, None)) + elif ch == '[': + self.index += 1 + self.tokens.append(ruyaml.FlowSequenceStartToken(None, None)) + elif ch == '{': + self.index += 1 + self.tokens.append(ruyaml.FlowMappingStartToken(None, None)) + elif ch == ']': + self.index += 1 + self.tokens.append(ruyaml.FlowSequenceEndToken(None, None)) + elif ch == '}': + self.index += 1 + self.tokens.append(ruyaml.FlowMappingEndToken(None, None)) + elif ch == '?': + self.index += 1 + self.tokens.append(ruyaml.KeyToken(None, None)) + elif ch == ':': + self.index += 1 + self.tokens.append(ruyaml.ValueToken(None, None)) + elif ch == ',': + self.index += 1 + self.tokens.append(ruyaml.FlowEntryToken(None, None)) + elif ch == '*' or ch == '&': + self.tokens.append(self.scan_alias()) + elif ch == '!': + self.tokens.append(self.scan_tag()) + elif ch == '"': + self.tokens.append(self.scan_scalar()) + else: + raise CanonicalError('invalid token') + self.scanned = True + + DIRECTIVE = '%YAML 1.1' + + def scan_directive(self): + if ( + self.data[self.index : self.index + len(self.DIRECTIVE)] == self.DIRECTIVE + and self.data[self.index + len(self.DIRECTIVE)] in ' \n\0' + ): + self.index += len(self.DIRECTIVE) + return ruyaml.DirectiveToken('YAML', (1, 1), None, None) + else: + raise CanonicalError('invalid directive') + + def scan_alias(self): + if self.data[self.index] == '*': + TokenClass = ruyaml.AliasToken + else: + TokenClass = ruyaml.AnchorToken + self.index += 1 + start = self.index + while self.data[self.index] not in ', \n\0': + self.index += 1 + value = self.data[start : self.index] + return TokenClass(value, None, None) + + def scan_tag(self): + self.index += 1 + start = self.index + while self.data[self.index] not in ' \n\0': + self.index += 1 + value = self.data[start : self.index] + if not value: + value = '!' + elif value[0] == '!': + value = 'tag:yaml.org,2002:' + value[1:] + elif value[0] == '<' and value[-1] == '>': + value = value[1:-1] + else: + value = '!' + value + return ruyaml.TagToken(value, None, None) + + QUOTE_CODES = {'x': 2, 'u': 4, 'U': 8} + + QUOTE_REPLACES = { + '\\': '\\', + '"': '"', + ' ': ' ', + 'a': '\x07', + 'b': '\x08', + 'e': '\x1B', + 'f': '\x0C', + 'n': '\x0A', + 'r': '\x0D', + 't': '\x09', + 'v': '\x0B', + 'N': '\u0085', + 'L': '\u2028', + 'P': '\u2029', + '_': '_', + '0': '\x00', + } + + def scan_scalar(self): + self.index += 1 + chunks = [] + start = self.index + ignore_spaces = False + while self.data[self.index] != '"': + if self.data[self.index] == '\\': + ignore_spaces = False + chunks.append(self.data[start : self.index]) + self.index += 1 + ch = self.data[self.index] + self.index += 1 + if ch == '\n': + ignore_spaces = True + elif ch in self.QUOTE_CODES: + length = self.QUOTE_CODES[ch] + code = int(self.data[self.index : self.index + length], 16) + chunks.append(chr(code)) + self.index += length + else: + if ch not in self.QUOTE_REPLACES: + raise CanonicalError('invalid escape code') + chunks.append(self.QUOTE_REPLACES[ch]) + start = self.index + elif self.data[self.index] == '\n': + chunks.append(self.data[start : self.index]) + chunks.append(' ') + self.index += 1 + start = self.index + ignore_spaces = True + elif ignore_spaces and self.data[self.index] == ' ': + self.index += 1 + start = self.index + else: + ignore_spaces = False + self.index += 1 + chunks.append(self.data[start : self.index]) + self.index += 1 + return ruyaml.ScalarToken("".join(chunks), False, None, None) + + def find_token(self): + found = False + while not found: + while self.data[self.index] in ' \t': + self.index += 1 + if self.data[self.index] == '#': + while self.data[self.index] != '\n': + self.index += 1 + if self.data[self.index] == '\n': + self.index += 1 + else: + found = True + + +class CanonicalParser: + def __init__(self): + self.events = [] + self.parsed = False + + def dispose(self): + pass + + # stream: STREAM-START document* STREAM-END + def parse_stream(self): + self.get_token(ruyaml.StreamStartToken) + self.events.append(ruyaml.StreamStartEvent(None, None)) + while not self.check_token(ruyaml.StreamEndToken): + if self.check_token(ruyaml.DirectiveToken, ruyaml.DocumentStartToken): + self.parse_document() + else: + raise CanonicalError( + 'document is expected, got ' + repr(self.tokens[0]) + ) + self.get_token(ruyaml.StreamEndToken) + self.events.append(ruyaml.StreamEndEvent(None, None)) + + # document: DIRECTIVE? DOCUMENT-START node + def parse_document(self): + # node = None + if self.check_token(ruyaml.DirectiveToken): + self.get_token(ruyaml.DirectiveToken) + self.get_token(ruyaml.DocumentStartToken) + self.events.append(ruyaml.DocumentStartEvent(None, None)) + self.parse_node() + self.events.append(ruyaml.DocumentEndEvent(None, None)) + + # node: ALIAS | ANCHOR? TAG? (SCALAR|sequence|mapping) + def parse_node(self): + if self.check_token(ruyaml.AliasToken): + self.events.append(ruyaml.AliasEvent(self.get_token_value(), None, None)) + else: + anchor = None + if self.check_token(ruyaml.AnchorToken): + anchor = self.get_token_value() + tag = None + if self.check_token(ruyaml.TagToken): + tag = self.get_token_value() + if self.check_token(ruyaml.ScalarToken): + self.events.append( + ruyaml.ScalarEvent( + anchor, tag, (False, False), self.get_token_value(), None, None + ) + ) + elif self.check_token(ruyaml.FlowSequenceStartToken): + self.events.append(ruyaml.SequenceStartEvent(anchor, tag, None, None)) + self.parse_sequence() + elif self.check_token(ruyaml.FlowMappingStartToken): + self.events.append(ruyaml.MappingStartEvent(anchor, tag, None, None)) + self.parse_mapping() + else: + raise CanonicalError( + "SCALAR, '[', or '{' is expected, got " + repr(self.tokens[0]) + ) + + # sequence: SEQUENCE-START (node (ENTRY node)*)? ENTRY? SEQUENCE-END + def parse_sequence(self): + self.get_token(ruyaml.FlowSequenceStartToken) + if not self.check_token(ruyaml.FlowSequenceEndToken): + self.parse_node() + while not self.check_token(ruyaml.FlowSequenceEndToken): + self.get_token(ruyaml.FlowEntryToken) + if not self.check_token(ruyaml.FlowSequenceEndToken): + self.parse_node() + self.get_token(ruyaml.FlowSequenceEndToken) + self.events.append(ruyaml.SequenceEndEvent(None, None)) + + # mapping: MAPPING-START (map_entry (ENTRY map_entry)*)? ENTRY? MAPPING-END + def parse_mapping(self): + self.get_token(ruyaml.FlowMappingStartToken) + if not self.check_token(ruyaml.FlowMappingEndToken): + self.parse_map_entry() + while not self.check_token(ruyaml.FlowMappingEndToken): + self.get_token(ruyaml.FlowEntryToken) + if not self.check_token(ruyaml.FlowMappingEndToken): + self.parse_map_entry() + self.get_token(ruyaml.FlowMappingEndToken) + self.events.append(ruyaml.MappingEndEvent(None, None)) + + # map_entry: KEY node VALUE node + def parse_map_entry(self): + self.get_token(ruyaml.KeyToken) + self.parse_node() + self.get_token(ruyaml.ValueToken) + self.parse_node() + + def parse(self): + self.parse_stream() + self.parsed = True + + def get_event(self): + if not self.parsed: + self.parse() + return self.events.pop(0) + + def check_event(self, *choices): + if not self.parsed: + self.parse() + if self.events: + if not choices: + return True + for choice in choices: + if isinstance(self.events[0], choice): + return True + return False + + def peek_event(self): + if not self.parsed: + self.parse() + return self.events[0] + + +class CanonicalLoader( + CanonicalScanner, CanonicalParser, Composer, Constructor, Resolver +): + def __init__(self, stream): + if hasattr(stream, 'read'): + stream = stream.read() + CanonicalScanner.__init__(self, stream) + CanonicalParser.__init__(self) + Composer.__init__(self) + Constructor.__init__(self) + Resolver.__init__(self) + + +ruyaml.CanonicalLoader = CanonicalLoader + + +def canonical_scan(stream): + yaml = ruyaml.YAML() + yaml.scanner = CanonicalScanner + return yaml.scan(stream) + + +ruyaml.canonical_scan = canonical_scan + + +def canonical_parse(stream): + return ruyaml.parse(stream, Loader=CanonicalLoader) + + +ruyaml.canonical_parse = canonical_parse + + +def canonical_compose(stream): + return ruyaml.compose(stream, Loader=CanonicalLoader) + + +ruyaml.canonical_compose = canonical_compose + + +def canonical_compose_all(stream): + return ruyaml.compose_all(stream, Loader=CanonicalLoader) + + +ruyaml.canonical_compose_all = canonical_compose_all + + +def canonical_load(stream): + return ruyaml.load(stream, Loader=CanonicalLoader) + + +ruyaml.canonical_load = canonical_load + + +def canonical_load_all(stream): + yaml = ruyaml.YAML(typ='safe', pure=True) + yaml.Loader = CanonicalLoader + return yaml.load_all(stream) + + +ruyaml.canonical_load_all = canonical_load_all |