FastAPI Production

Checklist/Docs/Testing workers and async jobs

Guide 10

Testing workers and async jobs

Audience: Engineers writing CI for API + workers. Goal: Confidence without flaky full-stack broker tests on every PR.

TL;DR pyramid

text
Domain / handler unit tests (many, fast)
API enqueue tests with fakes (many)
Contract tests for messages (some)
Integration with real broker (few, CI job or nightly)
Staging smoke e2e (per deploy)

Contents

  1. What to test
  2. Unit: handlers
  3. API: enqueue
  4. Idempotency tests
  5. Integration
  6. Staging smoke
  7. Fixtures and fakes
  8. CI layout

---

1. What to test

LayerAssert
DomainInvariants (pay twice, void paid)
HandlerStatus transitions, retries classification
API202, job row, publish called with routing key
ConsumerInbox prevents double side effect
SmokeOne real job completes end-to-end

2. Unit: handlers

python
# PSEUDOCODE
def test_export_skips_if_already_succeeded():
 job = Job(status="succeeded")
 db = FakeDb(job)
 handle_export(job.id, db=db, bus=FakeBus())
 assert db.export_calls == 0

def test_transient_error_bumps_attempt():
 ...

No real RabbitMQ required.

3. API: enqueue

python
# PSEUDOCODE
def test_post_export_returns_202_and_publishes(client, fake_bus):
 r = client.post("/exports", json={...}, headers={"Idempotency-Key": "k1"})
 assert r.status_code == 202
 assert fake_bus.messages[0]["queue"] == "jobs.heavy"
 assert db.jobs[0].status == "pending"

def test_duplicate_idempotency_no_second_publish(client, fake_bus):
 client.post(..., headers={"Idempotency-Key": "k1"})
 client.post(..., headers={"Idempotency-Key": "k1"})
 assert len(fake_bus.messages) == 1

4. Idempotency tests

  • Double delivery of same event_id → one email
  • try_mark_running race → only one worker proceeds
  • Outbox relay publishes each row once

5. Integration

Use Testcontainers or compose in CI:

  • Postgres + RabbitMQ
  • Run worker process
  • Enqueue via API
  • Poll job until succeeded (timeout)

Mark as optional/slow job if PR feedback must stay fast.

6. Staging smoke

text
POST /exports → poll GET /jobs/{id} → expect succeeded within SLO

Run after every deploy. Alert on failure.

7. Fixtures and fakes

python
# PSEUDOCODE
class FakeBus:
 def __init__(self):
 self.messages = []
 async def publish(self, queue, payload):
 self.messages.append({"queue": queue, "payload": payload})

Override Depends(get_bus) in FastAPI tests.

8. CI layout

text
lint → typecheck → unit+api → (integration optional) → build image

Never require Flower or production broker credentials for unit tests.

Related: 09 Errors · 03 Lifecycle

← All docs