95 lines
2.5 KiB
Python
95 lines
2.5 KiB
Python
from datetime import datetime
|
|
from ipaddress import IPv4Address
|
|
|
|
import pytest
|
|
from botocore.exceptions import ClientError
|
|
|
|
from layercake.dynamodb import (
|
|
DynamoDBCollection,
|
|
DynamoDBPersistenceLayer,
|
|
Key,
|
|
KeyPair,
|
|
TransactItems,
|
|
serialize,
|
|
)
|
|
|
|
|
|
def test_serialize():
|
|
assert serialize(
|
|
{
|
|
'id': '123',
|
|
'sk': 'abc',
|
|
'date': datetime.fromisoformat('2025-03-20T18:29:10.713994'),
|
|
'ip': IPv4Address('127.0.0.1'),
|
|
}
|
|
) == {
|
|
'id': {'S': '123'},
|
|
'sk': {'S': 'abc'},
|
|
'date': {'S': '2025-03-20T18:29:10.713994'},
|
|
'ip': {'S': '127.0.0.1'},
|
|
}
|
|
|
|
|
|
def test_key():
|
|
assert Key(('122', 'abc'), prefix='schedules') == 'schedules#122#abc'
|
|
|
|
|
|
def test_keypair():
|
|
assert KeyPair('123', 'abc') == {'id': '123', 'sk': 'abc'}
|
|
|
|
|
|
def test_transact_write_items(
|
|
dynamodb_seeds,
|
|
dynamodb_persistence_layer: DynamoDBPersistenceLayer,
|
|
):
|
|
transact = TransactItems(dynamodb_persistence_layer.table_name)
|
|
transact.put(item=KeyPair('5OxmMjL-ujoR5IMGegQz', '0'))
|
|
transact.put(item=KeyPair('cpf', '07879819908'))
|
|
transact.put(
|
|
item=KeyPair('email', 'sergio@somosbeta.com.br'),
|
|
cond_expr='attribute_not_exists(sk)',
|
|
)
|
|
transact.put(
|
|
item=KeyPair('5OxmMjL-ujoR5IMGegQz', 'emails#sergio@somosbeta.com.br'),
|
|
cond_expr='attribute_not_exists(sk)',
|
|
)
|
|
|
|
with pytest.raises(ClientError):
|
|
dynamodb_persistence_layer.transact_write_items(transact)
|
|
|
|
|
|
def test_collection(
|
|
dynamodb_seeds,
|
|
dynamodb_persistence_layer: DynamoDBPersistenceLayer,
|
|
):
|
|
collect = DynamoDBCollection(dynamodb_persistence_layer)
|
|
tenant_item = collect.get_item(
|
|
key=KeyPair(
|
|
pk='5OxmMjL-ujoR5IMGegQz',
|
|
sk=Key('tenant'),
|
|
),
|
|
raise_if_missing=False,
|
|
default={},
|
|
)
|
|
assert tenant_item == {}
|
|
|
|
email_item = collect.get_item(
|
|
key=KeyPair(
|
|
pk='5OxmMjL-ujoR5IMGegQz',
|
|
sk=Key('sergio@somosbeta.com.br', prefix='emails'),
|
|
),
|
|
default={},
|
|
)
|
|
assert email_item == {
|
|
'email_verified': True,
|
|
'mx_record_exists': True,
|
|
'sk': 'emails#sergio@somosbeta.com.br',
|
|
'email_primary': True,
|
|
'id': '5OxmMjL-ujoR5IMGegQz',
|
|
'create_date': '2019-03-25T00:00:00-03:00',
|
|
'update_date': '2023-11-09T12:13:04.308986-03:00',
|
|
}
|
|
|
|
with pytest.raises(DynamoDBCollection.MissingError):
|
|
collect.get_item(key=KeyPair('5OxmMjL-ujoR5IMGegQz', 'notfound'))
|