blob: 6d6f59d0d059129301aadbabd3d32023fbf779e7 [file] [log] [blame]
import yaml, canonical
import pprint
def _convert_structure(loader):
if loader.check_event(yaml.ScalarEvent):
event = loader.get_event()
if event.tag or event.anchor or event.value:
return True
else:
return None
elif loader.check_event(yaml.SequenceStartEvent):
loader.get_event()
sequence = []
while not loader.check_event(yaml.SequenceEndEvent):
sequence.append(_convert_structure(loader))
loader.get_event()
return sequence
elif loader.check_event(yaml.MappingStartEvent):
loader.get_event()
mapping = []
while not loader.check_event(yaml.MappingEndEvent):
key = _convert_structure(loader)
value = _convert_structure(loader)
mapping.append((key, value))
loader.get_event()
return mapping
elif loader.check_event(yaml.AliasEvent):
loader.get_event()
return '*'
else:
loader.get_event()
return '?'
def test_structure(data_filename, structure_filename, verbose=False):
nodes1 = []
nodes2 = eval(open(structure_filename, 'r').read())
try:
loader = yaml.Loader(open(data_filename, 'rb'))
while loader.check_event():
if loader.check_event(yaml.StreamStartEvent, yaml.StreamEndEvent,
yaml.DocumentStartEvent, yaml.DocumentEndEvent):
loader.get_event()
continue
nodes1.append(_convert_structure(loader))
if len(nodes1) == 1:
nodes1 = nodes1[0]
assert nodes1 == nodes2, (nodes1, nodes2)
finally:
if verbose:
print("NODES1:")
pprint.pprint(nodes1)
print("NODES2:")
pprint.pprint(nodes2)
test_structure.unittest = ['.data', '.structure']
def _compare_events(events1, events2, full=False):
assert len(events1) == len(events2), (len(events1), len(events2))
for event1, event2 in zip(events1, events2):
assert event1.__class__ == event2.__class__, (event1, event2)
if isinstance(event1, yaml.AliasEvent) and full:
assert event1.anchor == event2.anchor, (event1, event2)
if isinstance(event1, (yaml.ScalarEvent, yaml.CollectionStartEvent)):
if (event1.tag not in [None, '!'] and event2.tag not in [None, '!']) or full:
assert event1.tag == event2.tag, (event1, event2)
if isinstance(event1, yaml.ScalarEvent):
assert event1.value == event2.value, (event1, event2)
def test_parser(data_filename, canonical_filename, verbose=False):
events1 = None
events2 = None
try:
events1 = list(yaml.parse(open(data_filename, 'rb')))
events2 = list(yaml.canonical_parse(open(canonical_filename, 'rb')))
_compare_events(events1, events2)
finally:
if verbose:
print("EVENTS1:")
pprint.pprint(events1)
print("EVENTS2:")
pprint.pprint(events2)
test_parser.unittest = ['.data', '.canonical']
def test_parser_on_canonical(canonical_filename, verbose=False):
events1 = None
events2 = None
try:
events1 = list(yaml.parse(open(canonical_filename, 'rb')))
events2 = list(yaml.canonical_parse(open(canonical_filename, 'rb')))
_compare_events(events1, events2, full=True)
finally:
if verbose:
print("EVENTS1:")
pprint.pprint(events1)
print("EVENTS2:")
pprint.pprint(events2)
test_parser_on_canonical.unittest = ['.canonical']
def _compare_nodes(node1, node2):
assert node1.__class__ == node2.__class__, (node1, node2)
assert node1.tag == node2.tag, (node1, node2)
if isinstance(node1, yaml.ScalarNode):
assert node1.value == node2.value, (node1, node2)
else:
assert len(node1.value) == len(node2.value), (node1, node2)
for item1, item2 in zip(node1.value, node2.value):
if not isinstance(item1, tuple):
item1 = (item1,)
item2 = (item2,)
for subnode1, subnode2 in zip(item1, item2):
_compare_nodes(subnode1, subnode2)
def test_composer(data_filename, canonical_filename, verbose=False):
nodes1 = None
nodes2 = None
try:
nodes1 = list(yaml.compose_all(open(data_filename, 'rb')))
nodes2 = list(yaml.canonical_compose_all(open(canonical_filename, 'rb')))
assert len(nodes1) == len(nodes2), (len(nodes1), len(nodes2))
for node1, node2 in zip(nodes1, nodes2):
_compare_nodes(node1, node2)
finally:
if verbose:
print("NODES1:")
pprint.pprint(nodes1)
print("NODES2:")
pprint.pprint(nodes2)
test_composer.unittest = ['.data', '.canonical']
def _make_loader():
global MyLoader
class MyLoader(yaml.Loader):
def construct_sequence(self, node):
return tuple(yaml.Loader.construct_sequence(self, node))
def construct_mapping(self, node):
pairs = self.construct_pairs(node)
pairs.sort(key=(lambda i: str(i)))
return pairs
def construct_undefined(self, node):
return self.construct_scalar(node)
MyLoader.add_constructor('tag:yaml.org,2002:map', MyLoader.construct_mapping)
MyLoader.add_constructor(None, MyLoader.construct_undefined)
def _make_canonical_loader():
global MyCanonicalLoader
class MyCanonicalLoader(yaml.CanonicalLoader):
def construct_sequence(self, node):
return tuple(yaml.CanonicalLoader.construct_sequence(self, node))
def construct_mapping(self, node):
pairs = self.construct_pairs(node)
pairs.sort(key=(lambda i: str(i)))
return pairs
def construct_undefined(self, node):
return self.construct_scalar(node)
MyCanonicalLoader.add_constructor('tag:yaml.org,2002:map', MyCanonicalLoader.construct_mapping)
MyCanonicalLoader.add_constructor(None, MyCanonicalLoader.construct_undefined)
def test_constructor(data_filename, canonical_filename, verbose=False):
_make_loader()
_make_canonical_loader()
native1 = None
native2 = None
try:
native1 = list(yaml.load_all(open(data_filename, 'rb'), Loader=MyLoader))
native2 = list(yaml.load_all(open(canonical_filename, 'rb'), Loader=MyCanonicalLoader))
assert native1 == native2, (native1, native2)
finally:
if verbose:
print("NATIVE1:")
pprint.pprint(native1)
print("NATIVE2:")
pprint.pprint(native2)
test_constructor.unittest = ['.data', '.canonical']
if __name__ == '__main__':
import test_appliance
test_appliance.run(globals())