Turns out, setting up pytest and localstack is more difficult than expected, but got it working :)
I have two modes:
- Local mode: localstack is running in a terminal window using
localstack start. This is useful for TDD, as you don't spin up new containers everytime. CICD mode: localstack is not running and it would be automatically started before all tests start.
Install packages:
pip install docker boto3 localstack_utils localstack-clientSetup pytest conftest:
conftest.py
import boto3
import docker
import pytest
import localstack_client.session
from localstack_utils.localstack import startup_localstack, stop_localstack
# patch boto3 to automatically use localstack.
@pytest.fixture(autouse=True)
def boto3_localstack_patch(monkeypatch):
session_ls = localstack_client.session.Session()
monkeypatch.setattr(boto3, "client", session_ls.client)
monkeypatch.setattr(boto3, "resource", session_ls.resource)
# check if localstack running locally using docker lib, if not running, use localstack lib to start it.
def is_localstack_running() -> bool:
try:
docker_client = docker.DockerClient(base_url='unix://var/run/docker.sock')
container = docker_client.containers.get("localstack-main")
return container.attrs['State']['Status'] == "running"
except:
return False
@pytest.fixture(autouse=True, scope="session")
def setup_localstack():
if not is_localstack_running():
print("Setup localstack")
startup_localstack()
yield
if not is_localstack_running():
print("Teardown localstack")
stop_localstack()
Fixture to cleanup after each test:
@pytest.fixture(autouse=True)
def aws_fixture():
# print("Setup")
yield
# print("Cleanup")
# Cleanup S3
s3 = boto3.client("s3")
buckets = [item["Name"] for item in s3.list_buckets()["Buckets"]]
for bucket in buckets:
s3.delete_bucket(Bucket=bucket)
# Cleanup DynamoDB"
dynamodb = boto3.client("dynamodb")
tables_names = dynamodb.list_tables()["TableNames"]
for table_name in tables_names:
dynamodb.delete_table(TableName=table_name)
Then in the handler, add boto3 client. This funky way complies with AWS recommended way initialising global libraries and pytest can patch boto3 too.
handler.py
import functools
# DynamoDB Client
@functools.cache
def dynamodb_client() -> DynamoDBClient:
return boto3.client("dynamodb")
Thats it.
Full example:
handler_blog.py
from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSEvent
from aws_lambda_powertools.utilities.typing import LambdaContext
import functools
from mypy_boto3_dynamodb import DynamoDBClient
import boto3
# DynamoDB Client setup
@functools.cache
def dynamodb_client() -> DynamoDBClient:
return boto3.client("dynamodb")
def handler(event: SQSEvent, context: LambdaContext):
dynamodb_client().update_item(
TableName="super-table",
Key={"objectId": {"N": "123456"}},
ExpressionAttributeNames={"#name": "name"},
ExpressionAttributeValues={":name": {"S": "batman"}},
UpdateExpression="set #name = :name",
ReturnValues="NONE",
)
return []
handler_blog_test.py
import pytest
import boto3
from typing import Literal
from pydantic import BaseModel
from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSEvent
from .handler_blog import handler
@pytest.fixture(autouse=True)
def aws_fixture():
# print("Setup")
yield
# print("Cleanup")
# DynamoDB"
dynamodb = boto3.client("dynamodb")
tables_names = dynamodb.list_tables()["TableNames"]
for table_name in tables_names:
dynamodb.delete_table(TableName=table_name)
class CreateDynamoDB(BaseModel):
table_name: str
pk_name: str
pk_type: Literal["S", "N"]
@pytest.fixture
def create_dynamodb_table(create_dynamodb_table_config: CreateDynamoDB):
config = create_dynamodb_table_config
dynamodb = boto3.client("dynamodb")
dynamodb.create_table(
TableName=config.table_name,
KeySchema=[
{"AttributeName": config.pk_name, "KeyType": "HASH"},
],
AttributeDefinitions=[
{
"AttributeName": config.pk_name,
"AttributeType": config.pk_type,
}
],
BillingMode="PAY_PER_REQUEST",
)
class InsertDataSQSMessage(BaseModel):
id: int
name: str
def get_sqs_event_stub(body: InsertDataSQSMessage) -> SQSEvent:
return SQSEvent(
{
"Records": [
{
"messageId": "4f332f15-3930-4a00-8831-1706016678846",
"receiptHandle": "AQEB/123456789012+bjpfjcbH0fslWvMxNSXEJWn/VNCIi0TYmuZakYNQpQhhcl2EoPseeM4ctyfd/OQ5eiMqWhta+L+iZYIuHRQiIIjmMgJrfJsl6aVHI1vYQvTTwhxaBJh2582kvuAaRvQ0gbLzT/Pe+Zp+123456789012/2Luka8cdrsLlSHEHI+21N+tN5dOaxBoGCJk1wZti6UmcrEzz3T+123456789012/O+mbqSPvJEJnbGasJRUFcKIfocbokN4sMSl8eJJKN1QkWPqxinVmk1DkEYzyY+rzSTjE8IBgcGRrxc293eYDJdfzISXo8j97h83ITP4fm1vMDA2w0/cDvvL3m4ACmZjwoZWdfoBTvJwbB8bXEa86Ykew==",
"body": body.model_dump_json(),
"attributes": {
"ApproximateReceiveCount": "1",
"SentTimestamp": "1706016678845",
"SenderId": "XXXXXXXXXXXXXXXXXXXXX",
"ApproximateFirstReceiveTimestamp": "1706016678846",
},
"messageAttributes": {},
"md5OfBody": "28f07e09c08aba530422dd193f991111",
"eventSource": "aws:sqs",
"eventSourceARN": "arn:aws:sqs:eu-central-1:123456789012:data",
"awsRegion": "eu-central-1",
}
]
}
)
class TestHandlerBlog:
def setup_method(self, method):
pass
def teardown_method(self, method):
pass
create_dynamodb_table_config = CreateDynamoDB(table_name="super-table", pk_name="objectId", pk_type="N")
@pytest.mark.parametrize("create_dynamodb_table_config", [create_dynamodb_table_config])
def test_insert_some_data_to_dynamodb(self, create_dynamodb_table):
# Prepare
# Act
event = get_sqs_event_stub(InsertDataSQSMessage(id=18393972737, name="company"))
result = handler(event=event, context={})
# Assert function
assert result == []
# Assert Table content
response = boto3.client("dynamodb").scan(TableName="super-table")
table_records = response["Items"]
assert len(table_records) == 1
assert table_records == [
{
"objectId": {"N": "123456"},
"name": {"S": "batman"},
}
]
ps. I also mapped vs code key "F1" to "testing.runAtCursor" and "F2" to "testing.runAll".
Top comments (0)