Add local AWS test environment

Closes #5

Author: alexeygrigorev

.env.example

@@ -8,7 +8,14 @@ ALLOWED_HOSTS=localhost,127.0.0.1
 CSRF_TRUSTED_ORIGINS=
 EMAIL_BACKEND=django.core.mail.backends.console.EmailBackend
 AWS_SES_CONFIGURATION_SET=
+AWS_ENDPOINT_URL=
+AWS_ACCESS_KEY_ID=
+AWS_SECRET_ACCESS_KEY=
 SQS_TRANSACTIONAL_EMAIL_QUEUE_URL=
 SQS_CAMPAIGN_EMAIL_QUEUE_URL=
 SQS_EMAIL_EVENTS_QUEUE_URL=
 SQS_SES_WEBHOOKS_QUEUE_URL=
+TRANSACTIONAL_EMAIL_QUEUE_NAME=transactional-email
+CAMPAIGN_EMAIL_QUEUE_NAME=campaign-email
+SES_WEBHOOKS_QUEUE_NAME=ses-webhooks
+EMAIL_EVENTS_QUEUE_NAME=email-events

Makefile

@@ -1,4 +1,4 @@
-.PHONY: setup migrate run test lint format
+.PHONY: setup migrate run localstack test test-aws-local lint format
 
 setup:
 	@test -f .env || cp .env.example .env
@@ -11,9 +11,15 @@ migrate:
 run:
 	uv run python manage.py runserver
 
+localstack:
+	docker compose --profile aws-local up localstack
+
 test:
 	uv run pytest
 
+test-aws-local:
+	uv run pytest -m aws_local
+
 lint:
 	uv run ruff check .
 

conftest.py

@@ -0,0 +1,57 @@
+import os
+import socket
+from uuid import uuid4
+
+import boto3
+import pytest
+
+LOCALSTACK_ENDPOINT = os.environ.get("AWS_ENDPOINT_URL", "http://localhost:4566")
+AWS_REGION = os.environ.get("AWS_REGION", "us-east-1")
+
+os.environ.setdefault("AWS_ACCESS_KEY_ID", "test")
+os.environ.setdefault("AWS_SECRET_ACCESS_KEY", "test")
+os.environ.setdefault("AWS_DEFAULT_REGION", AWS_REGION)
+os.environ.setdefault("AWS_REGION", AWS_REGION)
+os.environ.setdefault("AWS_ENDPOINT_URL", LOCALSTACK_ENDPOINT)
+os.environ.setdefault("AWS_EC2_METADATA_DISABLED", "true")
+
+
+@pytest.fixture
+def aws_test_env(monkeypatch):
+    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "test")
+    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "test")
+    monkeypatch.setenv("AWS_DEFAULT_REGION", AWS_REGION)
+    monkeypatch.setenv("AWS_REGION", AWS_REGION)
+    monkeypatch.setenv("AWS_ENDPOINT_URL", LOCALSTACK_ENDPOINT)
+
+
+@pytest.fixture
+def localstack_available(aws_test_env):
+    host, port = _host_port_from_endpoint(LOCALSTACK_ENDPOINT)
+    try:
+        with socket.create_connection((host, port), timeout=0.5):
+            return True
+    except OSError:
+        pytest.skip(f"LocalStack is not available at {LOCALSTACK_ENDPOINT}")
+
+
+@pytest.fixture
+def local_sqs_client(localstack_available):
+    return boto3.client("sqs", region_name=AWS_REGION, endpoint_url=LOCALSTACK_ENDPOINT)
+
+
+@pytest.fixture
+def unique_queue_name():
+    def build(prefix):
+        return f"{prefix}-{uuid4().hex}"
+
+    return build
+
+
+def _host_port_from_endpoint(endpoint):
+    without_scheme = endpoint.removeprefix("http://").removeprefix("https://")
+    host_port = without_scheme.split("/", 1)[0]
+    if ":" not in host_port:
+        return host_port, 443 if endpoint.startswith("https://") else 80
+    host, port = host_port.rsplit(":", 1)
+    return host, int(port)

datamailer/settings.py

@@ -118,9 +118,14 @@ def bool_env(name, *, default):
 DEFAULT_FROM_EMAIL = os.environ.get("DEFAULT_FROM_EMAIL", "newsletter@example.com")
 
 AWS_REGION = os.environ.get("AWS_REGION", "us-east-1")
