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
|
import base64
import json
from enum import Enum
import pytest
from pydantic_core import PydanticSerializationError, SchemaSerializer, core_schema, to_json
def test_bytes():
s = SchemaSerializer(core_schema.bytes_schema())
assert s.to_python(b'foobar') == b'foobar'
assert s.to_python('emoji 💩'.encode()) == 'emoji 💩'.encode()
assert s.to_json(b'foobar') == b'"foobar"'
assert s.to_python(b'foobar', mode='json') == 'foobar'
json_emoji = s.to_json('emoji 💩'.encode())
# note! serde_json serializes unicode characters differently
assert json_emoji == b'"emoji \xf0\x9f\x92\xa9"'
assert json.loads(json_emoji) == 'emoji 💩'
def test_bytes_invalid_all():
s = SchemaSerializer(core_schema.bytes_schema())
assert s.to_python(b'\x81') == b'\x81'
msg = 'Error serializing to JSON: invalid utf-8 sequence of 1 bytes from index 0'
with pytest.raises(PydanticSerializationError, match=msg):
s.to_json(b'\x81')
def test_bytes_invalid_cpython():
# PyO3/pyo3#2770 is now fixed
s = SchemaSerializer(core_schema.bytes_schema())
with pytest.raises(UnicodeDecodeError, match="'utf-8' codec can't decode byte 0x81 in position 0: invalid utf-8"):
s.to_python(b'\x81', mode='json')
def test_bytes_dict_key():
s = SchemaSerializer(core_schema.dict_schema(core_schema.bytes_schema(), core_schema.int_schema()))
assert s.to_python({b'foobar': 123}) == {b'foobar': 123}
assert s.to_python({b'foobar': 123}, mode='json') == {'foobar': 123}
assert s.to_json({b'foobar': 123}) == b'{"foobar":123}'
def test_bytes_fallback():
s = SchemaSerializer(core_schema.bytes_schema())
with pytest.warns(
UserWarning,
match=r'Expected `bytes` - serialized value may not be as expected \[input_value=123, input_type=int\]',
):
assert s.to_python(123) == 123
with pytest.warns(
UserWarning,
match=r'Expected `bytes` - serialized value may not be as expected \[input_value=123, input_type=int\]',
):
assert s.to_python(123, mode='json') == 123
with pytest.warns(
UserWarning,
match=r'Expected `bytes` - serialized value may not be as expected \[input_value=123, input_type=int\]',
):
assert s.to_json(123) == b'123'
with pytest.warns(
UserWarning,
match=r"Expected `bytes` - serialized value may not be as expected \[input_value='foo', input_type=str\]",
):
assert s.to_json('foo') == b'"foo"'
class BytesSubclass(bytes):
pass
class BasicClass:
pass
class BytesMixin(bytes, BasicClass):
pass
class BytesEnum(bytes, Enum):
foo = b'foo-value'
bar = b'bar-value'
@pytest.mark.parametrize('schema_type', ['bytes', 'any'])
@pytest.mark.parametrize(
'input_value,expected_json',
[(BytesSubclass(b'foo'), 'foo'), (BytesMixin(b'foo'), 'foo'), (BytesEnum.foo, 'foo-value')],
)
def test_subclass_bytes(schema_type, input_value, expected_json):
s = SchemaSerializer({'type': schema_type})
v = s.to_python(input_value)
assert v == input_value
assert type(v) == type(input_value)
v = s.to_python(input_value, mode='json')
assert v == expected_json
assert type(v) == str
assert s.to_json(input_value) == json.dumps(expected_json).encode('utf-8')
def test_bytes_base64():
s = SchemaSerializer(core_schema.bytes_schema(), {'ser_json_bytes': 'base64'})
assert s.to_python(b'foobar') == b'foobar'
assert s.to_json(b'foobar') == b'"Zm9vYmFy"'
assert s.to_python(b'foobar', mode='json') == 'Zm9vYmFy'
assert base64.b64decode(s.to_python(b'foobar', mode='json').encode()) == b'foobar'
# with padding
assert s.to_json(b'foo bar') == b'"Zm9vIGJhcg=="'
assert s.to_python(b'foo bar', mode='json') == 'Zm9vIGJhcg=='
assert base64.b64decode(s.to_python(b'foo bar', mode='json').encode()) == b'foo bar'
def test_bytes_hex():
s = SchemaSerializer(core_schema.bytes_schema(), {'ser_json_bytes': 'hex'})
assert s.to_python(b'\xff\xff') == b'\xff\xff'
assert s.to_json(b'\xff\xff') == b'"ffff"'
assert s.to_python(b'\xff\xff', mode='json') == 'ffff' == b'\xff\xff'.hex()
def test_bytes_base64_dict_key():
s = SchemaSerializer(core_schema.dict_schema(core_schema.bytes_schema()), {'ser_json_bytes': 'base64'})
assert s.to_python({b'foo bar': 123}, mode='json') == {'Zm9vIGJhcg==': 123}
assert s.to_json({b'foo bar': 123}) == b'{"Zm9vIGJhcg==":123}'
def test_any_bytes_base64():
s = SchemaSerializer(core_schema.any_schema(), {'ser_json_bytes': 'base64'})
assert s.to_python(b'foobar') == b'foobar'
assert s.to_json(b'foobar') == b'"Zm9vYmFy"'
assert s.to_json({b'foobar': 123}) == b'{"Zm9vYmFy":123}'
assert s.to_python({b'foobar': 123}, mode='json') == {'Zm9vYmFy': 123}
class BasicModel:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
def test_bytes_mode_set_via_model_config_not_serializer_config():
s = SchemaSerializer(
core_schema.model_schema(
BasicModel,
core_schema.model_fields_schema(
{
'foo': core_schema.model_field(core_schema.bytes_schema()),
}
),
config=core_schema.CoreConfig(ser_json_bytes='base64'),
)
)
bm = BasicModel(foo=b'foobar')
assert s.to_python(bm) == {'foo': b'foobar'}
assert s.to_json(bm) == b'{"foo":"Zm9vYmFy"}'
assert s.to_python(bm, mode='json') == {'foo': 'Zm9vYmFy'}
# assert doesn't override serializer config
# in V3, we can change the serialization settings provided to to_json to override model config settings,
# but that'd be a breaking change
BasicModel.__pydantic_serializer__ = s
assert to_json(bm, bytes_mode='utf8') == b'{"foo":"Zm9vYmFy"}'
assert to_json({'foo': b'some bytes'}, bytes_mode='base64') == b'{"foo":"c29tZSBieXRlcw=="}'
assert to_json({'bar': bm}, bytes_mode='base64') == b'{"bar":{"foo":"Zm9vYmFy"}}'
|
