refactor: fill in types for mypy strict mode

This commit is contained in:
D. Bohdan 2023-07-22 21:31:11 +00:00
parent 725700ad41
commit 583a9e559c
3 changed files with 158 additions and 154 deletions

View File

@ -12,7 +12,7 @@ import json
import os.path import os.path
import re import re
import sys import sys
from typing import Any, Callable, Dict, List, Mapping, Sequence, Set, Tuple, Union from typing import Any, Callable, Dict, List, Mapping, Sequence, Set, Tuple, Union, cast
import cbor2 # type: ignore import cbor2 # type: ignore
import dateutil.parser import dateutil.parser
@ -33,7 +33,7 @@ class OrderedDumper(yaml.SafeDumper):
pass pass
def mapping_representer(dumper, data): def mapping_representer(dumper: Any, data: Any) -> Any:
return dumper.represent_mapping( return dumper.represent_mapping(
yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, data.items() yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, data.items()
) )
@ -48,7 +48,7 @@ class TimezoneLoader(yaml.SafeLoader):
pass pass
def timestamp_constructor(loader, node): def timestamp_constructor(loader: Any, node: Any) -> datetime.datetime:
return dateutil.parser.parse(node.value) return dateutil.parser.parse(node.value)
@ -60,7 +60,7 @@ for loader in loaders:
# === JSON === # === JSON ===
def json_default(obj) -> str: def json_default(obj: Any) -> str:
if isinstance(obj, datetime.datetime): if isinstance(obj, datetime.datetime):
return obj.isoformat() return obj.isoformat()
msg = f"{obj!r} is not JSON serializable" msg = f"{obj!r} is not JSON serializable"
@ -89,7 +89,7 @@ def extension_to_format(path: str) -> str:
def parse_command_line(argv: List[str]) -> argparse.Namespace: # noqa: C901. def parse_command_line(argv: List[str]) -> argparse.Namespace: # noqa: C901.
defaults = { defaults: Dict[str, Any] = {
"json_indent": 0, "json_indent": 0,
"ordered": True, "ordered": True,
"yaml_options": {}, "yaml_options": {},
@ -259,13 +259,13 @@ def parse_command_line(argv: List[str]) -> argparse.Namespace: # noqa: C901.
def traverse( def traverse(
col, col: Any,
dict_callback: Callable = lambda x: dict(x), dict_callback: Callable[[List[Tuple[Any, Any]]], Any] = lambda x: dict(x),
list_callback: Callable = lambda x: x, list_callback: Callable[[List[Tuple[Any, Any]]], Any] = lambda x: x,
key_callback: Callable = lambda x: x, key_callback: Callable[[Any], Any] = lambda x: x,
instance_callbacks: Set[Tuple[type, Any]] = set(), instance_callbacks: Set[Tuple[type, Any]] = set(),
default_callback: Callable = lambda x: x, default_callback: Callable[[Any], Any] = lambda x: x,
): ) -> Any:
if isinstance(col, dict): if isinstance(col, dict):
res = dict_callback( res = dict_callback(
[ [
@ -313,9 +313,11 @@ Document = Union[bool, bytes, datetime.datetime, Mapping, None, Sequence, str]
def decode_json(input_data: bytes) -> Document: def decode_json(input_data: bytes) -> Document:
try: try:
return json.loads( doc = json.loads(
input_data.decode("utf-8"), input_data.decode("utf-8"),
) )
return cast(Document, doc)
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
msg = f"Cannot parse as JSON ({e})" msg = f"Cannot parse as JSON ({e})"
raise ValueError(msg) raise ValueError(msg)
@ -323,7 +325,8 @@ def decode_json(input_data: bytes) -> Document:
def decode_msgpack(input_data: bytes) -> Document: def decode_msgpack(input_data: bytes) -> Document:
try: try:
return umsgpack.unpackb(input_data) doc = umsgpack.unpackb(input_data)
return cast(Document, doc)
except umsgpack.UnpackException as e: except umsgpack.UnpackException as e:
msg = f"Cannot parse as MessagePack ({e})" msg = f"Cannot parse as MessagePack ({e})"
raise ValueError(msg) raise ValueError(msg)
@ -331,7 +334,8 @@ def decode_msgpack(input_data: bytes) -> Document:
def decode_cbor(input_data: bytes) -> Document: def decode_cbor(input_data: bytes) -> Document:
try: try:
return cbor2.loads(input_data) doc = cbor2.loads(input_data)
return cast(Document, doc)
except cbor2.CBORDecodeError as e: except cbor2.CBORDecodeError as e:
msg = f"Cannot parse as CBOR ({e})" msg = f"Cannot parse as CBOR ({e})"
raise ValueError(msg) raise ValueError(msg)
@ -341,7 +345,7 @@ def decode_toml(input_data: bytes) -> Document:
try: try:
# Remove TOML Kit's custom classes. # Remove TOML Kit's custom classes.
# https://github.com/sdispater/tomlkit/issues/43 # https://github.com/sdispater/tomlkit/issues/43
return traverse( doc = traverse(
tomlkit.loads(input_data), tomlkit.loads(input_data),
instance_callbacks={ instance_callbacks={
(tomlkit.items.Bool, bool), (tomlkit.items.Bool, bool),
@ -381,6 +385,8 @@ def decode_toml(input_data: bytes) -> Document:
), ),
}, },
) )
return cast(Document, doc)
except tomlkit.exceptions.ParseError as e: except tomlkit.exceptions.ParseError as e:
msg = f"Cannot parse as TOML ({e})" msg = f"Cannot parse as TOML ({e})"
raise ValueError(msg) raise ValueError(msg)
@ -389,7 +395,8 @@ def decode_toml(input_data: bytes) -> Document:
def decode_yaml(input_data: bytes) -> Document: def decode_yaml(input_data: bytes) -> Document:
try: try:
loader = TimezoneLoader loader = TimezoneLoader
return yaml.load(input_data, loader) doc = yaml.load(input_data, loader)
return cast(Document, doc)
except (yaml.scanner.ScannerError, yaml.parser.ParserError) as e: except (yaml.scanner.ScannerError, yaml.parser.ParserError) as e:
msg = f"Cannot parse as YAML ({e})" msg = f"Cannot parse as YAML ({e})"
raise ValueError(msg) raise ValueError(msg)
@ -412,14 +419,14 @@ def decode(input_format: str, input_data: bytes) -> Document:
def encode_json( def encode_json(
data: Document, ordered: bool, indent: Union[bool, int] # noqa: FBT001 data: Document, ordered: bool, indent: Union[bool, int, None] # noqa: FBT001
) -> str: ) -> str:
if indent is True: if indent is True:
indent = 2 indent = 2
separators = (",", ": " if indent else ":") separators = (",", ": " if indent else ":")
def stringify_key(key): def stringify_key(key: Any) -> Any:
if isinstance(key, bool): if isinstance(key, bool):
return "true" if key else "false" return "true" if key else "false"
return "null" if key is None else key return "null" if key is None else key
@ -446,7 +453,7 @@ def encode_json(
def encode_msgpack(data: Document) -> bytes: def encode_msgpack(data: Document) -> bytes:
try: try:
return umsgpack.packb(data) return bytes(umsgpack.packb(data))
except umsgpack.UnsupportedTypeException as e: except umsgpack.UnsupportedTypeException as e:
msg = f"Cannot convert data to MessagePack ({e})" msg = f"Cannot convert data to MessagePack ({e})"
raise ValueError(msg) raise ValueError(msg)
@ -454,13 +461,13 @@ def encode_msgpack(data: Document) -> bytes:
def encode_cbor(data: Document) -> bytes: def encode_cbor(data: Document) -> bytes:
try: try:
return cbor2.dumps(data) return bytes(cbor2.dumps(data))
except cbor2.CBOREncodeError as e: except cbor2.CBOREncodeError as e:
msg = f"Cannot convert data to CBOR ({e})" msg = f"Cannot convert data to CBOR ({e})"
raise ValueError(msg) raise ValueError(msg)
def encode_toml(data: Mapping, ordered: bool) -> str: # noqa: FBT001 def encode_toml(data: Mapping[Any, Any], ordered: bool) -> str: # noqa: FBT001
try: try:
return tomlkit.dumps(data, sort_keys=not ordered) return tomlkit.dumps(data, sort_keys=not ordered)
except AttributeError as e: except AttributeError as e:
@ -478,7 +485,7 @@ def encode_toml(data: Mapping, ordered: bool) -> str: # noqa: FBT001
def encode_yaml( def encode_yaml(
data: Document, ordered: bool, yaml_options: Dict # noqa: FBT001 data: Document, ordered: bool, yaml_options: Dict[Any, Any] # noqa: FBT001
) -> str: ) -> str:
dumper = OrderedDumper if ordered else yaml.SafeDumper dumper = OrderedDumper if ordered else yaml.SafeDumper
try: try:
@ -500,9 +507,9 @@ def encode(
output_format: str, output_format: str,
data: Document, data: Document,
*, *,
json_indent: int, json_indent: Union[int, None],
ordered: bool, ordered: bool,
yaml_options: Dict, yaml_options: Dict[Any, Any],
) -> bytes: ) -> bytes:
if output_format == "json": if output_format == "json":
encoded = encode_json(data, ordered, json_indent).encode("utf-8") encoded = encode_json(data, ordered, json_indent).encode("utf-8")
@ -552,8 +559,8 @@ def remarshal(
output_format: str, output_format: str,
wrap: Union[str, None] = None, wrap: Union[str, None] = None,
unwrap: Union[str, None] = None, unwrap: Union[str, None] = None,
json_indent: int = 0, json_indent: Union[int, None] = 0,
yaml_options: Dict = {}, yaml_options: Dict[Any, Any] = {},
ordered: bool = True, # noqa: FBT001 ordered: bool = True, # noqa: FBT001
transform: Union[Callable[[Document], Document], None] = None, transform: Union[Callable[[Document], Document], None] = None,
) -> None: ) -> None:

View File

@ -3,4 +3,4 @@ import sys
sys.path.insert(0, os.path.abspath("..")) sys.path.insert(0, os.path.abspath(".."))
import remarshal # noqa: E402, F401 import remarshal as remarshal # noqa: E402, PLC0414

View File

@ -3,6 +3,8 @@
# Copyright (c) 2014-2020, 2023 D. Bohdan # Copyright (c) 2014-2020, 2023 D. Bohdan
# License: MIT # License: MIT
from __future__ import annotations
import datetime import datetime
import errno import errno
import os import os
@ -11,6 +13,7 @@ import re
import sys import sys
import tempfile import tempfile
import unittest import unittest
from typing import Any, Callable, Dict, List, Tuple, Union
import cbor2 # type: ignore import cbor2 # type: ignore
import pytest import pytest
@ -20,7 +23,7 @@ from .context import remarshal
TEST_PATH = os.path.dirname(os.path.realpath(__file__)) TEST_PATH = os.path.dirname(os.path.realpath(__file__))
def data_file_path(filename): def data_file_path(filename: str) -> str:
path_list = [TEST_PATH] path_list = [TEST_PATH]
if re.match(r"example\.(json|msgpack|toml|yaml|cbor)$", filename): if re.match(r"example\.(json|msgpack|toml|yaml|cbor)$", filename):
path_list.append("..") path_list.append("..")
@ -28,46 +31,47 @@ def data_file_path(filename):
return os.path.join(*path_list) return os.path.join(*path_list)
def read_file(filename, binary=False): def read_file(filename: str) -> bytes:
with open(data_file_path(filename), "rb") as f: with open(data_file_path(filename), "rb") as f:
content = f.read() return f.read()
if not binary:
content = content.decode("utf-8")
return content
def sorted_dict(pairs): def sorted_dict(pairs: List[Tuple[Any, Any]]) -> Dict[Any, Any]:
return dict(sorted(pairs)) return dict(sorted(pairs))
def toml_signature(data): def toml_signature(data: bytes | str) -> List[str]:
"""A lossy representation for TOML example data for comparison.""" """Return a lossy representation of TOML example data for comparison."""
def strip_more(line): def strip_more(line: str) -> str:
return re.sub(r" *#.*$", "", line.strip()).replace(" ", "") return re.sub(r" *#.*$", "", line.strip()).replace(" ", "")
def f(lst): def sig_lines(lst: List[str]) -> List[str]:
def q(line): def should_drop(line: str) -> bool:
return ( return (
line.startswith("#") line.startswith("#")
or line in ("", "]") or line in ("", "]")
or re.match(r'^".*",?$', line) or bool(re.match(r'^".*",?$', line))
or re.match(r"^hosts", line) or bool(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 should_drop(strip_more(line))]
)
return f(data.split("\n")) str_data = data if isinstance(data, str) else data.decode("utf-8")
return sig_lines(str_data.split("\n"))
class TestRemarshal(unittest.TestCase): class TestRemarshal(unittest.TestCase):
def temp_filename(self): def temp_filename(self) -> str:
fd, temp_filename = tempfile.mkstemp() fd, temp_filename = tempfile.mkstemp()
os.close(fd) os.close(fd)
self.temp_files.append(temp_filename) self.temp_files.append(temp_filename)
return temp_filename return temp_filename
def assert_cbor_same(self, output, reference): def assert_cbor_same(self, output: bytes, reference: bytes) -> None:
# To date, Python's CBOR libraries don't support encoding to # To date, Python's CBOR libraries don't support encoding to
# canonical-form CBOR, so we have to parse and deep-compare. # canonical-form CBOR, so we have to parse and deep-compare.
output_dec = cbor2.loads(output) output_dec = cbor2.loads(output)
@ -76,17 +80,18 @@ class TestRemarshal(unittest.TestCase):
def convert_and_read( def convert_and_read(
self, self,
input, input: str,
input_format, input_format: str,
output_format, output_format: str,
wrap=None, wrap: Union[str, None] = None,
unwrap=None, unwrap: Union[str, None] = None,
json_indent=True, json_indent: Union[int, None] = 2,
yaml_options={}, yaml_options: Dict[Any, Any] = {},
ordered=False, ordered: bool = False, # noqa: FBT001
binary=False, transform: Union[
transform=None, Callable[[remarshal.Document], remarshal.Document], None
): ] = None,
) -> bytes:
output_filename = self.temp_filename() output_filename = self.temp_filename()
remarshal.remarshal( remarshal.remarshal(
data_file_path(input), data_file_path(input),
@ -100,49 +105,49 @@ class TestRemarshal(unittest.TestCase):
ordered=ordered, ordered=ordered,
transform=transform, transform=transform,
) )
return read_file(output_filename, binary)
def setUp(self): return read_file(output_filename)
def setUp(self) -> None:
self.maxDiff = None self.maxDiff = None
self.temp_files = [] self.temp_files: List[str] = []
def tearDown(self): def tearDown(self) -> None:
for filename in self.temp_files: for filename in self.temp_files:
os.remove(filename) os.remove(filename)
def test_json2json(self): def test_json2json(self) -> None:
output = self.convert_and_read("example.json", "json", "json") output = self.convert_and_read("example.json", "json", "json")
reference = read_file("example.json") reference = read_file("example.json")
assert output == reference assert output == reference
def test_msgpack2msgpack(self): def test_msgpack2msgpack(self) -> None:
output = self.convert_and_read( output = self.convert_and_read(
"example.msgpack", "example.msgpack",
"msgpack", "msgpack",
"msgpack", "msgpack",
binary=True,
ordered=True, ordered=True,
) )
reference = read_file("example.msgpack", binary=True) reference = read_file("example.msgpack")
assert output == reference assert output == reference
def test_toml2toml(self): def test_toml2toml(self) -> None:
output = self.convert_and_read("example.toml", "toml", "toml") output = self.convert_and_read("example.toml", "toml", "toml")
reference = read_file("example.toml") reference = read_file("example.toml")
assert toml_signature(output) == toml_signature(reference) assert toml_signature(output) == toml_signature(reference)
def test_yaml2yaml(self): def test_yaml2yaml(self) -> None:
output = self.convert_and_read("example.yaml", "yaml", "yaml") output = self.convert_and_read("example.yaml", "yaml", "yaml")
reference = read_file("example.yaml") reference = read_file("example.yaml")
assert output == reference assert output == reference
def test_cbor2cbor(self): def test_cbor2cbor(self) -> None:
output = self.convert_and_read("example.cbor", "cbor", "cbor", binary=True) output = self.convert_and_read("example.cbor", "cbor", "cbor")
reference = read_file("example.cbor", binary=True) reference = read_file("example.cbor")
self.assert_cbor_same(output, reference) self.assert_cbor_same(output, reference)
def test_json2msgpack(self): def test_json2msgpack(self) -> None:
def patch(x): def patch(x: Any) -> Any:
x["owner"]["dob"] = datetime.datetime(1979, 5, 27, 7, 32) # noqa: DTZ001 x["owner"]["dob"] = datetime.datetime(1979, 5, 27, 7, 32) # noqa: DTZ001
return x return x
@ -150,15 +155,14 @@ class TestRemarshal(unittest.TestCase):
"example.json", "example.json",
"json", "json",
"msgpack", "msgpack",
binary=True,
ordered=True, ordered=True,
transform=patch, transform=patch,
) )
reference = read_file("example.msgpack", binary=True) reference = read_file("example.msgpack")
assert output == reference assert output == reference
def test_json2cbor(self): def test_json2cbor(self) -> None:
def patch(x): def patch(x: Any) -> Any:
x["owner"]["dob"] = datetime.datetime( x["owner"]["dob"] = datetime.datetime(
1979, 5, 27, 7, 32, 0, 0, datetime.timezone.utc 1979, 5, 27, 7, 32, 0, 0, datetime.timezone.utc
) )
@ -168,17 +172,16 @@ class TestRemarshal(unittest.TestCase):
"example.json", "example.json",
"json", "json",
"cbor", "cbor",
binary=True,
ordered=True, ordered=True,
transform=patch, transform=patch,
) )
reference = read_file("example.cbor", binary=True) reference = read_file("example.cbor")
self.assert_cbor_same(output, reference) self.assert_cbor_same(output, reference)
def test_json2toml(self): def test_json2toml(self) -> None:
output = self.convert_and_read("example.json", "json", "toml") output = self.convert_and_read("example.json", "json", "toml").decode("utf-8")
reference = read_file("example.toml") reference = read_file("example.toml").decode("utf-8")
output_sig = toml_signature(output) output_sig = toml_signature(output)
# The date in 'example.json' is a string. # The date in 'example.json' is a string.
reference_sig = toml_signature( reference_sig = toml_signature(
@ -186,215 +189,209 @@ class TestRemarshal(unittest.TestCase):
) )
assert output_sig == reference_sig assert output_sig == reference_sig
def test_json2yaml(self): def test_json2yaml(self) -> None:
output = self.convert_and_read("example.json", "json", "yaml") output = self.convert_and_read("example.json", "json", "yaml").decode("utf-8")
reference = read_file("example.yaml") reference = read_file("example.yaml").decode("utf-8")
# The date in 'example.json' is a string. # The date in 'example.json' is a string.
reference_patched = reference.replace( 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 assert output == reference_patched
def test_msgpack2json(self): def test_msgpack2json(self) -> None:
output = self.convert_and_read("example.msgpack", "msgpack", "json") output = self.convert_and_read("example.msgpack", "msgpack", "json")
reference = read_file("example.json") reference = read_file("example.json")
assert output == reference assert output == reference
def test_msgpack2toml(self): def test_msgpack2toml(self) -> None:
output = self.convert_and_read("example.msgpack", "msgpack", "toml") output = self.convert_and_read("example.msgpack", "msgpack", "toml")
reference = read_file("example.toml") reference = read_file("example.toml")
assert toml_signature(output) == toml_signature(reference) assert toml_signature(output) == toml_signature(reference)
def test_msgpack2yaml(self): def test_msgpack2yaml(self) -> None:
output = self.convert_and_read("example.msgpack", "msgpack", "yaml") output = self.convert_and_read("example.msgpack", "msgpack", "yaml")
reference = read_file("example.yaml") reference = read_file("example.yaml")
assert output == reference assert output == reference
def test_msgpack2cbor(self): def test_msgpack2cbor(self) -> None:
output = self.convert_and_read( output = self.convert_and_read(
"example.msgpack", "example.msgpack",
"msgpack", "msgpack",
"cbor", "cbor",
binary=True,
) )
reference = read_file("example.cbor", binary=True) reference = read_file("example.cbor")
self.assert_cbor_same(output, reference) self.assert_cbor_same(output, reference)
def test_toml2json(self): def test_toml2json(self) -> None:
output = self.convert_and_read("example.toml", "toml", "json") output = self.convert_and_read("example.toml", "toml", "json")
reference = read_file("example.json") reference = read_file("example.json")
assert output == reference assert output == reference
def test_toml2msgpack(self): def test_toml2msgpack(self) -> None:
output = self.convert_and_read( output = self.convert_and_read(
"example.toml", "example.toml",
"toml", "toml",
"msgpack", "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")
assert output == reference assert output == reference
def test_toml2yaml(self): def test_toml2yaml(self) -> None:
output = self.convert_and_read("example.toml", "toml", "yaml") output = self.convert_and_read("example.toml", "toml", "yaml")
reference = read_file("example.yaml") reference = read_file("example.yaml")
assert output == reference assert output == reference
def test_toml2cbor(self): def test_toml2cbor(self) -> None:
output = self.convert_and_read( output = self.convert_and_read(
"example.toml", "example.toml",
"toml", "toml",
"cbor", "cbor",
binary=True,
) )
reference = read_file("example.cbor", binary=True) reference = read_file("example.cbor")
self.assert_cbor_same(output, reference) self.assert_cbor_same(output, reference)
def test_yaml2json(self): def test_yaml2json(self) -> None:
output = self.convert_and_read("example.yaml", "yaml", "json") output = self.convert_and_read("example.yaml", "yaml", "json")
reference = read_file("example.json") reference = read_file("example.json")
assert output == reference assert output == reference
def test_yaml2msgpack(self): def test_yaml2msgpack(self) -> None:
output = self.convert_and_read( output = self.convert_and_read(
"example.yaml", "example.yaml",
"yaml", "yaml",
"msgpack", "msgpack",
ordered=True, ordered=True,
binary=True,
) )
reference = read_file("example.msgpack", binary=True) reference = read_file("example.msgpack")
assert output == reference assert output == reference
def test_yaml2toml(self): def test_yaml2toml(self) -> None:
output = self.convert_and_read("example.yaml", "yaml", "toml") output = self.convert_and_read("example.yaml", "yaml", "toml")
reference = read_file("example.toml") reference = read_file("example.toml")
assert toml_signature(output) == toml_signature(reference) assert toml_signature(output) == toml_signature(reference)
def test_yaml2cbor(self): def test_yaml2cbor(self) -> None:
output = self.convert_and_read( output = self.convert_and_read(
"example.yaml", "example.yaml",
"yaml", "yaml",
"cbor", "cbor",
binary=True,
) )
reference = read_file("example.cbor", binary=True) reference = read_file("example.cbor")
self.assert_cbor_same(output, reference) self.assert_cbor_same(output, reference)
def test_cbor2json(self): def test_cbor2json(self) -> None:
output = self.convert_and_read("example.cbor", "cbor", "json") output = self.convert_and_read("example.cbor", "cbor", "json")
reference = read_file("example.json") reference = read_file("example.json")
assert output == reference assert output == reference
def test_cbor2toml(self): def test_cbor2toml(self) -> None:
output = self.convert_and_read("example.cbor", "cbor", "toml") output = self.convert_and_read("example.cbor", "cbor", "toml")
reference = read_file("example.toml") reference = read_file("example.toml")
output_sig = toml_signature(output) output_sig = toml_signature(output)
reference_sig = toml_signature(reference) reference_sig = toml_signature(reference)
assert output_sig == reference_sig assert output_sig == reference_sig
def test_cbor2yaml(self): def test_cbor2yaml(self) -> None:
output = self.convert_and_read("example.cbor", "cbor", "yaml") output = self.convert_and_read("example.cbor", "cbor", "yaml")
reference = read_file("example.yaml") reference = read_file("example.yaml")
assert output == reference assert output == reference
def test_cbor2msgpack(self): def test_cbor2msgpack(self) -> None:
output = self.convert_and_read( output = self.convert_and_read(
"example.cbor", "example.cbor",
"cbor", "cbor",
"msgpack", "msgpack",
binary=True,
ordered=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")
assert output == reference assert output == reference
def test_missing_wrap(self): def test_missing_wrap(self) -> None:
with pytest.raises(TypeError): with pytest.raises(TypeError):
self.convert_and_read("array.json", "json", "toml") self.convert_and_read("array.json", "json", "toml")
def test_wrap(self): def test_wrap(self) -> None:
output = self.convert_and_read("array.json", "json", "toml", wrap="data") output = self.convert_and_read("array.json", "json", "toml", wrap="data")
reference = read_file("array.toml") reference = read_file("array.toml")
assert output == reference assert output == reference
def test_unwrap(self): def test_unwrap(self) -> None:
output = self.convert_and_read( output = self.convert_and_read(
"array.toml", "toml", "json", unwrap="data", json_indent=None "array.toml", "toml", "json", json_indent=None, unwrap="data"
) )
reference = read_file("array.json") reference = read_file("array.json")
assert output == reference assert output == reference
def test_malformed_json(self): def test_malformed_json(self) -> None:
with pytest.raises(ValueError): with pytest.raises(ValueError):
self.convert_and_read("garbage", "json", "yaml") self.convert_and_read("garbage", "json", "yaml")
def test_malformed_toml(self): def test_malformed_toml(self) -> None:
with pytest.raises(ValueError): with pytest.raises(ValueError):
self.convert_and_read("garbage", "toml", "yaml") self.convert_and_read("garbage", "toml", "yaml")
def test_malformed_yaml(self): def test_malformed_yaml(self) -> None:
with pytest.raises(ValueError): with pytest.raises(ValueError):
self.convert_and_read("garbage", "yaml", "json") self.convert_and_read("garbage", "yaml", "json")
def test_binary_to_json(self): def test_binary_to_json(self) -> None:
with pytest.raises(ValueError): with pytest.raises(ValueError):
self.convert_and_read("bin.msgpack", "msgpack", "json") self.convert_and_read("bin.msgpack", "msgpack", "json")
with pytest.raises(ValueError): with pytest.raises(ValueError):
self.convert_and_read("bin.yml", "yaml", "json") self.convert_and_read("bin.yml", "yaml", "json")
def test_binary_to_msgpack(self): def test_binary_to_msgpack(self) -> None:
self.convert_and_read("bin.yml", "yaml", "msgpack", binary=True) self.convert_and_read("bin.yml", "yaml", "msgpack")
def test_binary_to_toml(self): def test_binary_to_toml(self) -> None:
with pytest.raises(ValueError): with pytest.raises(ValueError):
self.convert_and_read("bin.msgpack", "msgpack", "toml") self.convert_and_read("bin.msgpack", "msgpack", "toml")
with pytest.raises(ValueError): with pytest.raises(ValueError):
self.convert_and_read("bin.yml", "yaml", "toml") self.convert_and_read("bin.yml", "yaml", "toml")
def test_binary_to_yaml(self): def test_binary_to_yaml(self) -> None:
self.convert_and_read("bin.msgpack", "msgpack", "yaml") self.convert_and_read("bin.msgpack", "msgpack", "yaml")
def test_binary_to_cbor(self): def test_binary_to_cbor(self) -> None:
self.convert_and_read("bin.msgpack", "msgpack", "cbor", binary=True) self.convert_and_read("bin.msgpack", "msgpack", "cbor")
def test_yaml_style_default(self): def test_yaml_style_default(self) -> None:
output = self.convert_and_read("long-line.json", "json", "yaml") output = self.convert_and_read("long-line.json", "json", "yaml")
reference = read_file("long-line-default.yaml") reference = read_file("long-line-default.yaml")
assert output == reference assert output == reference
def test_yaml_style_single_quote(self): def test_yaml_style_single_quote(self) -> None:
output = self.convert_and_read( 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 assert output == reference
def test_yaml_style_double_quote(self): def test_yaml_style_double_quote(self) -> None:
output = self.convert_and_read( 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 assert output == reference
def test_yaml_style_pipe(self): def test_yaml_style_pipe(self) -> None:
output = self.convert_and_read( 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 assert output == reference
def test_yaml_style_gt(self): def test_yaml_style_gt(self) -> None:
output = self.convert_and_read( 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 assert output == reference
def test_argv0_to_format(self): def test_argv0_to_format(self) -> None:
def test_format_string(s): def test_format_string(s: str) -> None:
for from_str in "json", "toml", "yaml": for from_str in "json", "toml", "yaml":
for to_str in "json", "toml", "yaml": for to_str in "json", "toml", "yaml":
from_parsed, to_parsed = remarshal.argv0_to_format( from_parsed, to_parsed = remarshal.argv0_to_format(
@ -406,7 +403,7 @@ class TestRemarshal(unittest.TestCase):
test_format_string("{0}2{1}.exe") test_format_string("{0}2{1}.exe")
test_format_string("{0}2{1}-script.py") test_format_string("{0}2{1}-script.py")
def test_format_detection(self): def test_format_detection(self) -> None:
ext_to_fmt = { ext_to_fmt = {
"json": "json", "json": "json",
"toml": "toml", "toml": "toml",
@ -423,32 +420,32 @@ class TestRemarshal(unittest.TestCase):
assert args.input_format == ext_to_fmt[from_ext] assert args.input_format == ext_to_fmt[from_ext]
assert args.output_format == ext_to_fmt[to_ext] assert args.output_format == ext_to_fmt[to_ext]
def test_format_detection_failure_input_stdin(self): def test_format_detection_failure_input_stdin(self) -> None:
with pytest.raises(SystemExit) as cm: 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 assert cm.value.code == 2
def test_format_detection_failure_input_txt(self): def test_format_detection_failure_input_txt(self) -> None:
with pytest.raises(SystemExit) as cm: 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 assert cm.value.code == 2
def test_format_detection_failure_output_txt(self): def test_format_detection_failure_output_txt(self) -> None:
with pytest.raises(SystemExit) as cm: 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 assert cm.value.code == 2
def test_run_no_args(self): def test_run_no_args(self) -> None:
with pytest.raises(SystemExit) as cm: with pytest.raises(SystemExit) as cm:
remarshal.run([sys.argv[0]]) remarshal.run([sys.argv[0]])
assert cm.value.code == 2 assert cm.value.code == 2
def test_run_help(self): def test_run_help(self) -> None:
with pytest.raises(SystemExit) as cm: with pytest.raises(SystemExit) as cm:
remarshal.run([sys.argv[0], "--help"]) remarshal.run([sys.argv[0], "--help"])
assert cm.value.code == 0 assert cm.value.code == 0
def test_run_no_input_file(self): def test_run_no_input_file(self) -> None:
args = [ args = [
sys.argv[0], sys.argv[0],
"-if", "-if",
@ -461,7 +458,7 @@ class TestRemarshal(unittest.TestCase):
remarshal.run(args) remarshal.run(args)
assert cm.value.errno == errno.ENOENT assert cm.value.errno == errno.ENOENT
def test_run_no_output_dir(self): def test_run_no_output_dir(self) -> None:
args = [ args = [
sys.argv[0], sys.argv[0],
"-if", "-if",
@ -476,18 +473,18 @@ class TestRemarshal(unittest.TestCase):
remarshal.run(args) remarshal.run(args)
assert cm.value.errno == errno.ENOENT assert cm.value.errno == errno.ENOENT
def test_run_no_output_format(self): def test_run_no_output_format(self) -> None:
with pytest.raises(SystemExit) as cm: 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 assert cm.value.code == 2
def test_run_short_commands(self): def test_run_short_commands(self) -> None:
for output_format in ["cbor", "json", "msgpack", "toml", "yaml"]: for output_format in ["cbor", "json", "msgpack", "toml", "yaml"]:
remarshal.run( remarshal.run(
[f"json2{output_format}", "-i", data_file_path("example.json")] [f"json2{output_format}", "-i", data_file_path("example.json")]
) )
def test_ordered_simple(self): def test_ordered_simple(self) -> None:
formats = ("json", "toml", "yaml") formats = ("json", "toml", "yaml")
for from_ in formats: for from_ in formats:
for to in formats: for to in formats:
@ -504,44 +501,44 @@ class TestRemarshal(unittest.TestCase):
) )
assert output == reference, message assert output == reference, message
def test_ordered_yaml2yaml(self): def test_ordered_yaml2yaml(self) -> None:
output = self.convert_and_read("example.yaml", "yaml", "yaml", ordered=True) output = self.convert_and_read("example.yaml", "yaml", "yaml", ordered=True)
reference = read_file("example.yaml") reference = read_file("example.yaml")
assert output == reference assert output == reference
def test_bool_null_key_yaml2json(self): def test_bool_null_key_yaml2json(self) -> None:
output = self.convert_and_read( output = self.convert_and_read(
"bool-null-key.yaml", "bool-null-key.yaml",
"yaml", "yaml",
"json", "json",
json_indent=False, json_indent=0,
ordered=True, ordered=True,
) )
reference = read_file("bool-null-key.json") reference = read_file("bool-null-key.json")
assert output == reference assert output == reference
def test_yaml_width_default(self): def test_yaml_width_default(self) -> None:
output = self.convert_and_read( output = self.convert_and_read(
"long-line.json", "long-line.json",
"json", "json",
"yaml", "yaml",
) ).decode("utf-8")
assert len([char for char in output if char == "\n"]) == 4 assert len([char for char in output if char == "\n"]) == 4
def test_yaml_width_30(self): def test_yaml_width_30(self) -> None:
output = self.convert_and_read( output = self.convert_and_read(
"long-line.json", "json", "yaml", yaml_options={"width": 5} "long-line.json", "json", "yaml", yaml_options={"width": 5}
) ).decode("utf-8")
assert len([char for char in output if char == "\n"]) == 21 assert len([char for char in output if char == "\n"]) == 21
def test_yaml_width_120(self): def test_yaml_width_120(self) -> None:
output = self.convert_and_read( output = self.convert_and_read(
"long-line.json", "json", "yaml", yaml_options={"width": 120} "long-line.json", "json", "yaml", yaml_options={"width": 120}
) ).decode("utf-8")
assert len([char for char in output if char == "\n"]) == 3 assert len([char for char in output if char == "\n"]) == 3
def test_yaml_ident_5(self): def test_yaml_ident_5(self) -> None:
output = self.convert_and_read( output = self.convert_and_read(
"long-line.json", "json", "yaml", yaml_options={"indent": 5} "long-line.json", "json", "yaml", yaml_options={"indent": 5}
) ).decode("utf-8")
assert set(re.findall(r"\n +", output)) == {"\n ", "\n "} assert set(re.findall(r"\n +", output)) == {"\n ", "\n "}