tests/lib3/test_yaml_ext.py

   1
   2 import _yaml, yaml
   3 import types, pprint
   4
   5 yaml.PyBaseLoader = yaml.BaseLoader
   6 yaml.PySafeLoader = yaml.SafeLoader
   7 yaml.PyLoader = yaml.Loader
   8 yaml.PyBaseDumper = yaml.BaseDumper
   9 yaml.PySafeDumper = yaml.SafeDumper
  10 yaml.PyDumper = yaml.Dumper
  11
  12 old_scan = yaml.scan
  13 def new_scan(stream, Loader=yaml.CLoader):
  14     return old_scan(stream, Loader)
  15
  16 old_parse = yaml.parse
  17 def new_parse(stream, Loader=yaml.CLoader):
  18     return old_parse(stream, Loader)
  19
  20 old_compose = yaml.compose
  21 def new_compose(stream, Loader=yaml.CLoader):
  22     return old_compose(stream, Loader)
  23
  24 old_compose_all = yaml.compose_all
  25 def new_compose_all(stream, Loader=yaml.CLoader):
  26     return old_compose_all(stream, Loader)
  27
  28 old_load = yaml.load
  29 def new_load(stream, Loader=yaml.CLoader):
  30     return old_load(stream, Loader)
  31
  32 old_load_all = yaml.load_all
  33 def new_load_all(stream, Loader=yaml.CLoader):
  34     return old_load_all(stream, Loader)
  35
  36 old_safe_load = yaml.safe_load
  37 def new_safe_load(stream):
  38     return old_load(stream, yaml.CSafeLoader)
  39
  40 old_safe_load_all = yaml.safe_load_all
  41 def new_safe_load_all(stream):
  42     return old_load_all(stream, yaml.CSafeLoader)
  43
  44 old_emit = yaml.emit
  45 def new_emit(events, stream=None, Dumper=yaml.CDumper, **kwds):
  46     return old_emit(events, stream, Dumper, **kwds)
  47
  48 old_serialize = yaml.serialize
  49 def new_serialize(node, stream, Dumper=yaml.CDumper, **kwds):
  50     return old_serialize(node, stream, Dumper, **kwds)
  51
  52 old_serialize_all = yaml.serialize_all
  53 def new_serialize_all(nodes, stream=None, Dumper=yaml.CDumper, **kwds):
  54     return old_serialize_all(nodes, stream, Dumper, **kwds)
  55
  56 old_dump = yaml.dump
  57 def new_dump(data, stream=None, Dumper=yaml.CDumper, **kwds):
  58     return old_dump(data, stream, Dumper, **kwds)
  59
  60 old_dump_all = yaml.dump_all
  61 def new_dump_all(documents, stream=None, Dumper=yaml.CDumper, **kwds):
  62     return old_dump_all(documents, stream, Dumper, **kwds)
  63
  64 old_safe_dump = yaml.safe_dump
  65 def new_safe_dump(data, stream=None, **kwds):
  66     return old_dump(data, stream, yaml.CSafeDumper, **kwds)
  67
  68 old_safe_dump_all = yaml.safe_dump_all
  69 def new_safe_dump_all(documents, stream=None, **kwds):
  70     return old_dump_all(documents, stream, yaml.CSafeDumper, **kwds)
  71
  72 def _set_up():
  73     yaml.BaseLoader = yaml.CBaseLoader
  74     yaml.SafeLoader = yaml.CSafeLoader
  75     yaml.Loader = yaml.CLoader
  76     yaml.BaseDumper = yaml.CBaseDumper
  77     yaml.SafeDumper = yaml.CSafeDumper
  78     yaml.Dumper = yaml.CDumper
  79     yaml.scan = new_scan
  80     yaml.parse = new_parse
  81     yaml.compose = new_compose
  82     yaml.compose_all = new_compose_all
  83     yaml.load = new_load
  84     yaml.load_all = new_load_all
  85     yaml.safe_load = new_safe_load
  86     yaml.safe_load_all = new_safe_load_all
  87     yaml.emit = new_emit
  88     yaml.serialize = new_serialize
  89     yaml.serialize_all = new_serialize_all
  90     yaml.dump = new_dump
  91     yaml.dump_all = new_dump_all
  92     yaml.safe_dump = new_safe_dump
  93     yaml.safe_dump_all = new_safe_dump_all
  94
  95 def _tear_down():
  96     yaml.BaseLoader = yaml.PyBaseLoader
  97     yaml.SafeLoader = yaml.PySafeLoader
  98     yaml.Loader = yaml.PyLoader
  99     yaml.BaseDumper = yaml.PyBaseDumper
 100     yaml.SafeDumper = yaml.PySafeDumper
 101     yaml.Dumper = yaml.PyDumper
 102     yaml.scan = old_scan
 103     yaml.parse = old_parse
 104     yaml.compose = old_compose
 105     yaml.compose_all = old_compose_all
 106     yaml.load = old_load
 107     yaml.load_all = old_load_all
 108     yaml.safe_load = old_safe_load
 109     yaml.safe_load_all = old_safe_load_all
 110     yaml.emit = old_emit
 111     yaml.serialize = old_serialize
 112     yaml.serialize_all = old_serialize_all
 113     yaml.dump = old_dump
 114     yaml.dump_all = old_dump_all
 115     yaml.safe_dump = old_safe_dump
 116     yaml.safe_dump_all = old_safe_dump_all
 117
 118 def test_c_version(verbose=False):
 119     if verbose:
 120         print(_yaml.get_version())
 121         print(_yaml.get_version_string())
 122     assert ("%s.%s.%s" % _yaml.get_version()) == _yaml.get_version_string(),    \
 123             (_yaml.get_version(), _yaml.get_version_string())
 124
 125 def _compare_scanners(py_data, c_data, verbose):
 126     py_tokens = list(yaml.scan(py_data, Loader=yaml.PyLoader))
 127     c_tokens = []
 128     try:
 129         for token in yaml.scan(c_data, Loader=yaml.CLoader):
 130             c_tokens.append(token)
 131         assert len(py_tokens) == len(c_tokens), (len(py_tokens), len(c_tokens))
 132         for py_token, c_token in zip(py_tokens, c_tokens):
 133             assert py_token.__class__ == c_token.__class__, (py_token, c_token)
 134             if hasattr(py_token, 'value'):
 135                 assert py_token.value == c_token.value, (py_token, c_token)
 136             if isinstance(py_token, yaml.StreamEndToken):
 137                 continue
 138             py_start = (py_token.start_mark.index, py_token.start_mark.line, py_token.start_mark.column)
 139             py_end = (py_token.end_mark.index, py_token.end_mark.line, py_token.end_mark.column)
 140             c_start = (c_token.start_mark.index, c_token.start_mark.line, c_token.start_mark.column)
 141             c_end = (c_token.end_mark.index, c_token.end_mark.line, c_token.end_mark.column)
 142             assert py_start == c_start, (py_start, c_start)
 143             assert py_end == c_end, (py_end, c_end)
 144     finally:
 145         if verbose:
 146             print("PY_TOKENS:")
 147             pprint.pprint(py_tokens)
 148             print("C_TOKENS:")
 149             pprint.pprint(c_tokens)
 150
 151 def test_c_scanner(data_filename, canonical_filename, verbose=False):
 152     _compare_scanners(open(data_filename, 'rb'),
 153             open(data_filename, 'rb'), verbose)
 154     _compare_scanners(open(data_filename, 'rb').read(),
 155             open(data_filename, 'rb').read(), verbose)
 156     _compare_scanners(open(canonical_filename, 'rb'),
 157             open(canonical_filename, 'rb'), verbose)
 158     _compare_scanners(open(canonical_filename, 'rb').read(),
 159             open(canonical_filename, 'rb').read(), verbose)
 160
 161 test_c_scanner.unittest = ['.data', '.canonical']
 162 test_c_scanner.skip = ['.skip-ext']
 163
 164 def _compare_parsers(py_data, c_data, verbose):
 165     py_events = list(yaml.parse(py_data, Loader=yaml.PyLoader))
 166     c_events = []
 167     try:
 168         for event in yaml.parse(c_data, Loader=yaml.CLoader):
 169             c_events.append(event)
 170         assert len(py_events) == len(c_events), (len(py_events), len(c_events))
 171         for py_event, c_event in zip(py_events, c_events):
 172             for attribute in ['__class__', 'anchor', 'tag', 'implicit',
 173                                 'value', 'explicit', 'version', 'tags']:
 174                 py_value = getattr(py_event, attribute, None)
 175                 c_value = getattr(c_event, attribute, None)
 176                 assert py_value == c_value, (py_event, c_event, attribute)
 177     finally:
 178         if verbose:
 179             print("PY_EVENTS:")
 180             pprint.pprint(py_events)
 181             print("C_EVENTS:")
 182             pprint.pprint(c_events)
 183
 184 def test_c_parser(data_filename, canonical_filename, verbose=False):
 185     _compare_parsers(open(data_filename, 'rb'),
 186             open(data_filename, 'rb'), verbose)
 187     _compare_parsers(open(data_filename, 'rb').read(),
 188             open(data_filename, 'rb').read(), verbose)
 189     _compare_parsers(open(canonical_filename, 'rb'),
 190             open(canonical_filename, 'rb'), verbose)
 191     _compare_parsers(open(canonical_filename, 'rb').read(),
 192             open(canonical_filename, 'rb').read(), verbose)
 193
 194 test_c_parser.unittest = ['.data', '.canonical']
 195 test_c_parser.skip = ['.skip-ext']
 196
 197 def _compare_emitters(data, verbose):
 198     events = list(yaml.parse(data, Loader=yaml.PyLoader))
 199     c_data = yaml.emit(events, Dumper=yaml.CDumper)
 200     if verbose:
 201         print(c_data)
 202     py_events = list(yaml.parse(c_data, Loader=yaml.PyLoader))
 203     c_events = list(yaml.parse(c_data, Loader=yaml.CLoader))
 204     try:
 205         assert len(events) == len(py_events), (len(events), len(py_events))
 206         assert len(events) == len(c_events), (len(events), len(c_events))
 207         for event, py_event, c_event in zip(events, py_events, c_events):
 208             for attribute in ['__class__', 'anchor', 'tag', 'implicit',
 209                                 'value', 'explicit', 'version', 'tags']:
 210                 value = getattr(event, attribute, None)
 211                 py_value = getattr(py_event, attribute, None)
 212                 c_value = getattr(c_event, attribute, None)
 213                 if attribute == 'tag' and value in [None, '!'] \
 214                         and py_value in [None, '!'] and c_value in [None, '!']:
 215                     continue
 216                 if attribute == 'explicit' and (py_value or c_value):
 217                     continue
 218                 assert value == py_value, (event, py_event, attribute)
 219                 assert value == c_value, (event, c_event, attribute)
 220     finally:
 221         if verbose:
 222             print("EVENTS:")
 223             pprint.pprint(events)
 224             print("PY_EVENTS:")
 225             pprint.pprint(py_events)
 226             print("C_EVENTS:")
 227             pprint.pprint(c_events)
 228
 229 def test_c_emitter(data_filename, canonical_filename, verbose=False):
 230     _compare_emitters(open(data_filename, 'rb').read(), verbose)
 231     _compare_emitters(open(canonical_filename, 'rb').read(), verbose)
 232
 233 test_c_emitter.unittest = ['.data', '.canonical']
 234 test_c_emitter.skip = ['.skip-ext']
 235
 236 def wrap_ext_function(function):
 237     def wrapper(*args, **kwds):
 238         _set_up()
 239         try:
 240             function(*args, **kwds)
 241         finally:
 242             _tear_down()
 243     wrapper.__name__ = '%s_ext' % function.__name__
 244     wrapper.unittest = function.unittest
 245     wrapper.skip = getattr(function, 'skip', [])+['.skip-ext']
 246     return wrapper
 247
 248 def wrap_ext(collections):
 249     functions = []
 250     if not isinstance(collections, list):
 251         collections = [collections]
 252     for collection in collections:
 253         if not isinstance(collection, dict):
 254             collection = vars(collection)
 255         for key in sorted(collection):
 256             value = collection[key]
 257             if isinstance(value, types.FunctionType) and hasattr(value, 'unittest'):
 258                 functions.append(wrap_ext_function(value))
 259     for function in functions:
 260         assert function.__name__ not in globals()
 261         globals()[function.__name__] = function
 262
 263 import test_tokens, test_structure, test_errors, test_resolver, test_constructor,   \
 264         test_emitter, test_representer, test_recursive, test_input_output
 265 wrap_ext([test_tokens, test_structure, test_errors, test_resolver, test_constructor,
 266         test_emitter, test_representer, test_recursive, test_input_output])
 267
 268 if __name__ == '__main__':
 269     import test_appliance
 270     test_appliance.run(globals())
 271