1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187
|
"""Tests for msgspec functionality."""
from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
from typing import (
Any,
Callable,
Dict,
List,
Literal,
Mapping,
MutableMapping,
MutableSequence,
NamedTuple,
Sequence,
)
from attrs import define
from hypothesis import given
from msgspec import Struct, to_builtins
from pytest import fixture
from cattrs.fns import identity
from cattrs.preconf.json import make_converter as make_json_converter
from cattrs.preconf.msgspec import MsgspecJsonConverter as Conv
from cattrs.preconf.msgspec import make_converter
from ..typed import simple_typed_classes
@define
class A:
a: int
@define
class B:
"""This class should not be passed through to msgspec."""
a: Any
@define
class C:
"""This class should not be passed through due to a private attribute."""
_a: int
@dataclass
class DataclassA:
a: int
@dataclass
class DataclassC:
"""Msgspec doesn't skip private attributes on dataclasses, so this should work OOB."""
_a: int
class N(NamedTuple):
a: int
class NA(NamedTuple):
"""A complex namedtuple."""
a: A
class NC(NamedTuple):
"""A complex namedtuple."""
a: C
class E(Enum):
TEST = "test"
@fixture
def converter() -> Conv:
return make_converter()
def is_passthrough(fn: Callable) -> bool:
return fn in (identity, to_builtins)
def test_unstructure_passthrough(converter: Conv):
"""Passthrough for simple types works."""
assert converter.get_unstructure_hook(int) == identity
assert converter.get_unstructure_hook(float) == identity
assert converter.get_unstructure_hook(str) == identity
assert is_passthrough(converter.get_unstructure_hook(bytes))
assert converter.get_unstructure_hook(None) == identity
assert is_passthrough(converter.get_unstructure_hook(Literal[1]))
assert is_passthrough(converter.get_unstructure_hook(E))
# Any is special-cased, and we cannot know if it'll match
# the msgspec behavior.
assert not is_passthrough(converter.get_unstructure_hook(List))
assert not is_passthrough(converter.get_unstructure_hook(Sequence))
assert not is_passthrough(converter.get_unstructure_hook(MutableSequence))
assert not is_passthrough(converter.get_unstructure_hook(List[Any]))
assert not is_passthrough(converter.get_unstructure_hook(Sequence))
assert not is_passthrough(converter.get_unstructure_hook(MutableSequence))
assert is_passthrough(converter.get_unstructure_hook(List[int]))
assert is_passthrough(converter.get_unstructure_hook(Sequence[int]))
assert is_passthrough(converter.get_unstructure_hook(MutableSequence[int]))
def test_unstructure_pt_product_types(converter: Conv):
"""Passthrough for product types (attrs, dataclasses...) works."""
assert is_passthrough(converter.get_unstructure_hook(A))
assert not is_passthrough(converter.get_unstructure_hook(B))
assert not is_passthrough(converter.get_unstructure_hook(C))
assert is_passthrough(converter.get_unstructure_hook(DataclassA))
assert is_passthrough(converter.get_unstructure_hook(DataclassC))
assert converter.unstructure(DataclassC(1)) == {"_a": 1}
assert is_passthrough(converter.get_unstructure_hook(N))
assert is_passthrough(converter.get_unstructure_hook(NA))
assert not is_passthrough(converter.get_unstructure_hook(NC))
def test_unstructure_pt_mappings(converter: Conv):
"""Mapping are passed through for unstructuring."""
assert is_passthrough(converter.get_unstructure_hook(Dict[str, str]))
assert is_passthrough(converter.get_unstructure_hook(Dict[int, int]))
assert not is_passthrough(converter.get_unstructure_hook(Dict))
assert not is_passthrough(converter.get_unstructure_hook(dict))
assert not is_passthrough(converter.get_unstructure_hook(Dict[int, B]))
assert not is_passthrough(converter.get_unstructure_hook(Mapping))
assert not is_passthrough(converter.get_unstructure_hook(MutableMapping))
assert is_passthrough(converter.get_unstructure_hook(Dict[int, A]))
assert is_passthrough(converter.get_unstructure_hook(Mapping[int, int]))
assert is_passthrough(converter.get_unstructure_hook(MutableMapping[int, int]))
def test_dump_hook(converter: Conv):
"""Passthrough for dump hooks works."""
assert converter.get_dumps_hook(A) == converter.encoder.encode
assert converter.get_dumps_hook(Dict[str, str]) == converter.encoder.encode
# msgspec cannot handle these, so cattrs does.
assert converter.get_dumps_hook(B) == converter.dumps
def test_get_loads_hook(converter: Conv):
"""`Converter.get_loads_hook` works."""
hook = converter.get_loads_hook(A)
assert hook(b'{"a": 1}') == A(1)
def test_basic_structs(converter: Conv):
"""Handling msgspec structs works."""
class B(Struct):
b: int
assert converter.unstructure(B(1)) == {"b": 1}
assert converter.structure({"b": 1}, B) == B(1)
@given(simple_typed_classes(text_codec="ascii", allow_infinity=False, allow_nan=False))
def test_simple_classes(cls_and_vals):
cl, posargs, kwargs = cls_and_vals
msgspec = make_converter()
json = make_json_converter()
inst = cl(*posargs, **kwargs)
rebuilt_msgspec = msgspec.loads(msgspec.dumps(inst), cl)
rebuilt_json = json.loads(json.dumps(inst), cl)
assert rebuilt_msgspec == rebuilt_json
|