50 lines
1.4 KiB
Python
50 lines
1.4 KiB
Python
import os
|
|
|
|
import boto3
|
|
import jsonlines
|
|
import pytest
|
|
|
|
from layercake.dynamodb import DynamoDBPersistenceLayer
|
|
|
|
PYTEST_TABLE_NAME = os.getenv('PYTEST_TABLE_NAME', 'pytest')
|
|
# Check `pytest.ini` for more details.
|
|
DYNAMODB_ENDPOINT_URL = os.getenv('DYNAMODB_ENDPOINT_URL')
|
|
PK = os.getenv('DYNAMODB_PARTITION_KEY', 'pk')
|
|
SK = os.getenv('DYNAMODB_SORT_KEY', 'sk')
|
|
|
|
|
|
@pytest.fixture
|
|
def dynamodb_client():
|
|
client = boto3.client('dynamodb', endpoint_url=DYNAMODB_ENDPOINT_URL)
|
|
client.create_table(
|
|
AttributeDefinitions=[
|
|
{'AttributeName': PK, 'AttributeType': 'S'},
|
|
{'AttributeName': SK, 'AttributeType': 'S'},
|
|
],
|
|
TableName=PYTEST_TABLE_NAME,
|
|
KeySchema=[
|
|
{'AttributeName': PK, 'KeyType': 'HASH'},
|
|
{'AttributeName': SK, 'KeyType': 'RANGE'},
|
|
],
|
|
ProvisionedThroughput={
|
|
'ReadCapacityUnits': 123,
|
|
'WriteCapacityUnits': 123,
|
|
},
|
|
)
|
|
|
|
yield client
|
|
|
|
client.delete_table(TableName=PYTEST_TABLE_NAME)
|
|
|
|
|
|
@pytest.fixture()
|
|
def dynamodb_persistence_layer(dynamodb_client) -> DynamoDBPersistenceLayer:
|
|
return DynamoDBPersistenceLayer(PYTEST_TABLE_NAME, dynamodb_client)
|
|
|
|
|
|
@pytest.fixture()
|
|
def dynamodb_seeds(dynamodb_client):
|
|
with jsonlines.open('tests/seeds.jsonl') as lines:
|
|
for line in lines:
|
|
dynamodb_client.put_item(TableName=PYTEST_TABLE_NAME, Item=line)
|