Eight Bugs, Eight Tests: Testing a Hexagonal Python Service on AWS

In Serverless Python Microservices on AWS: My Reference Architecture the testing part is three sentences long. Fakes for the domain, moto for the adapters, end-to-end tests against a real environment. That is true, but it skips the interesting part: which bug does each of those levels catch, and why does the level before it miss it?

This post is the long version. I built a small service, planted eight bugs in it and wrote the test that catches each one. For each bug, one level of the pyramid is the first and cheapest place to catch it, and the level before it misses the bug. Later levels often fail too, only slower and with more setup. So no level replaces another.

This matters more now that Claude, Codex, GitHub Copilot or Cursor write many of our tests. To judge a generated test suite, you have to know which test should catch which bug.

All code is on GitHub: python-aws-testing-pyramid. The code there is correct. Every bug lives as a patch in docs/bugs/, so you can plant it with make bug-<n>, watch the right test fail and revert it with make unbug.

At a Glance

And the eight bugs:

The Example Service

The domain is workshop registrations. A workshop has a fixed number of seats. Signed-in users book one or two of them.

• Active seats never exceed the capacity. Cancelled registrations do not count.
• A person holds at most two seats per workshop.
• Cancelling is possible until 48 hours before the start, and only by the person who registered.
• No registrations after the start.
• Every save increments the workshop’s version. Saving a stale version fails (optimistic locking).

The layout follows the reference architecture: a uv workspace with a _domain package, an _infra package and a Lambdalith, one Lambda function for all routes, with the Powertools router. CDK deploys a shared, a stateful and a stateless stack.

The domain defines two ports:

workshop_service_domain/ports.py

class WorkshopRepository(Protocol):
    def get(self, workshop_id: str) -> Workshop | None: ...
    def save(self, workshop: Workshop) -> None: ...  # raises ConcurrentUpdateError
    def list_upcoming(self, now: datetime) -> list[Workshop]: ...


class Clock(Protocol):
    def now(self) -> datetime: ...

There is no DynamoDB term in them. No pk, no LastEvaluatedKey. Part 2 of this series puts a SQL adapter behind the same port, and neither the domain nor the unit tests will change.

The use cases get their dependencies through the constructor. On a ConcurrentUpdateError, RegisterForWorkshop and CancelRegistration reload the workshop and try again: three attempts in total, with exponential backoff and jitter. The backoff is a constructor parameter, so tests pass zero.

Unit Tests: Fakes Instead of Patching

The unit tests check rules and use cases. No AWS, no moto, no mock.patch, no MagicMock, no freezegun. Time comes only through the Clock port.

Most rules are checked directly on the Workshop aggregate, without any fake:

tests/unit/test_workshop.py

def test_cancelled_registrations_do_not_count_towards_capacity() -> None:
    workshop = make_workshop(capacity=2)
    registration = workshop.register("alice", 2, NOW)
    workshop.cancel(registration.id, "alice", NOW)

    workshop.register("bob", 2, NOW)

    assert workshop.free_seats == 0

The use-case tests get a FakeWorkshopRepository and a FakeClock through the constructor. Two details make the fake worth having:

1. It stores deep copies. get and save both copy with model_copy(deep=True). Without that, a test that changes a loaded workshop and forgets to call save still passes, because the “stored” object is the same object.
2. It checks the version like the real adapter. For the retry test it offers a test control instead of patching: fail_next_save_with_conflict(times=2).

tests/unit/test_use_cases.py

def test_register_retries_after_a_conflict(repository, clock) -> None:
    workshop = make_workshop()
    repository.save(workshop)
    repository.fail_next_save_with_conflict(times=2)

    RegisterForWorkshop(repository, clock, NO_WAIT)(workshop.id, "alice", 1)

    assert repository.save_calls == 1 + 3

One save sets up the workshop. Then come three attempts: two conflicts and one success.

Bug 1: Capacity Counts Cancelled Registrations

workshop_service_domain/models.py

return sum(r.seats for r in self.registrations if r.is_active)
return sum(r.seats for r in self.registrations)

