File: test_msgspec_cpython.py

package info (click to toggle)
python-cattrs 25.3.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 1,812 kB
  • sloc: python: 12,236; makefile: 155
file content (187 lines) | stat: -rw-r--r-- 5,450 bytes parent folder | download
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