The local-first test ladder

The introduction promised local first, cloud second. This chapter makes that operational: a four-rung ladder every lab and every capstone in this book climbs, the seam in your code that makes the rungs possible, and the exact commands for each rung. Read it before the first cloud lab, and return to it whenever a runbook says "test locally first," because this is what that sentence means.

The motivation is economic and it is about feedback loops. A bug caught by an in-memory test costs a second and zero dollars; the same bug caught by a 500-worker overnight fleet costs a night and a real invoice. Agent systems make this worse than ordinary software, because the expensive part (tokens) is spent per attempt, so a debug loop that runs in the cloud burns money at every iteration. The ladder exists to push each bug to the cheapest rung that can catch it.

The four rungs

The discipline is a rule, not a preference: never climb a rung to debug something the rung below could have caught. If a fleet run fails on a KeyError, that was a rung-1 bug that escaped, and the fix is a rung-1 test, not a more careful cloud run.

The seam that makes it work

Rungs only exist if the same code can run on all of them, and that requires one discipline in how you write workers: receive your clients, do not construct them. A worker that calls boto3.client("s3") inside itself is welded to rung 3. A worker that accepts s3 as an argument runs on every rung unchanged.

# welded to the cloud: untestable below rung 3
def process(doc_id):
    s3 = boto3.client("s3")           # <- constructed inside
    ...

# the seam: identical code on every rung
def process(doc_id, *, s3, ddb, model):
    ...                               # <- received

This is ordinary dependency injection, and it is worth naming because it is the single highest-leverage habit for testable agent systems. It also gives you the agent-specific win: if model is received rather than constructed, you can pass a replaying fake that returns recorded responses, and suddenly a test over a model is deterministic. That is Chapter 10's replay discipline pointed at your test suite.

Lab 0.1: rung 1, and the tests that are the rehearsal

The lab implements a document worker (the Capstone C shape: read from S3, extract with a model, write once to DynamoDB) and runs it against fakes, including a FakeBedrock that replays recorded responses.

python3 local_harness.py            # the narrated demo
python3 -m pytest local_harness.py -q   # the same thing as a test suite

--- rung 1: the worker against in-memory fakes ---
  first delivery : {'doc_id': 'inv-1', 'status': 'ok', 'fields': 3}
  redelivery     : {'doc_id': 'inv-1', 'status': 'duplicate'}
  ddb records    : 1 (one document, one record)
  model calls    : 1 (the duplicate spent no tokens)

--- the assertions that make it a test ---
  PASS test_happy_path
  PASS test_redelivery_is_idempotent
  PASS test_race_is_survivable

--- the same worker, higher rungs (one line changes) ---
  rung 2 (moto)     : with mock_aws(): s3=boto3.client('s3'), ...
  rung 3 (real AWS) : s3=boto3.client('s3'), ddb=boto3.client('dynamodb')

The three tests are chosen deliberately, because they are the three bugs that actually reach production in queue-fed agent systems, and all three are catchable for free:

• test_redelivery_is_idempotent proves the at-least-once contract is survivable: the second delivery writes nothing and spends no tokens. That token assertion is agent-specific and worth copying, a duplicate that re-invokes the model is a correctness bug and a cost bug at once.
• test_race_is_survivable forces the conditional-write path that only occurs when two workers hit the same document in the same instant, a race you cannot reliably reproduce in the cloud but can produce exactly, every time, with a fake.
• test_happy_path is the boring one, and it is what lets you refactor.

Rung 2: emulators, with the commands

Rung 2 runs the real AWS API shapes locally. Four tools cover this book's services; all are follow-along here (none is installed on this box).

moto (in-process Python mocks, the lightest option, and the one that fits the seam above most naturally):

pip install 'moto[all]' boto3 pytest

# test_worker_moto.py
import boto3, pytest
from moto import mock_aws
from local_harness import process_document, FakeBedrock

@mock_aws
def test_against_moto():
    s3 = boto3.client("s3", region_name="us-east-1")
    ddb = boto3.client("dynamodb", region_name="us-east-1")
    s3.create_bucket(Bucket="intake")
    s3.put_object(Bucket="intake", Key="inbox/a.txt", Body=b"Acme invoice A-1 $42")
    ddb.create_table(TableName="results",
        KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"},
                   {"AttributeName": "sk", "KeyType": "RANGE"}],
        AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"},
                              {"AttributeName": "sk", "AttributeType": "S"}],
        BillingMode="PAY_PER_REQUEST")
    out = process_document("inv-1", "inbox/a.txt", s3=s3, ddb=ddb,
                           model=FakeBedrock(['{"vendor":"Acme"}']))
    assert out["status"] == "ok"