+AWS_ENDPOINT_URL = os.environ.get("AWS_ENDPOINT_URL", "")
 PUBLIC_BASE_URL = os.environ.get("PUBLIC_BASE_URL", "http://localhost:8000").rstrip("/")
 AWS_SES_CONFIGURATION_SET = os.environ.get("AWS_SES_CONFIGURATION_SET", "")
 SQS_TRANSACTIONAL_EMAIL_QUEUE_URL = os.environ.get("SQS_TRANSACTIONAL_EMAIL_QUEUE_URL", "")
 SQS_CAMPAIGN_EMAIL_QUEUE_URL = os.environ.get("SQS_CAMPAIGN_EMAIL_QUEUE_URL", "")
 SQS_EMAIL_EVENTS_QUEUE_URL = os.environ.get("SQS_EMAIL_EVENTS_QUEUE_URL", "")
 SQS_SES_WEBHOOKS_QUEUE_URL = os.environ.get("SQS_SES_WEBHOOKS_QUEUE_URL", "")
+TRANSACTIONAL_EMAIL_QUEUE_NAME = os.environ.get("TRANSACTIONAL_EMAIL_QUEUE_NAME", "transactional-email")
+CAMPAIGN_EMAIL_QUEUE_NAME = os.environ.get("CAMPAIGN_EMAIL_QUEUE_NAME", "campaign-email")
+SES_WEBHOOKS_QUEUE_NAME = os.environ.get("SES_WEBHOOKS_QUEUE_NAME", "ses-webhooks")
+EMAIL_EVENTS_QUEUE_NAME = os.environ.get("EMAIL_EVENTS_QUEUE_NAME", "email-events")

docker-compose.yml

@@ -0,0 +1,15 @@
+services:
+  localstack:
+    image: localstack/localstack:stable
+    profiles: ["aws-local"]
+    ports:
+      - "4566:4566"
+    environment:
+      SERVICES: sqs,ses
+      AWS_DEFAULT_REGION: us-east-1
+      DEBUG: "0"
+    healthcheck:
+      test: ["CMD", "curl", "-fsS", "http://localhost:4566/_localstack/health"]
+      interval: 5s
+      timeout: 3s
+      retries: 20

docs/testing.md

@@ -0,0 +1,56 @@
+# Testing
+
+Datamailer tests must not require real AWS credentials for normal local development or CI.
+
+## Default Test Suite
+
+Run:
+
+```bash
+make test
+```
+
+The default suite uses:
+
+- Django unit/view tests.
+- Pure Python worker helper tests.
+- `botocore.stub.Stubber` for SES payload assertions.
+- Skipped LocalStack tests when no local AWS endpoint is running.
+
+## Local AWS Tests
+
+Run LocalStack:
+
+```bash
+make localstack
+```
+
+In another terminal:
+
+```bash
+AWS_ACCESS_KEY_ID=test \
+AWS_SECRET_ACCESS_KEY=test \
+AWS_ENDPOINT_URL=http://localhost:4566 \
+make test-aws-local
+```
+
+Local AWS tests are marked with:
+
+```python
+pytestmark = pytest.mark.aws_local
+```
+
+These tests use LocalStack for SQS wiring. They create unique queue names per test and do not depend on real AWS resources.
+
+## What Uses Mocks
+
+SES correctness should primarily use `botocore.stub.Stubber`. LocalStack SES support can be useful for smoke tests, but the important checks are exact payloads and how the app handles returned message IDs.
+
+Lambda workers should expose pure Python handlers that accept AWS event dictionaries. Tests invoke those handlers directly with SQS-shaped events instead of trying to run Lambda inside LocalStack.
+
+## AWS Safety Rules
+
+- Test settings use fake credentials.
+- Tests that call AWS-compatible endpoints must pass `endpoint_url`.
+- Real AWS credentials are reserved for staging smoke tests.
+- Standard SQS is at-least-once, so worker tests must cover idempotency as sender logic is added.

mailing/aws.py

@@ -0,0 +1,18 @@
+import boto3
+from django.conf import settings
+
+
+def aws_client(service_name, *, endpoint_url=None):
+    return boto3.client(
+        service_name,
+        region_name=settings.AWS_REGION,
+        endpoint_url=endpoint_url if endpoint_url is not None else settings.AWS_ENDPOINT_URL or None,
+    )
+
+
+def sqs_client(*, endpoint_url=None):
+    return aws_client("sqs", endpoint_url=endpoint_url)
+
+
+def ses_client(*, endpoint_url=None):
+    return aws_client("ses", endpoint_url=endpoint_url)