____ test_cancelled_registrations_do_not_count_towards_capacity ____
E   workshop_service_domain.errors.CapacityExceeded: 0 seats left
FAILED tests/unit/test_workshop.py::test_cancelled_registrations_do_not_count_towards_capacity
FAILED tests/unit/test_use_cases.py::test_cancel_retries_after_a_conflict
...
5 failed, 80 passed, 14 deselected

The 14 deselected tests are the aws variants. The other failures are one contract test, on fake and moto, and one handler test. Only the first unit test names the broken rule.

This is the cheapest bug in the post. It runs in milliseconds and needs nothing but Python. Every rule that can be checked here should be checked here.

Contract Tests: Keeping the Fake Honest

The most common objection to fakes is that they drift. The fake says one thing, the real database does another, and the unit tests stay green although the fake is wrong.

Contract tests are the answer. One test suite describes how every WorkshopRepository must behave, and a fixture runs it against all of them:

tests/contract/conftest.py

# One parameter per adapter. Part 2 adds "sql" here and nothing else.
@pytest.fixture(params=[FAKE, MOTO, pytest.param(AWS, marks=pytest.mark.aws)])
def repository(request):
    if request.param == FAKE:
        yield FakeWorkshopRepository()
    elif request.param == MOTO:
        with moto_table() as table:
            yield tracked_repository(table)
    else:
        yield tracked_repository(request.getfixturevalue("aws_table"))

tracked_repository is the real adapter, with cleanup after the test.

The suite checks a round trip with time zones and get with an unknown id. It also checks a stale save, that cancelled registrations are kept, and that list_upcoming skips past workshops and sorts by start. The aws variant runs against the deployed stage’s table. Only it proves that the fake behaves like real DynamoDB, not just like moto.

Bug 2: The Fake Does Not Check the Version

tests/fakes/repository.py

stored = self._workshops.get(workshop.id)
stored_version = stored.version if stored else 0
if stored_version != workshop.version:
    raise ConcurrentUpdateError(workshop.id)

All unit tests stay green. The retry test uses fail_next_save_with_conflict(), which still works. Nothing in the unit tests ever saves a stale version for real.

____ test_save_with_stale_version_raises[fake] ____
E   Failed: DID NOT RAISE ConcurrentUpdateError
FAILED tests/contract/test_workshop_repository.py::test_save_with_stale_version_raises[fake]
FAILED tests/contract/test_workshop_repository.py::test_two_new_workshops_with_the_same_id_conflict[fake]

The [moto] variant of the same test passes. The difference between the two lines is the whole argument for contract tests.

While building the repository, the contract tests also found a real difference between fake and adapter that I had not planted. Two registrations created at the same instant came back in a different order from DynamoDB than from the fake. The aggregate now keeps registrations ordered by (created_at, id) itself. Without contract tests, that would have been a flaky test somewhere much later.

Integration Tests: DynamoDB

The fake cannot tell you anything about DynamoDB items. That is what the integration tests are for: number types, the datetime round trip, the content of the transaction, the GSI query.

The adapter keeps one table with pk and sk:

get is a strongly consistent query on pk. save is one TransactWriteItems: a conditional update of META on version, plus a put of every registration that changed.

For list_upcoming, every item also carries gsi1pk=WORKSHOPS and gsi1sk=<starts_at>#<id>#<sk>. The GSI projects all attributes (ALL), so one query with gsi1sk > now returns whole upcoming workshops, registrations included. Timestamps are fixed-width ISO 8601 strings in UTC, so string order is time order.

The key names, the GSI and its projection live in one file, schema.py. The adapter, the moto fixture and the CDK stack all read them from there.

The same tests run against two backends:

tests/conftest.py

DYNAMODB_BACKENDS = [MOTO, pytest.param(AWS, marks=pytest.mark.aws)]

@pytest.fixture(params=DYNAMODB_BACKENDS)
def dynamodb_table(request):
    if request.param == AWS:
        yield request.getfixturevalue("aws_table")
    else:
        with moto_table() as table:
            yield table