Note what changed from rung 1: only the clients. The worker is untouched, which is the seam paying off. (One real difference to know: moto raises the true ClientError with ConditionalCheckFailedException, so catch that at rung 2 and above, not the lab's stand-in exception.)

LocalStack (containerized AWS, when you need services moto covers poorly, or want one endpoint for several services):

docker run --rm -d -p 4566:4566 --name localstack localstack/localstack
export AWS_ENDPOINT_URL=http://localhost:4566
aws --endpoint-url=$AWS_ENDPOINT_URL s3 mb s3://intake
python your_worker.py            # boto3 honors AWS_ENDPOINT_URL
docker stop localstack

DynamoDB Local (when you want just the table, fast and exact):

docker run --rm -d -p 8000:8000 amazon/dynamodb-local
aws dynamodb create-table --endpoint-url http://localhost:8000 \
  --table-name hive-state \
  --attribute-definitions AttributeName=pk,AttributeType=S AttributeName=sk,AttributeType=S \
  --key-schema AttributeName=pk,KeyType=HASH AttributeName=sk,KeyType=RANGE \
  --billing-mode PAY_PER_REQUEST

Step Functions Local (validate a state machine definition, including the distributed map shape, before deploying):

docker run --rm -d -p 8083:8083 amazon/aws-stepfunctions-local
aws stepfunctions --endpoint-url http://localhost:8083 create-state-machine \
  --name hive-local --definition file://statemachine.json \
  --role-arn arn:aws:iam::123456789012:role/DummyRole
aws stepfunctions --endpoint-url http://localhost:8083 start-execution \
  --state-machine-arn arn:aws:states:us-east-1:123456789012:stateMachine:hive-local \
  --input '{"run_id":"local","cap":100000}'

SAM CLI (invoke the Lambda handler in the real Lambda container image, catching packaging and handler-signature bugs):

sam local invoke HiveWorker -e events/one-shard.json

Rung 3: one account, one unit

Rung 3 is the first rung that costs money, so it is deliberately small: one work unit, one document, one question. Its purpose is to catch what no emulator can, real credentials, real IAM, real model behavior, real latency, and real token counts.

# the smallest possible real check: can I reach the model at all?
aws bedrock-runtime converse --model-id anthropic.claude-haiku-4-5 \
  --messages '[{"role":"user","content":[{"text":"reply with OK"}]}]' \
  --region us-east-1

# then one real unit through the real worker
python worker_cli.py --doc-id inv-1 --key inbox/a.txt   # ~cents

Two things to record at rung 3, because they are the inputs to rung 4's planning: the token count per unit (feeds the fleet arithmetic) and the wall clock per unit (feeds Little's law). A rung-3 run whose numbers you did not write down has to be repeated.

The ladder, per capstone

Each runbook maps onto the ladder like this:

Testing the agent-shaped parts

Three things are specific to agents and deserve their own local treatment, each already built earlier in the book:

• The model: replay recorded responses (Chapter 10). A test that calls a live model is not a test, it is a sample.
• The prompt: eval suites with a golden set, run in CI, are the prompt's unit tests; a prompt change with no eval is an unreviewed diff.
• The cost: the cost regression gate belongs at rung 1, where it fails a merge in milliseconds rather than surfacing on a bill.

Don't be confused: fakes vs mocks vs emulators. A fake is a working in-memory implementation (this lab's FakeDDB really stores items), so tests assert on behavior. A mock records calls and asserts they happened, which couples the test to the implementation and breaks on every refactor. An emulator (moto, LocalStack) runs the real API surface locally, catching shape errors a fake would let through. Prefer fakes for logic, emulators for wiring, and reach for mocks rarely, when the call itself is the thing being verified. A suite built mostly of mocks passes while the system is broken, which is the failure mode this ladder exists to prevent.

Full source

"""Lab 0.1: the local-first test harness. Rehearse the cloud on a laptop.

Every cloud lab in this book has a local rehearsal, and they all use one
trick: the worker never CONSTRUCTS its clients, it RECEIVES them. That
single seam lets the identical worker code run against

  rung 1  in-memory fakes        (this file: instant, free, deterministic)
  rung 2  moto / LocalStack      (real AWS API shapes, still local)
  rung 3  one AWS account        (real service, small blast radius)
  rung 4  production             (the runbook)

and the code under test does not change between rungs, which is what
makes a green rung-1 test mean something.

The lab also shows the agent-specific half: a FakeBedrock that REPLAYS
recorded model responses, so the test is deterministic even though the
system under test is a model. That is Part 1's replay discipline
(Chapter 10) applied to testing.

Run it as a script for the narrated demo, or as `pytest local_harness.py`
for the assertions. Standard library only.
"""

from __future__ import annotations

import json


# --- the fakes: in-memory stand-ins with the same call shapes -------------

class FakeDDB:
    """The DynamoDB calls this book actually uses, in a dict."""

    def __init__(self):
        self.items: dict[tuple[str, str], dict] = {}
        self.calls: list[str] = []

    def put_item(self, TableName, Item, ConditionExpression=None):
        self.calls.append("put_item")
        key = (Item["pk"]["S"], Item["sk"]["S"])
        if ConditionExpression == "attribute_not_exists(pk)" and key in self.items:
            raise ConditionalCheckFailed(f"{key} exists")   # the real behavior
        self.items[key] = Item

    def get_item(self, TableName, Key):
        self.calls.append("get_item")
        item = self.items.get((Key["pk"]["S"], Key["sk"]["S"]))
        return {"Item": item} if item else {}


