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
|
import json
from enum import IntEnum
import pytest
from pydantic_core import CoreConfig, SchemaSerializer, core_schema
try:
import numpy
except ImportError:
numpy = None
class IntSubClass(int):
pass
class MyIntEnum(IntEnum):
one = 1
two = 2
class FloatSubClass(float):
pass
# A number well outside of i64 range
_BIG_NUMBER_BYTES = b'1' + (b'0' * 40)
@pytest.mark.parametrize('custom_type_schema', [None, 'any'])
@pytest.mark.parametrize(
'schema_type,value,expected_python,expected_json',
[
('int', 1, 1, b'1'),
('int', int(_BIG_NUMBER_BYTES), int(_BIG_NUMBER_BYTES), _BIG_NUMBER_BYTES),
('bool', True, True, b'true'),
('bool', False, False, b'false'),
('float', 1.0, 1.0, b'1.0'),
('float', 42.31415, 42.31415, b'42.31415'),
('none', None, None, b'null'),
('int', IntSubClass(42), IntSubClass(42), b'42'),
('int', MyIntEnum.one, MyIntEnum.one, b'1'),
('float', FloatSubClass(42), FloatSubClass(42), b'42.0'),
],
)
def test_simple_serializers(schema_type, value, expected_python, expected_json, custom_type_schema):
if custom_type_schema is None:
schema = {'type': schema_type}
else:
schema = {'type': custom_type_schema}
s = SchemaSerializer(schema)
v = s.to_python(value)
assert v == expected_python
assert type(v) == type(expected_python)
assert s.to_json(value) == expected_json
v_json = s.to_python(value, mode='json')
v_json_expected = json.loads(expected_json)
assert v_json == v_json_expected
assert type(v_json) == type(v_json_expected)
def test_int_to_float():
"""
See https://github.com/pydantic/pydantic-core/pull/866
"""
s = SchemaSerializer(core_schema.float_schema())
v_plain = s.to_python(1)
assert v_plain == 1
assert type(v_plain) == int
v_plain_subclass = s.to_python(IntSubClass(1))
assert v_plain_subclass == IntSubClass(1)
assert type(v_plain_subclass) == IntSubClass
v_json = s.to_python(1, mode='json')
assert v_json == 1.0
assert type(v_json) == float
v_json_subclass = s.to_python(IntSubClass(1), mode='json')
assert v_json_subclass == 1
assert type(v_json_subclass) == float
assert s.to_json(1) == b'1.0'
assert s.to_json(IntSubClass(1)) == b'1.0'
def test_int_to_float_key():
"""
See https://github.com/pydantic/pydantic-core/pull/866
"""
s = SchemaSerializer(core_schema.dict_schema(core_schema.float_schema(), core_schema.float_schema()))
v_plain = s.to_python({1: 1})
assert v_plain == {1: 1}
assert type(list(v_plain.keys())[0]) == int
assert type(v_plain[1]) == int
v_json = s.to_python({1: 1}, mode='json')
assert v_json == {'1': 1.0}
assert type(v_json['1']) == float
assert s.to_json({1: 1}) == b'{"1":1.0}'
@pytest.mark.parametrize('schema_type', ['int', 'bool', 'float', 'none'])
def test_simple_serializers_fallback(schema_type):
s = SchemaSerializer({'type': schema_type})
with pytest.warns(
UserWarning,
match=rf'Expected `{schema_type}` - serialized value may not be as expected \[input_value=\[1, 2, 3\], input_type=list\]',
):
assert s.to_python([1, 2, 3]) == [1, 2, 3]
with pytest.warns(
UserWarning,
match=rf"Expected `{schema_type}` - serialized value may not be as expected \[input_value=\[1, 2, b'bytes'\], input_type=list\]",
):
assert s.to_python([1, 2, b'bytes'], mode='json') == [1, 2, 'bytes']
with pytest.warns(
UserWarning,
match=rf'Expected `{schema_type}` - serialized value may not be as expected \[input_value=\[1, 2, 3\], input_type=list\]',
):
assert s.to_json([1, 2, 3]) == b'[1,2,3]'
@pytest.mark.skipif(numpy is None, reason='numpy is not installed')
def test_numpy():
s = SchemaSerializer(core_schema.float_schema())
v = s.to_python(numpy.float64(1.0))
assert v == 1.0
assert type(v) == numpy.float64
v = s.to_python(numpy.float64(1.0), mode='json')
assert v == 1.0
assert type(v) == float
assert s.to_json(numpy.float64(1.0)) == b'1.0'
@pytest.mark.parametrize(
'value,expected_json,config',
[
# default values of ser_json_inf_nan
(float('inf'), 'null', {}),
(float('-inf'), 'null', {}),
(float('nan'), 'null', {}),
# explicit values of ser_json_inf_nan
(float('inf'), 'null', CoreConfig(ser_json_inf_nan='null')),
(float('-inf'), 'null', CoreConfig(ser_json_inf_nan='null')),
(float('nan'), 'null', CoreConfig(ser_json_inf_nan='null')),
(float('inf'), 'Infinity', CoreConfig(ser_json_inf_nan='constants')),
(float('-inf'), '-Infinity', CoreConfig(ser_json_inf_nan='constants')),
(float('nan'), 'NaN', CoreConfig(ser_json_inf_nan='constants')),
(float('inf'), '"Infinity"', CoreConfig(ser_json_inf_nan='strings')),
(float('-inf'), '"-Infinity"', CoreConfig(ser_json_inf_nan='strings')),
(float('nan'), '"NaN"', CoreConfig(ser_json_inf_nan='strings')),
],
)
def test_float_inf_and_nan_serializers(value, expected_json, config):
s = SchemaSerializer(core_schema.float_schema(), config)
# Python can represent these values without needing any changes
assert s.to_python(value) is value
assert s.to_python(value, mode='json') is value
# Serialized JSON value respects the ser_json_inf_nan setting
assert s.to_json(value).decode() == expected_json
|