The aws variant uses the stage’s real table and the credentials from the environment: locally the developer role, in CI the pipeline role. Missing credentials fail the run, because a silent skip would be a false green in CI. The GSI is eventually consistent, so the aws variant polls briefly after a write.

Bug 3: starts_at as a Float

workshop_service_infra/dynamodb.py

":starts_at": _iso(workshop.starts_at),
":starts_at": workshop.starts_at.timestamp(),

The fake stores Python objects in a dict, so it does not care. The adapter’s serializer does: boto3 rejects the float before the request reaches moto.

FAILED tests/integration/dynamodb/test_dynamodb_repository.py::test_items_have_the_documented_layout[moto]
FAILED tests/integration/dynamodb/test_dynamodb_repository.py::test_timestamps_are_utc_iso_strings_and_numbers_are_numbers[moto]
...
E   TypeError: Float types are not supported. Use Decimal types instead.

The contract and handler tests on moto fail too, because they also run the real adapter.

Bug 4: The GSI Projection

moto is good at validating API calls. A GSI query with ConsistentRead=True fails on moto 5.2 with the same ValidationException as on DynamoDB. What moto cannot know is how your table was actually created. The moto fixture builds its table from schema.py. The real table comes from CDK, which should read the projection from schema.py as well. Bug 4 is a CDK stack that does not:

cdk/src/workshop_cdk/stateful.py

projection_type=dynamodb.ProjectionType(schema.GSI1_PROJECTION),
projection_type=dynamodb.ProjectionType.KEYS_ONLY,

make test stays green. Against the deployed table:

FAILED tests/contract/test_workshop_repository.py::test_list_upcoming_skips_past_workshops_and_sorts[aws] - KeyError: 'title'
FAILED tests/integration/dynamodb/test_dynamodb_repository.py::test_list_upcoming_reads_whole_workshops_from_the_index[aws] - KeyError: 'user_id'
================= 2 failed, 12 passed, 22 deselected in 30.60s =================

The index only returns the keys, so the adapter finds neither title on the META item nor user_id on a registration. The fix for the class of bug is the single source of truth in schema.py. The test for it is the same integration test, run against the table CDK really created.

Integration Tests: The Handler

functions/api separates build_app(repository, clock) from the Lambda entry point. The tests call build_app with the DynamoDB adapter on moto and a FakeClock, and send it REST API proxy events with authorizer claims. So they run the same code as the Lambda, from the event down to the DynamoDB items.

The tests check routing, validation, error mapping and the response schema:

Missing claims give 401. The error mapping is one explicit table in the app, and anything not in it becomes a logged 500.

Bug 5: A Conflict Becomes 500

functions/api/app.py

WorkshopAlreadyStarted: HTTPStatus.UNPROCESSABLE_ENTITY,
ConcurrentUpdateError: HTTPStatus.CONFLICT,

The unit tests call the use case and see the ConcurrentUpdateError, exactly as intended. They never see the HTTP status. The handler test replaces save on the adapter with a function that always raises ConcurrentUpdateError. It is the only patch in the integration tests:

____ test_conflict_after_all_retries_is_409 ____
E   AssertionError: assert (<HTTPStatus....lServerError') == (409, 'ConcurrentUpdateError')
E     At index 0 diff: <HTTPStatus.INTERNAL_SERVER_ERROR: 500> != 409
FAILED tests/integration/api/test_handler.py::test_conflict_after_all_retries_is_409

A 500 instead of a 409 is not cosmetic. A 409 tells the client to reload and try again. A 500 says the server is broken, and it counts against the error-rate alarm.

Architecture Tests

I wrote about the approach in Enforcing Python Architecture Rules with PyTestArch. For this layout the rules are:

1. workshop_service_domain imports neither boto3, botocore, workshop_service_infra nor functions.
2. workshop_service_infra does not import functions.
3. Only functions.api imports workshop_service_infra.
4. tests/unit imports neither workshop_service_infra nor moto, boto3, botocore, unittest or freezegun.

Rule 4 keeps the unit tests honest: fakes only, no patching.

Why does this need a test at all? The domain has its own pyproject.toml with only pydantic in it. But locally every workspace package shares one .venv, and in Lambda boto3 is there anyway. In this setup, the pyproject.toml prevents nothing.

Bug 6: A Use Case Catches ClientError

workshop_service_domain/use_cases.py

from botocore.exceptions import ClientError
...
        except ConcurrentUpdateError:
        except (ConcurrentUpdateError, ClientError):

It looks defensive, and every functional test stays green, because the code still works. Only the import graph shows it:

E   AssertionError: "workshop_service_domain.use_cases" imports "botocore.exceptions".
FAILED tests/architecture/test_import_rules.py::test_domain_has_no_infrastructure_imports

The damage is real, though. The retry now also repeats AccessDenied, throttling and validation errors: three attempts with backoff before the real error reaches the log. For throttling, that adds load at the moment DynamoDB asks for less. And the domain now knows it runs on AWS. The SQL adapter in Part 2 raises other exceptions, so the domain would need to import those as well.

E2E Tests Through API Gateway

The E2E tests run after the deploy and go only through API Gateway. The stage they run against comes from a CDK app in Python, parametrised by STAGE:

flowchart LR
    T["E2E and load tests"] -->|ID token| AG["API Gateway REST API<br/>Cognito authorizer"]
    AG --> L["Lambda<br/>functions.api"]
    L --> D[("DynamoDB<br/>workshops-&lt;stage&gt;")]
    T -->|"AdminCreateUser, AdminInitiateAuth"| C["Cognito user pool<br/>+ test client"]
    C -.-> AG
    T -->|"direct reads"| D
    S["SSM<br/>/workshops/&lt;stage&gt;/"] -.-> T
    S -.->|"table name, at deploy time"| L

Every request carries a real Cognito token and passes the authorizer, the Lambda and the Lambda’s IAM role. That makes this the only level here that runs the Lambda’s IAM policy. A CDK assertion test could check the policy statement, but only for what I already thought of.

The Lambda runs Python 3.14 on arm64, packaged with uv as in AWS Lambda Packaging with Python and uv. The Cognito test client only allows ADMIN_USER_PASSWORD_AUTH. That flow needs AWS credentials with rights on the user pool, so the client is useless without them. That is why it can exist in prod too.

A session fixture creates two Cognito users per run, alice and bob, and deletes them at the end. Two users, because rules depend on identity: the seat limit per person, cancelling only your own registration.

Every writing test checks two things: the API response and the items in DynamoDB.

tests/e2e/test_workshops_api.py

def test_bob_cannot_cancel_alices_registration(alice, bob, users, new_workshop, state):
    workshop = new_workshop()
    registration = expect(alice.register(workshop["id"], 1), 201)
    before = state.read(workshop["id"])

    expect(bob.cancel(workshop["id"], registration["id"]), 403)

    assert state.read(workshop["id"]) == before

The state reader reads raw items with a strongly consistent query, not through the adapter. A mapping bug that writes and reads the same wrong value would be invisible through the adapter. It translates the items into a neutral state: version, and per registration user_id, seats and status. Part 2 adds a SQL state reader for the same assertions.

Why look into the table at all, when the API answers are right? At deploy time, the stateless stack reads the table name from SSM. CDK uses that name for the Lambda’s TABLE_NAME and for its IAM grant. With a wrong SSM path, the Lambda would write to another stage’s table without any AccessDenied, and the API answers would still agree with each other. Only a look into the expected table shows it.

The cancellation deadline needs no time travel here: a workshop in 24 hours gives 422 on cancel, one in seven days gives 204.

The tests cover every DynamoDB operation the Lambda performs: get, list_upcoming and save. On a 500 they print the x-amzn-RequestId, so the error is easy to find in the Lambda log.

Bug 7: The Lambda Policy Misses the Index

cdk/src/workshop_cdk/stateless.py

table.grant_read_write_data(function)
function.add_to_role_policy(
    iam.PolicyStatement(
        actions=["dynamodb:Query", "dynamodb:PutItem", "dynamodb:UpdateItem"],
        resources=[table.table_arn],
    )
)

A hand-written policy that looks complete. It allows the table but not <table ARN>/index/*. save still works: TransactWriteItems has no IAM action of its own, so IAM checks PutItem and UpdateItem for each item in the transaction.

Nothing before E2E catches it. moto checks no IAM, and the aws variant runs with the developer or pipeline role, which usually has broader rights. I could let it assume the Lambda role, but then the Lambda role would have to trust those roles, only for tests.

E   AssertionError: GET /api/workshops: expected 200, got 500 {"error":"InternalServerError","message":"internal error"} (x-amzn-RequestId: 80843cb5-7f13-43fa-9564-960aed52434d)
FAILED tests/e2e/test_workshops_api.py::test_create_register_list_cancel
========================= 1 failed, 7 passed in 21.25s =========================

Seven of eight E2E tests stay green. Creating, registering and cancelling all work. Only the one call that touches the index fails. That is why every Lambda operation needs at least one test.

Load Tests: Invariants Under Real Concurrency

The load test checks invariants, not only latency. It runs manually, by default against a stage of its own, and the runs in this post did too.

The scenario: one workshop with 20 seats and 100 Cognito test users. 100 Locust users start at the same time, each with its own token, and each books one seat. On 201 or 422 it stops, on 409 it tries again. At the end the test reads the items directly from DynamoDB, with the same state reader as the E2E tests, and checks:

• exactly 20 active seats: no overbooking, and no unused seat despite demand
• no 5xx responses

A hook at the end sets the exit code, so a violated invariant fails the run.

A clean run on a fresh stage in eu-central-1, October 2026:

responses per status: {201: 20, 409: 9, 422: 80}
p50 2300 ms, p95 2600 ms
active seats in DynamoDB: 20 of 20
invariants hold

The 409s are conflicts that were still there after the Lambda’s three attempts; the Locust users then tried again. I did not break the latency down. All 100 requests arrive at once, so cold starts are a likely part of it.

Bug 8: TransactionConflict Is Not Translated

workshop_service_infra/dynamodb.py

_CONFLICT_CODES = {"ConditionalCheckFailed", "TransactionConflict"}
_CONFLICT_CODES = {"ConditionalCheckFailed"}

When two TransactWriteItems touch the same item at the same time, DynamoDB cancels one of them with a TransactionCanceledException. Its cancellation reason is either ConditionalCheckFailed (the version moved on) or TransactionConflict (another transaction was in flight on the item). Both mean “someone else changed this workshop”. The adapter translated only one of them.

sequenceDiagram
    participant A as Lambda A
    participant D as DynamoDB
    participant B as Lambda B
    A->>D: TransactWriteItems (META, version 5)
    B->>D: TransactWriteItems (META, version 5)
    Note over D: A's transaction is still in flight
    D-->>B: TransactionCanceledException (TransactionConflict)
    D-->>A: OK, version 6
    Note over B: With bug 8: 500 instead of a retry

The adapter reads the code from the CancellationReasons of the exception (_is_conflict). That translation matters more than it looks. The AWS SDKs do not retry a cancelled transaction (DynamoDB docs). So the retry in the use case is the only one there is.

Single requests never cause a transaction conflict, neither against moto nor against AWS. moto 5.2 goes further. In my tests, it ran parallel transactions one after the other and never reported TransactionConflict at all. So every level before this one stays green.

Error report
# occurrences      Error
1                  POST POST /registrations: 500 {"error":"InternalServerError","message":"internal error"}
responses per status: {201: 20, 422: 79, 500: 1}
active seats in DynamoDB: 20 of 20
INVARIANTS VIOLATED: 1 responses with 5xx

One 500 in 100 requests. The data is still correct, the seats add up. That makes this kind of bug easy to miss in production: it is rare and only shows under load.

The clean run had nine 409s, so conflicts were common. Most of them must have been ConditionalCheckFailed: a Lambda read version 5, and another one had already written version 6. TransactionConflict needs two transactions on the item at the same moment. If it were common, bug 8 would have produced more than one 500.

What Runs When

Locally and in CI the same Make targets run (see the README). Every branch gets its own stage, and main deploys prod. Without a relational database, a deploy takes about three minutes in CI and a destroy about one. Three GitHub Actions workflows run the targets:

1. Pull request: make test, then deploy a stage for the branch, make test-aws, make test-e2e, and destroy the stage, also when a test fails. About five minutes in total.
2. Push to main: make test, deploy prod, make smoke (no token must give 401) and make test-e2e.
3. Manual: the load test, against a stage of its own or prod (never during a prod deploy). A bug demo plants bug 4, 7 or 8 on a stage of its own.

Tests Written With AI

AI tools now write much of the code, and often the tests too. If the same tool writes the code and the tests, both can share the same wrong idea, and everything stays green.

The bugs in this post show where that matters most: the levels above the unit tests. Their input must not only have the right structure, it must also make sense for the domain. The cancellation deadline needs a workshop in 24 hours and one in seven days. The ownership rule needs two users. The load test needs fewer seats than users, or it never creates a conflict.

The assertions must be right for the domain as well. “Exactly 20 active seats” is a business rule, and no tool can read it from code that might be wrong. Bug 2 is the same: the contract tests catch the generated fake only because someone decided that a stale save must fail.

I built the service for this post with Claude Code, and I had to step in and steer a lot. That is how I split the work now. I decide what goes in and what must come out: the test data, the scenarios and the assertions. The AI tools do the wiring: fixtures, API clients, the code that calls the service, and much of the service code itself.

For each generated test, I ask one question: would it fail if the behaviour were wrong? Planting a bug, as with make bug-<n>, answers it quickly.

Limitations and Trade-offs

• The GSI has a constant partition key. gsi1pk=WORKSHOPS becomes a hot partition with many workshops. Fine for this example, not for a big catalogue.
• A save writes at most 99 changed registrations. DynamoDB allows 100 actions per transaction.
• moto does not cover everything. No IAM, no throttling, no transaction conflicts. It is a fast first check, not a replacement for real DynamoDB.
• The load bug is rare, and I ran it once. Bug 8 produced a single 500 in one run on a fresh stage. I have not measured how often a run misses it. Fewer seats or more users raise the pressure if you want more 500s.
• E2E and load tests also write to prod. Their workshops show up in GET /workshops while a run lasts. The fixtures delete them afterwards, also when a test fails.
• Short-lived stages show no long-term behaviour. A load test against a stage that is ten minutes old says nothing about weeks in production.

Wrapping Up

1. Give every level one job: the bug class it is the cheapest place for.
2. Fakes are fine if a contract suite keeps them honest.
3. Run the same integration tests against moto and AWS. moto checks API calls, AWS your table.
4. Build the test table and the real table from one schema file.
5. Test the HTTP layer separately. The handler maps errors to status codes, not the use cases.
6. Enforce the hexagon with a test. A pyproject.toml prevents nothing in a shared .venv.
7. E2E is the only place where the Lambda’s IAM policy runs. Cover every operation, and check the table, not just the response.
8. Load tests are for invariants. “Exactly 20 seats, no 5xx” is the test, latency a by-product.

If AI writes your tests, decide the test data and the assertions yourself, and let the tool do the wiring.

The service, the eight bug patches and the workflows are on GitHub: python-aws-testing-pyramid.

Much of how I test my serverless projects comes from Yan Cui’s course Testing Serverless Architectures. I recommend it without reservation.

Coming Up in Part 2

Part 2 adds a SqlWorkshopRepository with SQLModel and Aurora Serverless v2, without changing the domain or the unit tests, only the contract fixture and the E2E state reader. It also covers SQLite vs. Postgres in tests, migrations, and what the connection limit does to the load test.

Further Reading

• Serverless Python Microservices on AWS: My Reference Architecture
• Enforcing Python Architecture Rules with PyTestArch
• AWS Lambda Packaging with Python and uv
• Testing Serverless Architectures, the course by Yan Cui
• Amazon DynamoDB Transactions: How it works
• The repository for this post