mailing/ses.py

@@ -0,0 +1,20 @@
+from django.conf import settings
+
+
+def send_email(*, ses_client, source, to_email, subject, html_body, text_body=""):
+    body = {"Html": {"Charset": "UTF-8", "Data": html_body}}
+    if text_body:
+        body["Text"] = {"Charset": "UTF-8", "Data": text_body}
+
+    params = {
+        "Source": source,
+        "Destination": {"ToAddresses": [to_email]},
+        "Message": {
+            "Subject": {"Charset": "UTF-8", "Data": subject},
+            "Body": body,
+        },
+    }
+    if settings.AWS_SES_CONFIGURATION_SET:
+        params["ConfigurationSetName"] = settings.AWS_SES_CONFIGURATION_SET
+
+    return ses_client.send_email(**params)["MessageId"]

mailing/sqs.py

@@ -0,0 +1,44 @@
+import json
+
+
+def partial_batch_response(failed_message_ids):
+    return {
+        "batchItemFailures": [
+            {"itemIdentifier": message_id}
+            for message_id in failed_message_ids
+        ]
+    }
+
+
+def records_from_messages(messages, *, event_source_arn="arn:aws:sqs:us-east-1:000000000000:test"):
+    return {
+        "Records": [
+            {
+                "messageId": message["MessageId"],
+                "receiptHandle": message["ReceiptHandle"],
+                "body": message["Body"],
+                "attributes": message.get("Attributes", {}),
+                "messageAttributes": message.get("MessageAttributes", {}),
+                "md5OfBody": message.get("MD5OfBody", ""),
+                "eventSource": "aws:sqs",
+                "eventSourceARN": event_source_arn,
+                "awsRegion": "us-east-1",
+            }
+            for message in messages
+        ]
+    }
+
+
+def json_body(record):
+    return json.loads(record["body"])
+
+
+def process_sqs_event(event, handler):
+    failed_message_ids = []
+    for record in event.get("Records", []):
+        try:
+            handler(json_body(record), record)
+        except Exception:
+            failed_message_ids.append(record["messageId"])
+
+    return partial_batch_response(failed_message_ids)

mailing/tests/test_localstack_sqs.py

@@ -0,0 +1,57 @@
+import json
+
+import pytest
+
+from mailing.sqs import json_body, process_sqs_event, records_from_messages
+
+pytestmark = pytest.mark.aws_local
+
+
+def test_localstack_sqs_can_enqueue_and_receive(local_sqs_client, unique_queue_name):
+    queue_url = local_sqs_client.create_queue(
+        QueueName=unique_queue_name("transactional-email"),
+    )["QueueUrl"]
+
+    local_sqs_client.send_message(
+        QueueUrl=queue_url,
+        MessageBody=json.dumps({"type": "transactional_email", "message_id": "msg_123"}),
+    )
+
+    response = local_sqs_client.receive_message(
+        QueueUrl=queue_url,
+        MaxNumberOfMessages=1,
+        WaitTimeSeconds=1,
+    )
+
+    event = records_from_messages(response["Messages"])
+    processed = []
+    batch_response = process_sqs_event(event, lambda body, record: processed.append(body))
+
+    assert json_body(event["Records"][0]) == {"type": "transactional_email", "message_id": "msg_123"}
+    assert processed == [{"type": "transactional_email", "message_id": "msg_123"}]
+    assert batch_response == {"batchItemFailures": []}
+
+
+def test_localstack_sqs_can_attach_dead_letter_queue(local_sqs_client, unique_queue_name):
+    dlq_url = local_sqs_client.create_queue(QueueName=unique_queue_name("transactional-email-dlq"))["QueueUrl"]
+    dlq_arn = local_sqs_client.get_queue_attributes(
+        QueueUrl=dlq_url,
+        AttributeNames=["QueueArn"],
+    )["Attributes"]["QueueArn"]
+
+    queue_url = local_sqs_client.create_queue(
+        QueueName=unique_queue_name("transactional-email"),
+        Attributes={
+            "RedrivePolicy": json.dumps({
+                "deadLetterTargetArn": dlq_arn,
+                "maxReceiveCount": "3",
+            })
+        },
+    )["QueueUrl"]
+
+    attributes = local_sqs_client.get_queue_attributes(
+        QueueUrl=queue_url,
+        AttributeNames=["RedrivePolicy"],
+    )["Attributes"]
+
+    assert json.loads(attributes["RedrivePolicy"])["deadLetterTargetArn"] == dlq_arn