refactor: apply Ruff, mypy fixes; reformat

This commit is contained in:
D. Bohdan 2023-07-08 17:23:06 +00:00
parent 6d6085b2de
commit 434d8b2bd5
3 changed files with 364 additions and 401 deletions

View File

@ -6,42 +6,38 @@
from __future__ import print_function
import argparse
import cbor2
import cbor2 # type: ignore
import datetime
import dateutil.parser
import io
import json
import os.path
import re
import string
import sys
import tomlkit
import umsgpack
import umsgpack # type: ignore
import yaml
__version__ = '0.14.0'
__version__ = "0.14.0"
FORMATS = ['cbor', 'json', 'msgpack', 'toml', 'yaml']
FORMATS = ["cbor", "json", "msgpack", "toml", "yaml"]
# === YAML ===
# An ordered dumper for PyYAML.
class OrderedDumper(yaml.SafeDumper):
pass
def mapping_representer(dumper, data):
return dumper.represent_mapping(
yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
data.items()
yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, data.items()
)
OrderedDumper.add_representer(
dict,
mapping_representer
)
OrderedDumper.add_representer(dict, mapping_representer)
# Fix loss of time zone information in PyYAML.
@ -56,19 +52,11 @@ def timestamp_constructor(loader, node):
loaders = [TimezoneLoader]
for loader in loaders:
loader.add_constructor(
u'tag:yaml.org,2002:timestamp',
timestamp_constructor
)
loader.add_constructor("tag:yaml.org,2002:timestamp", timestamp_constructor)
# === JSON ===
if hasattr(json, 'JSONDecodeError'):
JSONDecodeError = json.JSONDecodeError
else:
JSONDecodeError = ValueError
def json_default(obj):
if isinstance(obj, datetime.datetime):
@ -78,9 +66,10 @@ def json_default(obj):
# === CLI ===
def argv0_to_format(argv0):
possible_format = '(' + '|'.join(FORMATS) + ')'
match = re.search('^' + possible_format + '2' + possible_format, argv0)
possible_format = "(" + "|".join(FORMATS) + ")"
match = re.search("^" + possible_format + "2" + possible_format, argv0)
if match:
from_, to = match.groups()
return True, from_, to
@ -93,8 +82,8 @@ def extension_to_format(path):
ext = ext[1:]
if ext == 'yml':
ext = 'yaml'
if ext == "yml":
ext = "yaml"
return ext if ext in FORMATS else None
@ -104,99 +93,90 @@ def parse_command_line(argv):
format_from_argv0, argv0_from, argv0_to = argv0_to_format(me)
parser = argparse.ArgumentParser(
description='Convert between CBOR, JSON, MessagePack, TOML, and YAML.'
description="Convert between CBOR, JSON, MessagePack, TOML, and YAML."
)
input_group = parser.add_mutually_exclusive_group()
input_group.add_argument("input", nargs="?", default="-", help="input file")
input_group.add_argument(
'input',
nargs='?',
default='-',
help='input file'
)
input_group.add_argument(
'-i', '--input',
dest='input_flag',
metavar='input',
"-i",
"--input",
dest="input_flag",
metavar="input",
default=None,
help='input file'
help="input file",
)
output_group = parser.add_mutually_exclusive_group()
output_group.add_argument("output", nargs="?", default="-", help="output file")
output_group.add_argument(
'output',
nargs='?',
default='-',
help='output file'
)
output_group.add_argument(
'-o', '--output',
dest='output_flag',
metavar='output',
"-o",
"--output",
dest="output_flag",
metavar="output",
default=None,
help='output file'
help="output file",
)
if not format_from_argv0:
parser.add_argument(
'--if', '-if', '--input-format',
dest='input_format',
"--if",
"-if",
"--input-format",
dest="input_format",
help="input format",
choices=FORMATS
choices=FORMATS,
)
parser.add_argument(
'--of',
'-of',
'--output-format',
dest='output_format',
"--of",
"-of",
"--output-format",
dest="output_format",
help="output format",
choices=FORMATS
choices=FORMATS,
)
if not format_from_argv0 or argv0_to == 'json':
if not format_from_argv0 or argv0_to == "json":
parser.add_argument(
'--indent-json',
dest='indent_json',
metavar='n',
"--indent-json",
dest="indent_json",
metavar="n",
type=int,
default=None,
help='indent JSON output'
help="indent JSON output",
)
if not format_from_argv0 or argv0_to == 'yaml':
if not format_from_argv0 or argv0_to == "yaml":
parser.add_argument(
'--yaml-style',
dest='yaml_style',
"--yaml-style",
dest="yaml_style",
default=None,
help='YAML formatting style',
choices=['', '\'', '"', '|', '>']
help="YAML formatting style",
choices=["", "'", '"', "|", ">"],
)
parser.add_argument(
'--wrap',
dest='wrap',
metavar='key',
"--wrap",
dest="wrap",
metavar="key",
default=None,
help='wrap the data in a map type with the given key'
help="wrap the data in a map type with the given key",
)
parser.add_argument(
'--unwrap',
dest='unwrap',
metavar='key',
"--unwrap",
dest="unwrap",
metavar="key",
default=None,
help='only output the data stored under the given key'
help="only output the data stored under the given key",
)
parser.add_argument(
'-p', '--preserve-key-order',
dest='ordered',
action='store_true',
help='preserve the order of dictionary/mapping keys when encoding'
)
parser.add_argument(
'-v', '--version',
action='version',
version=__version__
"-p",
"--preserve-key-order",
dest="ordered",
action="store_true",
help="preserve the order of dictionary/mapping keys when encoding",
)
parser.add_argument("-v", "--version", action="version", version=__version__)
args = parser.parse_args(args=argv[1:])
@ -212,30 +192,31 @@ def parse_command_line(argv):
args.input_format = argv0_from
args.output_format = argv0_to
if argv0_to != 'json':
args.__dict__['indent_json'] = None
if argv0_to != 'yaml':
args.__dict__['yaml_style'] = None
if argv0_to != "json":
args.__dict__["indent_json"] = None
if argv0_to != "yaml":
args.__dict__["yaml_style"] = None
else:
if args.input_format is None:
args.input_format = extension_to_format(args.input)
if args.input_format is None:
parser.error('Need an explicit input format')
parser.error("Need an explicit input format")
if args.output_format is None:
args.output_format = extension_to_format(args.output)
if args.output_format is None:
parser.error('Need an explicit output format')
parser.error("Need an explicit output format")
# Wrap yaml_style.
args.__dict__['yaml_options'] = {'default_style': args.yaml_style}
del args.__dict__['yaml_style']
args.__dict__["yaml_options"] = {"default_style": args.yaml_style}
del args.__dict__["yaml_style"]
return args
# === Parser/serializer wrappers ===
def traverse(
col,
dict_callback=lambda x: dict(x),
@ -245,28 +226,39 @@ def traverse(
default_callback=lambda x: x,
):
if isinstance(col, dict):
res = dict_callback([
(key_callback(k), traverse(
v,
dict_callback,
list_callback,
key_callback,
instance_callbacks,
default_callback,
)) for (k, v) in col.items()
])
res = dict_callback(
[
(
key_callback(k),
traverse(
v,
dict_callback,
list_callback,
key_callback,
instance_callbacks,
default_callback,
),
)
for (k, v) in col.items()
]
)
elif isinstance(col, list):
res = list_callback([traverse(
x,
dict_callback,
list_callback,
key_callback,
instance_callbacks,
default_callback,
) for x in col])
res = list_callback(
[
traverse(
x,
dict_callback,
list_callback,
key_callback,
instance_callbacks,
default_callback,
)
for x in col
]
)
else:
matched = False
for (t, callback) in instance_callbacks:
for t, callback in instance_callbacks:
if isinstance(col, t):
matched = True
res = callback(col)
@ -281,24 +273,24 @@ def traverse(
def decode_json(input_data):
try:
return json.loads(
input_data.decode('utf-8'),
input_data.decode("utf-8"),
)
except JSONDecodeError as e:
raise ValueError('Cannot parse as JSON ({0})'.format(e))
except json.JSONDecodeError as e:
raise ValueError("Cannot parse as JSON ({0})".format(e))
def decode_msgpack(input_data):
try:
return umsgpack.unpackb(input_data)
except umsgpack.UnpackException as e:
raise ValueError('Cannot parse as MessagePack ({0})'.format(e))
raise ValueError("Cannot parse as MessagePack ({0})".format(e))
def decode_cbor(input_data):
try:
return cbor2.loads(input_data)
except cbor2.CBORDecodeError as e:
raise ValueError('Cannot parse as CBOR ({0})'.format(e))
raise ValueError("Cannot parse as CBOR ({0})".format(e))
def decode_toml(input_data):
@ -309,59 +301,65 @@ def decode_toml(input_data):
tomlkit.loads(input_data),
instance_callbacks={
(tomlkit.items.Bool, bool),
(tomlkit.items.Date, lambda x: datetime.date(
x.year,
x.month,
x.day,
)),
(tomlkit.items.DateTime, lambda x: datetime.datetime(
x.year,
x.month,
x.day,
x.hour,
x.minute,
x.second,
x.microsecond,
x.tzinfo,
)),
(
tomlkit.items.Date,
lambda x: datetime.date(
x.year,
x.month,
x.day,
),
),
(
tomlkit.items.DateTime,
lambda x: datetime.datetime(
x.year,
x.month,
x.day,
x.hour,
x.minute,
x.second,
x.microsecond,
x.tzinfo,
),
),
(tomlkit.items.Float, float),
(tomlkit.items.Integer, int),
(tomlkit.items.String, str),
(tomlkit.items.Time, lambda x: datetime.time(
x.hour,
x.minute,
x.second,
x.microsecond,
x.tzinfo,
)),
}
(
tomlkit.items.Time,
lambda x: datetime.time(
x.hour,
x.minute,
x.second,
x.microsecond,
x.tzinfo,
),
),
},
)
except tomlkit.exceptions.ParseError as e:
raise ValueError('Cannot parse as TOML ({0})'.format(e))
raise ValueError("Cannot parse as TOML ({0})".format(e))
def decode_yaml(input_data):
try:
loader = TimezoneLoader
return yaml.load(
input_data,
loader
)
return yaml.load(input_data, loader)
except (yaml.scanner.ScannerError, yaml.parser.ParserError) as e:
raise ValueError('Cannot parse as YAML ({0})'.format(e))
raise ValueError("Cannot parse as YAML ({0})".format(e))
def decode(input_format, input_data):
decoder = {
'cbor': decode_cbor,
'json': decode_json,
'msgpack': decode_msgpack,
'toml': decode_toml,
'yaml': decode_yaml,
"cbor": decode_cbor,
"json": decode_json,
"msgpack": decode_msgpack,
"toml": decode_toml,
"yaml": decode_yaml,
}
if input_format not in decoder:
raise ValueError('Unknown input format: {0}'.format(input_format))
raise ValueError("Unknown input format: {0}".format(input_format))
return decoder[input_format](input_data)
@ -371,9 +369,9 @@ def encode_json(data, ordered, indent):
indent = 2
if indent:
separators = (',', ': ')
separators = (",", ": ")
else:
separators = (',', ':')
separators = (",", ":")
def stringify_key(key):
if isinstance(key, bool):
@ -384,33 +382,36 @@ def encode_json(data, ordered, indent):
return key
try:
return json.dumps(
traverse(
data,
key_callback=stringify_key,
),
default=json_default,
ensure_ascii=False,
indent=indent,
separators=separators,
sort_keys=not ordered
) + "\n"
return (
json.dumps(
traverse(
data,
key_callback=stringify_key,
),
default=json_default,
ensure_ascii=False,
indent=indent,
separators=separators,
sort_keys=not ordered,
)
+ "\n"
)
except TypeError as e:
raise ValueError('Cannot convert data to JSON ({0})'.format(e))
raise ValueError("Cannot convert data to JSON ({0})".format(e))
def encode_msgpack(data):
try:
return umsgpack.packb(data)
except umsgpack.UnsupportedTypeException as e:
raise ValueError('Cannot convert data to MessagePack ({0})'.format(e))
raise ValueError("Cannot convert data to MessagePack ({0})".format(e))
def encode_cbor(data):
try:
return cbor2.dumps(data)
except cbor2.CBOREncodeError as e:
raise ValueError('Cannot convert data to CBOR ({0})'.format(e))
raise ValueError("Cannot convert data to CBOR ({0})".format(e))
def encode_toml(data, ordered):
@ -419,14 +420,14 @@ def encode_toml(data, ordered):
except AttributeError as e:
if str(e) == "'list' object has no attribute 'as_string'":
raise ValueError(
'Cannot convert non-dictionary data to '
"Cannot convert non-dictionary data to "
'TOML; use "wrap" to wrap it in a '
'dictionary'
"dictionary"
)
else:
raise e
except (TypeError, ValueError) as e:
raise ValueError('Cannot convert data to TOML ({0})'.format(e))
raise ValueError("Cannot convert data to TOML ({0})".format(e))
def encode_yaml(data, ordered, yaml_options):
@ -442,11 +443,12 @@ def encode_yaml(data, ordered, yaml_options):
**yaml_options
)
except yaml.representer.RepresenterError as e:
raise ValueError('Cannot convert data to YAML ({0})'.format(e))
raise ValueError("Cannot convert data to YAML ({0})".format(e))
# === Main ===
def run(argv):
args = parse_command_line(argv)
remarshal(
@ -458,7 +460,7 @@ def run(argv):
args.unwrap,
args.indent_json,
args.yaml_options,
args.ordered
args.ordered,
)
@ -475,15 +477,15 @@ def remarshal(
transform=None,
):
try:
if input == '-':
input_file = getattr(sys.stdin, 'buffer', sys.stdin)
if input == "-":
input_file = getattr(sys.stdin, "buffer", sys.stdin)
else:
input_file = open(input, 'rb')
input_file = open(input, "rb")
if output == '-':
output_file = getattr(sys.stdout, 'buffer', sys.stdout)
if output == "-":
output_file = getattr(sys.stdout, "buffer", sys.stdout)
else:
output_file = open(output, 'wb')
output_file = open(output, "wb")
input_data = input_file.read()
@ -499,44 +501,42 @@ def remarshal(
if transform:
parsed = transform(parsed)
if output_format == 'json':
if output_format == "json":
output_data = encode_json(parsed, ordered, indent_json)
elif output_format == 'msgpack':
elif output_format == "msgpack":
output_data = encode_msgpack(parsed)
elif output_format == 'toml':
elif output_format == "toml":
output_data = encode_toml(parsed, ordered)
elif output_format == 'yaml':
elif output_format == "yaml":
output_data = encode_yaml(parsed, ordered, yaml_options)
elif output_format == 'cbor':
elif output_format == "cbor":
output_data = encode_cbor(parsed)
else:
raise ValueError(
'Unknown output format: {0}'.format(output_format)
)
raise ValueError("Unknown output format: {0}".format(output_format))
if output_format == 'msgpack' or output_format == 'cbor':
if output_format == "msgpack" or output_format == "cbor":
encoded = output_data
else:
encoded = output_data.encode('utf-8')
encoded = output_data.encode("utf-8")
output_file.write(encoded)
output_file.close()
finally:
if 'input_file' in locals():
if "input_file" in locals():
input_file.close()
if 'output_file' in locals():
if "output_file" in locals():
output_file.close()
def main():
try:
run(sys.argv)
except KeyboardInterrupt as e:
except KeyboardInterrupt:
pass
except (IOError, ValueError) as e:
print('Error: {0}'.format(e), file=sys.stderr)
print("Error: {0}".format(e), file=sys.stderr)
sys.exit(1)
if __name__ == '__main__':
if __name__ == "__main__":
main()

View File

@ -1,5 +1,6 @@
import os
import sys
sys.path.insert(0, os.path.abspath('..'))
import remarshal
sys.path.insert(0, os.path.abspath(".."))
import remarshal # noqa: E402, F401

View File

@ -5,7 +5,6 @@
from .context import remarshal
import collections
import datetime
import errno
import os
@ -15,7 +14,7 @@ import sys
import tempfile
import unittest
import cbor2
import cbor2 # type: ignore
import pytest
TEST_PATH = os.path.dirname(os.path.realpath(__file__))
@ -23,17 +22,17 @@ TEST_PATH = os.path.dirname(os.path.realpath(__file__))
def data_file_path(filename):
path_list = [TEST_PATH]
if re.match(r'example\.(json|msgpack|toml|yaml|cbor)$', filename):
path_list.append('..')
if re.match(r"example\.(json|msgpack|toml|yaml|cbor)$", filename):
path_list.append("..")
path_list.append(filename)
return os.path.join(*path_list)
def read_file(filename, binary=False):
with open(data_file_path(filename), 'rb') as f:
with open(data_file_path(filename), "rb") as f:
content = f.read()
if not binary:
content = content.decode('utf-8')
content = content.decode("utf-8")
return content
@ -42,28 +41,27 @@ def sorted_dict(pairs):
def toml_signature(data):
'''A lossy representation for TOML example data for comparison.'''
"""A lossy representation for TOML example data for comparison."""
def strip_more(line):
return re.sub(r' *#.*$', '', line.strip()).replace(' ', '')
return re.sub(r" *#.*$", "", line.strip()).replace(" ", "")
def f(lst):
def q(line):
return (
line.startswith('#') or
line == u'' or
line == u']' or
re.match(r'^".*",?$', line) or
re.match(r'^hosts', line)
line.startswith("#")
or line == ""
or line == "]"
or re.match(r'^".*",?$', line)
or re.match(r"^hosts", line)
)
return sorted(
[strip_more(line) for line in lst if not q(strip_more(line))]
)
return sorted([strip_more(line) for line in lst if not q(strip_more(line))])
return f(data.split("\n"))
class TestRemarshal(unittest.TestCase):
def temp_filename(self):
fd, temp_filename = tempfile.mkstemp()
os.close(fd)
@ -101,7 +99,7 @@ class TestRemarshal(unittest.TestCase):
indent_json=indent_json,
yaml_options=yaml_options,
ordered=ordered,
transform=transform
transform=transform,
)
return read_file(output_filename, binary)
@ -114,362 +112,331 @@ class TestRemarshal(unittest.TestCase):
os.remove(filename)
def test_json2json(self):
output = self.convert_and_read('example.json', 'json', 'json')
reference = read_file('example.json')
output = self.convert_and_read("example.json", "json", "json")
reference = read_file("example.json")
assert output == reference
def test_msgpack2msgpack(self):
output = self.convert_and_read(
'example.msgpack',
'msgpack',
'msgpack',
"example.msgpack",
"msgpack",
"msgpack",
binary=True,
ordered=True,
)
reference = read_file('example.msgpack', binary=True)
reference = read_file("example.msgpack", binary=True)
assert output == reference
def test_toml2toml(self):
output = self.convert_and_read('example.toml', 'toml', 'toml')
reference = read_file('example.toml')
output = self.convert_and_read("example.toml", "toml", "toml")
reference = read_file("example.toml")
assert toml_signature(output) == toml_signature(reference)
def test_yaml2yaml(self):
output = self.convert_and_read('example.yaml', 'yaml', 'yaml')
reference = read_file('example.yaml')
output = self.convert_and_read("example.yaml", "yaml", "yaml")
reference = read_file("example.yaml")
assert output == reference
def test_cbor2cbor(self):
output = self.convert_and_read(
'example.cbor',
'cbor',
'cbor',
binary=True
)
reference = read_file('example.cbor', binary=True)
output = self.convert_and_read("example.cbor", "cbor", "cbor", binary=True)
reference = read_file("example.cbor", binary=True)
self.assert_cbor_same(output, reference)
def test_json2msgpack(self):
def patch(x):
x['owner']['dob'] = datetime.datetime(1979, 5, 27, 7, 32)
x["owner"]["dob"] = datetime.datetime(1979, 5, 27, 7, 32)
return x
output = self.convert_and_read(
'example.json',
'json',
'msgpack',
"example.json",
"json",
"msgpack",
binary=True,
ordered=True,
transform=patch,
)
reference = read_file('example.msgpack', binary=True)
reference = read_file("example.msgpack", binary=True)
assert output == reference
def test_json2cbor(self):
def patch(x):
x['owner']['dob'] = datetime.datetime(
1979, 5, 27,
7, 32, 0, 0,
datetime.timezone.utc
x["owner"]["dob"] = datetime.datetime(
1979, 5, 27, 7, 32, 0, 0, datetime.timezone.utc
)
return x
output = self.convert_and_read(
'example.json',
'json',
'cbor',
"example.json",
"json",
"cbor",
binary=True,
ordered=True,
transform=patch,
)
reference = read_file('example.cbor', binary=True)
reference = read_file("example.cbor", binary=True)
self.assert_cbor_same(output, reference)
def test_json2toml(self):
output = self.convert_and_read('example.json', 'json', 'toml')
reference = read_file('example.toml')
output = self.convert_and_read("example.json", "json", "toml")
reference = read_file("example.toml")
output_sig = toml_signature(output)
# The date in 'example.json' is a string.
reference_sig = toml_signature(
reference.replace(
'1979-05-27T07:32:00Z',
'"1979-05-27T07:32:00+00:00"'
)
reference.replace("1979-05-27T07:32:00Z", '"1979-05-27T07:32:00+00:00"')
)
assert output_sig == reference_sig
def test_json2yaml(self):
output = self.convert_and_read('example.json', 'json', 'yaml')
reference = read_file('example.yaml')
output = self.convert_and_read("example.json", "json", "yaml")
reference = read_file("example.yaml")
# The date in 'example.json' is a string.
reference_patched = reference.replace(
'1979-05-27 07:32:00+00:00',
"'1979-05-27T07:32:00+00:00'"
"1979-05-27 07:32:00+00:00", "'1979-05-27T07:32:00+00:00'"
)
assert output == reference_patched
def test_msgpack2json(self):
output = self.convert_and_read('example.msgpack', 'msgpack', 'json')
reference = read_file('example.json')
output = self.convert_and_read("example.msgpack", "msgpack", "json")
reference = read_file("example.json")
assert output == reference
def test_msgpack2toml(self):
output = self.convert_and_read('example.msgpack', 'msgpack', 'toml')
reference = read_file('example.toml')
output = self.convert_and_read("example.msgpack", "msgpack", "toml")
reference = read_file("example.toml")
assert toml_signature(output) == toml_signature(reference)
def test_msgpack2yaml(self):
output = self.convert_and_read('example.msgpack', 'msgpack', 'yaml')
reference = read_file('example.yaml')
output = self.convert_and_read("example.msgpack", "msgpack", "yaml")
reference = read_file("example.yaml")
assert output == reference
def test_msgpack2cbor(self):
output = self.convert_and_read(
'example.msgpack', 'msgpack', 'cbor',
"example.msgpack",
"msgpack",
"cbor",
binary=True,
)
reference = read_file('example.cbor', binary=True)
reference = read_file("example.cbor", binary=True)
self.assert_cbor_same(output, reference)
def test_toml2json(self):
output = self.convert_and_read('example.toml', 'toml', 'json')
reference = read_file('example.json')
output = self.convert_and_read("example.toml", "toml", "json")
reference = read_file("example.json")
assert output == reference
def test_toml2msgpack(self):
output = self.convert_and_read(
'example.toml',
'toml',
'msgpack',
"example.toml",
"toml",
"msgpack",
binary=True,
transform=lambda col: remarshal.traverse(
col,
dict_callback=sorted_dict
),
transform=lambda col: remarshal.traverse(col, dict_callback=sorted_dict),
)
reference = read_file('example.msgpack', binary=True)
reference = read_file("example.msgpack", binary=True)
assert output == reference
def test_toml2yaml(self):
output = self.convert_and_read('example.toml', 'toml', 'yaml')
reference = read_file('example.yaml')
output = self.convert_and_read("example.toml", "toml", "yaml")
reference = read_file("example.yaml")
assert output == reference
def test_toml2cbor(self):
output = self.convert_and_read(
'example.toml', 'toml', 'cbor',
"example.toml",
"toml",
"cbor",
binary=True,
)
reference = read_file('example.cbor', binary=True)
reference = read_file("example.cbor", binary=True)
self.assert_cbor_same(output, reference)
def test_yaml2json(self):
output = self.convert_and_read('example.yaml', 'yaml', 'json')
reference = read_file('example.json')
output = self.convert_and_read("example.yaml", "yaml", "json")
reference = read_file("example.json")
assert output == reference
def test_yaml2msgpack(self):
output = self.convert_and_read(
'example.yaml',
'yaml',
'msgpack',
"example.yaml",
"yaml",
"msgpack",
ordered=True,
binary=True,
)
reference = read_file('example.msgpack', binary=True)
reference = read_file("example.msgpack", binary=True)
assert output == reference
def test_yaml2toml(self):
output = self.convert_and_read('example.yaml', 'yaml', 'toml')
reference = read_file('example.toml')
output = self.convert_and_read("example.yaml", "yaml", "toml")
reference = read_file("example.toml")
assert toml_signature(output) == toml_signature(reference)
def test_yaml2cbor(self):
output = self.convert_and_read(
'example.yaml', 'yaml', 'cbor',
"example.yaml",
"yaml",
"cbor",
binary=True,
)
reference = read_file('example.cbor', binary=True)
reference = read_file("example.cbor", binary=True)
self.assert_cbor_same(output, reference)
def test_cbor2json(self):
output = self.convert_and_read('example.cbor', 'cbor', 'json')
reference = read_file('example.json')
output = self.convert_and_read("example.cbor", "cbor", "json")
reference = read_file("example.json")
assert output == reference
def test_cbor2toml(self):
output = self.convert_and_read('example.cbor', 'cbor', 'toml')
reference = read_file('example.toml')
output = self.convert_and_read("example.cbor", "cbor", "toml")
reference = read_file("example.toml")
output_sig = toml_signature(output)
reference_sig = toml_signature(reference)
assert output_sig == reference_sig
def test_cbor2yaml(self):
output = self.convert_and_read('example.cbor', 'cbor', 'yaml')
reference = read_file('example.yaml')
output = self.convert_and_read("example.cbor", "cbor", "yaml")
reference = read_file("example.yaml")
assert output == reference
def test_cbor2msgpack(self):
output = self.convert_and_read(
'example.cbor',
'cbor',
'msgpack',
"example.cbor",
"cbor",
"msgpack",
binary=True,
ordered=True,
transform=lambda col: remarshal.traverse(
col,
dict_callback=sorted_dict
),
transform=lambda col: remarshal.traverse(col, dict_callback=sorted_dict),
)
reference = read_file('example.msgpack', binary=True)
reference = read_file("example.msgpack", binary=True)
assert output == reference
def test_missing_wrap(self):
with pytest.raises(ValueError) as context:
output = self.convert_and_read('array.json', 'json', 'toml')
with pytest.raises(ValueError):
self.convert_and_read("array.json", "json", "toml")
def test_wrap(self):
output = self.convert_and_read('array.json', 'json', 'toml',
wrap='data')
reference = read_file('array.toml')
output = self.convert_and_read("array.json", "json", "toml", wrap="data")
reference = read_file("array.toml")
assert output == reference
def test_unwrap(self):
output = self.convert_and_read('array.toml', 'toml', 'json',
unwrap='data',
indent_json=None)
reference = read_file('array.json')
output = self.convert_and_read(
"array.toml", "toml", "json", unwrap="data", indent_json=None
)
reference = read_file("array.json")
assert output == reference
def test_malformed_json(self):
with pytest.raises(ValueError) as context:
self.convert_and_read('garbage', 'json', 'yaml')
with pytest.raises(ValueError):
self.convert_and_read("garbage", "json", "yaml")
def test_malformed_toml(self):
with pytest.raises(ValueError) as context:
self.convert_and_read('garbage', 'toml', 'yaml')
with pytest.raises(ValueError):
self.convert_and_read("garbage", "toml", "yaml")
def test_malformed_yaml(self):
with pytest.raises(ValueError) as context:
self.convert_and_read('garbage', 'yaml', 'json')
with pytest.raises(ValueError):
self.convert_and_read("garbage", "yaml", "json")
def test_binary_to_json(self):
with pytest.raises(ValueError) as context:
self.convert_and_read('bin.msgpack', 'msgpack', 'json')
with pytest.raises(ValueError) as context:
self.convert_and_read('bin.yml', 'yaml', 'json')
with pytest.raises(ValueError):
self.convert_and_read("bin.msgpack", "msgpack", "json")
with pytest.raises(ValueError):
self.convert_and_read("bin.yml", "yaml", "json")
def test_binary_to_msgpack(self):
self.convert_and_read('bin.yml', 'yaml', 'msgpack', binary=True)
self.convert_and_read("bin.yml", "yaml", "msgpack", binary=True)
def test_binary_to_toml(self):
with pytest.raises(ValueError) as context:
self.convert_and_read('bin.msgpack', 'msgpack', 'toml')
with pytest.raises(ValueError) as context:
self.convert_and_read('bin.yml', 'yaml', 'toml')
with pytest.raises(ValueError):
self.convert_and_read("bin.msgpack", "msgpack", "toml")
with pytest.raises(ValueError):
self.convert_and_read("bin.yml", "yaml", "toml")
def test_binary_to_yaml(self):
self.convert_and_read('bin.msgpack', 'msgpack', 'yaml')
self.convert_and_read("bin.msgpack", "msgpack", "yaml")
def test_binary_to_cbor(self):
self.convert_and_read('bin.msgpack', 'msgpack', 'cbor', binary=True)
self.convert_and_read("bin.msgpack", "msgpack", "cbor", binary=True)
def test_yaml_style_default(self):
output = self.convert_and_read('long-line.json', 'json', 'yaml')
reference = read_file('long-line-default.yaml')
output = self.convert_and_read("long-line.json", "json", "yaml")
reference = read_file("long-line-default.yaml")
assert output == reference
def test_yaml_style_single_quote(self):
output = self.convert_and_read(
'long-line.json',
'json',
'yaml',
yaml_options={'default_style': "'"}
"long-line.json", "json", "yaml", yaml_options={"default_style": "'"}
)
reference = read_file('long-line-single-quote.yaml')
reference = read_file("long-line-single-quote.yaml")
assert output == reference
def test_yaml_style_double_quote(self):
output = self.convert_and_read(
'long-line.json',
'json',
'yaml',
yaml_options={'default_style': '"'}
"long-line.json", "json", "yaml", yaml_options={"default_style": '"'}
)
reference = read_file('long-line-double-quote.yaml')
reference = read_file("long-line-double-quote.yaml")
assert output == reference
def test_yaml_style_pipe(self):
output = self.convert_and_read(
'long-line.json',
'json',
'yaml',
yaml_options={'default_style': '|'}
"long-line.json", "json", "yaml", yaml_options={"default_style": "|"}
)
reference = read_file('long-line-pipe.yaml')
reference = read_file("long-line-pipe.yaml")
assert output == reference
def test_yaml_style_gt(self):
output = self.convert_and_read(
'long-line.json',
'json',
'yaml',
yaml_options={'default_style': '>'}
"long-line.json", "json", "yaml", yaml_options={"default_style": ">"}
)
reference = read_file('long-line-gt.yaml')
reference = read_file("long-line-gt.yaml")
assert output == reference
def test_argv0_to_format(self):
def test_format_string(s):
for from_str in 'json', 'toml', 'yaml':
for to_str in 'json', 'toml', 'yaml':
for from_str in "json", "toml", "yaml":
for to_str in "json", "toml", "yaml":
found, from_parsed, to_parsed = remarshal.argv0_to_format(
s.format(from_str, to_str)
)
assert (found, from_parsed, to_parsed) == \
(found, from_str, to_str)
assert (found, from_parsed, to_parsed) == (found, from_str, to_str)
test_format_string('{0}2{1}')
test_format_string('{0}2{1}.exe')
test_format_string('{0}2{1}-script.py')
test_format_string("{0}2{1}")
test_format_string("{0}2{1}.exe")
test_format_string("{0}2{1}-script.py")
def test_format_detection(self):
ext_to_fmt = {
'json': 'json',
'toml': 'toml',
'yaml': 'yaml',
'yml': 'yaml',
"json": "json",
"toml": "toml",
"yaml": "yaml",
"yml": "yaml",
}
for from_ext in ext_to_fmt.keys():
for to_ext in ext_to_fmt.keys():
args = remarshal.parse_command_line([
sys.argv[0],
'input.' + from_ext,
'output.' + to_ext
])
args = remarshal.parse_command_line(
[sys.argv[0], "input." + from_ext, "output." + to_ext]
)
assert args.input_format == ext_to_fmt[from_ext]
assert args.output_format == ext_to_fmt[to_ext]
def test_format_detection_failure_input_stdin(self):
with pytest.raises(SystemExit) as cm:
remarshal.parse_command_line([sys.argv[0], '-'])
remarshal.parse_command_line([sys.argv[0], "-"])
assert cm.value.code == 2
def test_format_detection_failure_input_txt(self):
with pytest.raises(SystemExit) as cm:
remarshal.parse_command_line([sys.argv[0], 'input.txt'])
remarshal.parse_command_line([sys.argv[0], "input.txt"])
assert cm.value.code == 2
def test_format_detection_failure_output_txt(self):
with pytest.raises(SystemExit) as cm:
remarshal.parse_command_line([
sys.argv[0],
'input.json',
'output.txt'
])
remarshal.parse_command_line([sys.argv[0], "input.json", "output.txt"])
assert cm.value.code == 2
def test_run_no_args(self):
@ -479,18 +446,18 @@ class TestRemarshal(unittest.TestCase):
def test_run_help(self):
with pytest.raises(SystemExit) as cm:
remarshal.run([sys.argv[0], '--help'])
remarshal.run([sys.argv[0], "--help"])
assert cm.value.code == 0
def test_run_no_input_file(self):
with pytest.raises(IOError) as cm:
args = [
sys.argv[0],
'-if',
'json',
'-of',
'json',
'fake-input-file-that-almost-certainly-doesnt-exist-2382'
"-if",
"json",
"-of",
"json",
"fake-input-file-that-almost-certainly-doesnt-exist-2382",
]
remarshal.run(args)
assert cm.value.errno == errno.ENOENT
@ -499,55 +466,50 @@ class TestRemarshal(unittest.TestCase):
with pytest.raises(IOError) as cm:
args = [
sys.argv[0],
'-if',
'json',
'-of',
'json',
'-o',
'this_path/almost-certainly/doesnt-exist-5836',
data_file_path('example.json')
"-if",
"json",
"-of",
"json",
"-o",
"this_path/almost-certainly/doesnt-exist-5836",
data_file_path("example.json"),
]
remarshal.run(args)
assert cm.value.errno == errno.ENOENT
def test_run_no_output_format(self):
with pytest.raises(SystemExit) as cm:
remarshal.run([sys.argv[0], data_file_path('array.toml')])
remarshal.run([sys.argv[0], data_file_path("array.toml")])
assert cm.value.code == 2
def test_ordered_simple(self):
for from_ in 'json', 'toml', 'yaml':
for to in 'json', 'toml', 'yaml':
for from_ in "json", "toml", "yaml":
for to in "json", "toml", "yaml":
output = self.convert_and_read(
'order.' + from_,
"order." + from_, from_, to, ordered=True
)
reference = read_file("order." + to)
message = "failed for %s to %s (%r instead of %r)" % (
from_,
to,
ordered=True
)
reference = read_file('order.' + to)
message = 'failed for %s to %s (%r instead of %r)' % (
from_, to, output, reference
output,
reference,
)
assert output == reference, message
def test_ordered_yaml2yaml(self):
output = self.convert_and_read(
'example.yaml',
'yaml',
'yaml',
ordered=True
)
reference = read_file('example.yaml')
output = self.convert_and_read("example.yaml", "yaml", "yaml", ordered=True)
reference = read_file("example.yaml")
assert output == reference
def test_bool_null_key_yaml2json(self):
output = self.convert_and_read(
'bool-null-key.yaml',
'yaml',
'json',
"bool-null-key.yaml",
"yaml",
"json",
indent_json=False,
ordered=True,
)
reference = read_file('bool-null-key.json')
reference = read_file("bool-null-key.json")
assert output == reference