/
test_utils.py
62 lines (51 loc) · 1.87 KB
/
test_utils.py
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
import base64
from decimal import Decimal
from functools import partial
import pytest
from aiodynamo.utils import deserialize, dy2py
from boto3.dynamodb.types import DYNAMODB_CONTEXT, TypeDeserializer
def test_binary_decode():
assert dy2py({"test": {"B": base64.b64encode(b"hello")}}, float) == {
"test": b"hello"
}
@pytest.mark.parametrize(
"value,numeric_type,result",
[
({"N": "1.2",}, float, 1.2),
({"NS": ["1.2"]}, float, {1.2}),
({"N": "1.2"}, DYNAMODB_CONTEXT.create_decimal, Decimal("1.2")),
({"NS": ["1.2"]}, DYNAMODB_CONTEXT.create_decimal, {Decimal("1.2")}),
],
)
def test_numeric_decode(value, numeric_type, result):
assert deserialize(value, numeric_type) == result
def test_serde_compatibility():
def generate_item(nest):
item = {
"hash": {"S": "string",},
"range": {"B": base64.b64encode(b"bytes"),},
"null": {"NULL": True},
"true": {"BOOL": True},
"false": {"BOOL": False},
"int": {"N": "42"},
"float": {"N": "4.2"},
"numeric_set": {"NS": ["42", "4.2"]},
"string_set": {"SS": ["hello", "world"]},
"binary_set": {
"BS": [base64.b64encode(b"hello"), base64.b64encode(b"world")]
},
}
if nest:
item["list"] = {"L": [{"M": generate_item(False)}]}
return item
item = generate_item(True)
class BinaryDeserializer(TypeDeserializer):
def _deserialize_b(self, value):
return base64.b64decode(value)
def deserialize_item(item, deserializer):
return {k: deserializer(v) for k, v in item.items()}
fast = deserialize_item(
item, partial(deserialize, numeric_type=DYNAMODB_CONTEXT.create_decimal)
)
boto = deserialize_item(item, BinaryDeserializer().deserialize)
assert fast == boto