1 files changed, 358 insertions, 0 deletions
diff --git a/tests/lib3/canonical.py b/tests/lib3/canonical.py
new file mode 100644
index 0000000..e04477b
--- /dev/null
+++ b/tests/lib3/canonical.py
@@ -0,0 +1,358 @@
+
+import yaml, yaml.composer, yaml.constructor, yaml.resolver
+
+class CanonicalError(yaml.YAMLError):
+    pass
+
+class CanonicalScanner:
+
+    def __init__(self, data):
+        if isinstance(data, bytes):
+            try:
+                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(yaml.StreamStartToken(None, None))
+        while True:
+            self.find_token()
+            ch = self.data[self.index]
+            if ch == '\0':
+                self.tokens.append(yaml.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(yaml.DocumentStartToken(None, None))
+            elif ch == '[':
+                self.index += 1
+                self.tokens.append(yaml.FlowSequenceStartToken(None, None))
+            elif ch == '{':
+                self.index += 1
+                self.tokens.append(yaml.FlowMappingStartToken(None, None))
+            elif ch == ']':
+                self.index += 1
+                self.tokens.append(yaml.FlowSequenceEndToken(None, None))
+            elif ch == '}':
+                self.index += 1
+                self.tokens.append(yaml.FlowMappingEndToken(None, None))
+            elif ch == '?':
+                self.index += 1
+                self.tokens.append(yaml.KeyToken(None, None))
+            elif ch == ':':
+                self.index += 1
+                self.tokens.append(yaml.ValueToken(None, None))
+            elif ch == ',':
+                self.index += 1
+                self.tokens.append(yaml.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 yaml.DirectiveToken('YAML', (1, 1), None, None)
+        else:
+            raise CanonicalError("invalid directive")
+
+    def scan_alias(self):
+        if self.data[self.index] == '*':
+            TokenClass = yaml.AliasToken
+        else:
+            TokenClass = yaml.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 yaml.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 yaml.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
+
+    # stream: STREAM-START document* STREAM-END
+    def parse_stream(self):
+        self.get_token(yaml.StreamStartToken)
+        self.events.append(yaml.StreamStartEvent(None, None))
+        while not self.check_token(yaml.StreamEndToken):
+            if self.check_token(yaml.DirectiveToken, yaml.DocumentStartToken):
+                self.parse_document()
+            else:
+                raise CanonicalError("document is expected, got "+repr(self.tokens[0]))
+        self.get_token(yaml.StreamEndToken)
+        self.events.append(yaml.StreamEndEvent(None, None))
+
+    # document: DIRECTIVE? DOCUMENT-START node
+    def parse_document(self):
+        node = None
+        if self.check_token(yaml.DirectiveToken):
+            self.get_token(yaml.DirectiveToken)
+        self.get_token(yaml.DocumentStartToken)
+        self.events.append(yaml.DocumentStartEvent(None, None))
+        self.parse_node()
+        self.events.append(yaml.DocumentEndEvent(None, None))
+
+    # node: ALIAS | ANCHOR? TAG? (SCALAR|sequence|mapping)
+    def parse_node(self):
+        if self.check_token(yaml.AliasToken):
+            self.events.append(yaml.AliasEvent(self.get_token_value(), None, None))
+        else:
+            anchor = None
+            if self.check_token(yaml.AnchorToken):
+                anchor = self.get_token_value()
+            tag = None
+            if self.check_token(yaml.TagToken):
+                tag = self.get_token_value()
+            if self.check_token(yaml.ScalarToken):
+                self.events.append(yaml.ScalarEvent(anchor, tag, (False, False), self.get_token_value(), None, None))
+            elif self.check_token(yaml.FlowSequenceStartToken):
+                self.events.append(yaml.SequenceStartEvent(anchor, tag, None, None))
+                self.parse_sequence()
+            elif self.check_token(yaml.FlowMappingStartToken):
+                self.events.append(yaml.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(yaml.FlowSequenceStartToken)
+        if not self.check_token(yaml.FlowSequenceEndToken):
+            self.parse_node()
+            while not self.check_token(yaml.FlowSequenceEndToken):
+                self.get_token(yaml.FlowEntryToken)
+                if not self.check_token(yaml.FlowSequenceEndToken):
+                    self.parse_node()
+        self.get_token(yaml.FlowSequenceEndToken)
+        self.events.append(yaml.SequenceEndEvent(None, None))
+
+    # mapping: MAPPING-START (map_entry (ENTRY map_entry)*)? ENTRY? MAPPING-END
+    def parse_mapping(self):
+        self.get_token(yaml.FlowMappingStartToken)
+        if not self.check_token(yaml.FlowMappingEndToken):
+            self.parse_map_entry()
+            while not self.check_token(yaml.FlowMappingEndToken):
+                self.get_token(yaml.FlowEntryToken)
+                if not self.check_token(yaml.FlowMappingEndToken):
+                    self.parse_map_entry()
+        self.get_token(yaml.FlowMappingEndToken)
+        self.events.append(yaml.MappingEndEvent(None, None))
+
+    # map_entry: KEY node VALUE node
+    def parse_map_entry(self):
+        self.get_token(yaml.KeyToken)
+        self.parse_node()
+        self.get_token(yaml.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,
+        yaml.composer.Composer, yaml.constructor.Constructor, yaml.resolver.Resolver):
+
+    def __init__(self, stream):
+        if hasattr(stream, 'read'):
+            stream = stream.read()
+        CanonicalScanner.__init__(self, stream)
+        CanonicalParser.__init__(self)
+        yaml.composer.Composer.__init__(self)
+        yaml.constructor.Constructor.__init__(self)
+        yaml.resolver.Resolver.__init__(self)
+
+yaml.CanonicalLoader = CanonicalLoader
+
+def canonical_scan(stream):
+    return yaml.scan(stream, Loader=CanonicalLoader)
+
+yaml.canonical_scan = canonical_scan
+
+def canonical_parse(stream):
+    return yaml.parse(stream, Loader=CanonicalLoader)
+
+yaml.canonical_parse = canonical_parse
+
+def canonical_compose(stream):
+    return yaml.compose(stream, Loader=CanonicalLoader)
+
+yaml.canonical_compose = canonical_compose
+
+def canonical_compose_all(stream):
+    return yaml.compose_all(stream, Loader=CanonicalLoader)
+
+yaml.canonical_compose_all = canonical_compose_all
+
+def canonical_load(stream):
+    return yaml.load(stream, Loader=CanonicalLoader)
+
+yaml.canonical_load = canonical_load
+
+def canonical_load_all(stream):
+    return yaml.load_all(stream, Loader=CanonicalLoader)
+
+yaml.canonical_load_all = canonical_load_all
+