class ConditionalCheckFailed(Exception):
    """Stands in for ddb.exceptions.ConditionalCheckFailedException."""


class FakeS3:
    def __init__(self, objects: dict[str, str]):
        self.objects = objects

    def get_object(self, Bucket, Key):
        return {"Body": _Body(self.objects[Key])}


class _Body:
    def __init__(self, text: str):
        self._text = text

    def read(self) -> bytes:
        return self._text.encode()


class FakeBedrock:
    """Replays recorded model responses: deterministic tests over a model."""

    def __init__(self, responses: list[str]):
        self.responses = list(responses)
        self.calls = 0

    def invoke(self, prompt: str) -> str:
        self.calls += 1
        return self.responses.pop(0) if self.responses else "[]"


# --- the system under test: note it RECEIVES its clients ------------------

def process_document(doc_id: str, key: str, *, s3, ddb, model,
                     bucket="intake") -> dict:
    """Extract fields from a document and record the result exactly once.

    The only thing that changes between rungs is what s3/ddb/model are.
    """
    if "Item" in ddb.get_item(TableName="results",
                              Key={"pk": {"S": doc_id}, "sk": {"S": "result"}}):
        return {"doc_id": doc_id, "status": "duplicate"}   # idempotent

    text = s3.get_object(Bucket=bucket, Key=key)["Body"].read().decode()
    extracted = json.loads(model.invoke(f"Extract fields from: {text}"))
    try:
        ddb.put_item(TableName="results",
                     Item={"pk": {"S": doc_id}, "sk": {"S": "result"},
                           "data": {"S": json.dumps(extracted)}},
                     ConditionExpression="attribute_not_exists(pk)")
    except ConditionalCheckFailed:
        return {"doc_id": doc_id, "status": "raced"}
    return {"doc_id": doc_id, "status": "ok", "fields": len(extracted)}


# --- rung 1: the tests, which are the rehearsal --------------------------

INVOICE = '{"vendor":"Acme","invoice_number":"A-1","total_cents":4200}'


def build():
    return (FakeS3({"inbox/a.txt": "Acme invoice A-1 total $42.00"}),
            FakeDDB(),
            FakeBedrock([INVOICE, INVOICE]))


def test_happy_path():
    s3, ddb, model = build()
    out = process_document("inv-1", "inbox/a.txt", s3=s3, ddb=ddb, model=model)
    assert out["status"] == "ok" and out["fields"] == 3
    assert ("inv-1", "result") in ddb.items


def test_redelivery_is_idempotent():
    """The queue WILL deliver twice; the effect must happen once."""
    s3, ddb, model = build()
    first = process_document("inv-1", "inbox/a.txt", s3=s3, ddb=ddb, model=model)
    second = process_document("inv-1", "inbox/a.txt", s3=s3, ddb=ddb, model=model)
    assert first["status"] == "ok"
    assert second["status"] == "duplicate"
    assert len(ddb.items) == 1          # one document, one record
    assert model.calls == 1             # and the duplicate cost no tokens


def test_race_is_survivable():
    """Two workers, same doc, same instant: the conditional write decides."""
    s3, ddb, model = build()
    ddb.items[("inv-2", "result")] = {"pk": {"S": "inv-2"}, "sk": {"S": "result"}}
    ddb.get_item = lambda **kw: {}      # both workers saw "not done" (a race)
    out = process_document("inv-2", "inbox/a.txt", s3=s3, ddb=ddb, model=model)
    assert out["status"] == "raced"     # lost the race, did not double-write


if __name__ == "__main__":
    print("--- rung 1: the worker against in-memory fakes ---")
    s3, ddb, model = build()
    print(f"  first delivery : {process_document('inv-1', 'inbox/a.txt', s3=s3, ddb=ddb, model=model)}")
    print(f"  redelivery     : {process_document('inv-1', 'inbox/a.txt', s3=s3, ddb=ddb, model=model)}")
    print(f"  ddb records    : {len(ddb.items)} (one document, one record)")
    print(f"  model calls    : {model.calls} (the duplicate spent no tokens)")

    print("\n--- the assertions that make it a test ---")
    for t in (test_happy_path, test_redelivery_is_idempotent,
              test_race_is_survivable):
        t()
        print(f"  PASS {t.__name__}")

    print("\n--- the same worker, higher rungs (one line changes) ---")
    print("  rung 2 (moto)     : with mock_aws(): s3=boto3.client('s3'), ...")
    print("  rung 3 (real AWS) : s3=boto3.client('s3'), ddb=boto3.client('dynamodb')")
    print("  the worker's code is identical on every rung; only the")
    print("  clients differ, which is why a green rung 1 predicts rung 3.")

👉 Next: the production bar, where "production ready" becomes twelve testable gates, several of which this ladder is how you test.