Module 8: Final Project — TaskFlow API

Complete tests: isolation, integration, benchmarks

Next →

TaskFlow isn't complete without tests. This capsule covers the integration test suite that guarantees the 7 patterns work correctly: aggressive RLS isolation tests (tenant A can't read tenant B's data), tests for CRUD + cursor pagination + soft delete, optimistic locking tests with simulated concurrency, bulk endpoint tests with valid and invalid data, audit log tests verifying the triggers work, and benchmarks measured to include in BENCHMARKS.md.

The tests use real Postgres via testcontainers, not mocks. Slow (~3s setup) but the only way to truly test the patterns.

testcontainers setup

# tests/conftest.py
import pytest
import asyncio
import uuid
from typing import AsyncGenerator
from testcontainers.postgres import PostgresContainer
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy import text
from httpx import AsyncClient
import jwt

from app.main import app
from app.config import settings
from app.database import Base


@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(scope="session")
async def postgres_container():
    """Postgres container for tests."""
    pg = PostgresContainer("postgres:16")
    pg.start()
    yield pg
    pg.stop()


@pytest.fixture(scope="session")
async def engine(postgres_container):
    """Engine pointing at the testcontainer + migrations applied."""
    db_url = postgres_container.get_connection_url(driver="asyncpg")
    settings.database_url = db_url

    engine = create_async_engine(db_url, echo=False)

    # Apply migrations
    from alembic.config import Config
    from alembic import command
    alembic_cfg = Config("alembic.ini")
    alembic_cfg.set_main_option("sqlalchemy.url", db_url)
    command.upgrade(alembic_cfg, "head")

    # Create app_user (no RLS bypass)
    async with engine.begin() as conn:
        await conn.execute(text("""
            DO $$ BEGIN
                CREATE ROLE app_user LOGIN PASSWORD 'app_password';
            EXCEPTION WHEN duplicate_object THEN NULL;
            END $$;
        """))
        await conn.execute(text("GRANT ALL ON ALL TABLES IN SCHEMA public TO app_user"))
        await conn.execute(text("GRANT USAGE ON ALL SEQUENCES IN SCHEMA public TO app_user"))
        await conn.execute(text("GRANT USAGE ON SCHEMA audit TO app_user"))
        await conn.execute(text("GRANT ALL ON ALL TABLES IN SCHEMA audit TO app_user"))

    yield engine
    await engine.dispose()


@pytest.fixture
async def db_session(engine) -> AsyncGenerator[AsyncSession, None]:
    """Session without tenant context — for setup."""
    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
    async with SessionLocal() as session:
        yield session


@pytest.fixture
async def tenant_a_id(db_session):
    """Create tenant A, return the id as a string."""
    result = await db_session.execute(text(
        "INSERT INTO tenants (name) VALUES ('Tenant A') ON CONFLICT (name) DO NOTHING RETURNING id"
    ))
    row = result.first()
    if row:
        await db_session.commit()
        return str(row[0])
    # Already exists, fetch
    result = await db_session.execute(text(
        "SELECT id FROM tenants WHERE name = 'Tenant A'"
    ))
    return str(result.scalar())


@pytest.fixture
async def tenant_b_id(db_session):
    result = await db_session.execute(text(
        "INSERT INTO tenants (name) VALUES ('Tenant B') ON CONFLICT (name) DO NOTHING RETURNING id"
    ))
    row = result.first()
    if row:
        await db_session.commit()
        return str(row[0])
    result = await db_session.execute(text(
        "SELECT id FROM tenants WHERE name = 'Tenant B'"
    ))
    return str(result.scalar())


@pytest.fixture
def tenant_a_token(tenant_a_id):
    payload = {
        "sub": "a@test.com",
        "user_id": str(uuid.uuid4()),
        "tenant_id": tenant_a_id,
    }
    return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)


@pytest.fixture
def tenant_b_token(tenant_b_id):
    payload = {
        "sub": "b@test.com",
        "user_id": str(uuid.uuid4()),
        "tenant_id": tenant_b_id,
    }
    return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)


@pytest.fixture
async def client():
    """HTTP client for tests."""
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac


@pytest.fixture
async def project_a_id(client, tenant_a_token):
    """Create a project for tenant A."""
    response = await client.post(
        "/projects",
        json={"name": "Project A"},
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    return response.json()["id"]


@pytest.fixture
async def project_b_id(client, tenant_b_token):
    response = await client.post(
        "/projects",
        json={"name": "Project B"},
        headers={"Authorization": f"Bearer {tenant_b_token}"}
    )
    return response.json()["id"]

Test 1: RLS isolation (the critical one)

# tests/test_rls_isolation.py
import pytest
from httpx import AsyncClient
from sqlalchemy import text


@pytest.mark.asyncio
async def test_tenant_a_cannot_read_tenant_b_tasks(
    client: AsyncClient,
    tenant_a_token: str,
    tenant_b_token: str,
    project_a_id: str,
    project_b_id: str,
):
    # Tenant A creates a task
    response_a = await client.post(
        "/tasks",
        json={"project_id": project_a_id, "title": "A's Secret Task", "status": "pending"},
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    assert response_a.status_code == 201

    # Tenant B creates a task
    response_b = await client.post(
        "/tasks",
        json={"project_id": project_b_id, "title": "B's Task", "status": "pending"},
        headers={"Authorization": f"Bearer {tenant_b_token}"}
    )
    assert response_b.status_code == 201

    # Tenant A lists — sees only its own
    list_a = await client.get(
        "/tasks",
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    a_titles = [t["title"] for t in list_a.json()["items"]]
    assert "A's Secret Task" in a_titles
    assert "B's Task" not in a_titles

    # Tenant B lists — sees only its own
    list_b = await client.get(
        "/tasks",
        headers={"Authorization": f"Bearer {tenant_b_token}"}
    )
    b_titles = [t["title"] for t in list_b.json()["items"]]
    assert "B's Task" in b_titles
    assert "A's Secret Task" not in b_titles


@pytest.mark.asyncio
async def test_tenant_a_cannot_get_tenant_b_task_directly(
    client, tenant_a_token, tenant_b_token, project_b_id
):
    # Tenant B creates a task
    response_b = await client.post(
        "/tasks",
        json={"project_id": project_b_id, "title": "B's Task", "status": "pending"},
        headers={"Authorization": f"Bearer {tenant_b_token}"}
    )
    task_b_id = response_b.json()["id"]

    # Tenant A tries to read B's task directly
    response = await client.get(
        f"/tasks/{task_b_id}",
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    # RLS makes the row invisible — 404 from A's perspective
    assert response.status_code == 404


@pytest.mark.asyncio
async def test_tenant_a_cannot_update_tenant_b_task(
    client, tenant_a_token, tenant_b_token, project_b_id
):
    # B creates
    response_b = await client.post(
        "/tasks",
        json={"project_id": project_b_id, "title": "B's Task", "status": "pending"},
        headers={"Authorization": f"Bearer {tenant_b_token}"}
    )
    task_b_id = response_b.json()["id"]
    etag = response_b.headers.get("ETag", '"1"')

    # A tries to update
    response = await client.put(
        f"/tasks/{task_b_id}",
        json={"title": "Hacked"},
        headers={"Authorization": f"Bearer {tenant_a_token}", "If-Match": etag}
    )
    assert response.status_code == 404  # RLS makes it not find the row


@pytest.mark.asyncio
async def test_rls_blocks_raw_select_without_filter(engine, tenant_a_id, tenant_b_id):
    """Aggressive test: raw SQL with no WHERE tenant_id — RLS must filter."""
    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

    # Setup: insert as superuser (bypasses RLS)
    async with SessionLocal() as session:
        await session.execute(text("""
            INSERT INTO tasks (id, tenant_id, project_id, title, status, created_at, updated_at, version)
            SELECT gen_random_uuid(), :tid, gen_random_uuid(), 'Tenant A task', 'pending', NOW(), NOW(), 1
        """), {"tid": tenant_a_id})
        await session.execute(text("""
            INSERT INTO tasks (id, tenant_id, project_id, title, status, created_at, updated_at, version)
            SELECT gen_random_uuid(), :tid, gen_random_uuid(), 'Tenant B task', 'pending', NOW(), NOW(), 1
        """), {"tid": tenant_b_id})
        await session.commit()

    # Tenant A: set context, run a "malicious" query with no WHERE
    async with SessionLocal() as session:
        await session.execute(text(f"SET LOCAL app.tenant_id = '{tenant_a_id}'"))

        result = await session.execute(text("SELECT title FROM tasks"))
        rows = result.scalars().all()

        # RLS only allows seeing Tenant A tasks
        assert "Tenant A task" in rows
        assert "Tenant B task" not in rows


@pytest.mark.asyncio
async def test_no_tenant_id_set_blocks_all(engine):
    """Without SET app.tenant_id, no row is visible."""
    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

    async with SessionLocal() as session:
        # NO SET app.tenant_id

        result = await session.execute(text("SELECT title FROM tasks"))
        rows = result.scalars().all()
        # Without tenant_id set, RLS blocks all rows
        assert len(rows) == 0

Test 2: cursor pagination

# tests/test_tasks_pagination.py
@pytest.mark.asyncio
async def test_cursor_pagination_consistency(client, tenant_a_token, project_a_id):
    # Create 25 tasks
    for i in range(25):
        await client.post(
            "/tasks",
            json={"project_id": project_a_id, "title": f"T{i:02d}", "status": "pending"},
            headers={"Authorization": f"Bearer {tenant_a_token}"}
        )

    # Page 1
    p1 = await client.get(
        "/tasks?page_size=10",
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    assert p1.status_code == 200
    body1 = p1.json()
    assert len(body1["items"]) == 10
    assert body1["next_cursor"] is not None

    # Page 2
    p2 = await client.get(
        f"/tasks?page_size=10&cursor={body1['next_cursor']}",
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    body2 = p2.json()
    assert len(body2["items"]) == 10

    # No overlap
    ids1 = {t["id"] for t in body1["items"]}
    ids2 = {t["id"] for t in body2["items"]}
    assert ids1.isdisjoint(ids2)

    # Page 3 (last)
    p3 = await client.get(
        f"/tasks?page_size=10&cursor={body2['next_cursor']}",
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    body3 = p3.json()
    assert len(body3["items"]) == 5
    assert body3["next_cursor"] is None

Test 3: soft delete

# tests/test_tasks_soft_delete.py
@pytest.mark.asyncio
async def test_soft_delete_disappears_from_list(client, tenant_a_token, project_a_id):
    # Create task
    create = await client.post(
        "/tasks",
        json={"project_id": project_a_id, "title": "ToDelete", "status": "pending"},
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    task_id = create.json()["id"]

    # Verify it appears
    list_before = await client.get("/tasks", headers={"Authorization": f"Bearer {tenant_a_token}"})
    assert any(t["id"] == task_id for t in list_before.json()["items"])

    # Soft delete
    delete = await client.delete(
        f"/tasks/{task_id}",
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    assert delete.status_code == 204

    # Verify it disappears from lists
    list_after = await client.get("/tasks", headers={"Authorization": f"Bearer {tenant_a_token}"})
    assert all(t["id"] != task_id for t in list_after.json()["items"])

    # Direct GET: 404
    get_after = await client.get(
        f"/tasks/{task_id}",
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    assert get_after.status_code == 404

Test 4: optimistic locking

# tests/test_tasks_optimistic_lock.py
@pytest.mark.asyncio
async def test_concurrent_update_second_returns_412(client, tenant_a_token, project_a_id):
    # Create
    create = await client.post(
        "/tasks",
        json={"project_id": project_a_id, "title": "T", "status": "pending"},
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    task_id = create.json()["id"]
    initial_etag = create.headers["ETag"]

    # Update 1: success
    update1 = await client.put(
        f"/tasks/{task_id}",
        json={"title": "Updated"},
        headers={"Authorization": f"Bearer {tenant_a_token}", "If-Match": initial_etag}
    )
    assert update1.status_code == 200

    # Update 2 with the old etag: 412
    update2 = await client.put(
        f"/tasks/{task_id}",
        json={"title": "Other"},
        headers={"Authorization": f"Bearer {tenant_a_token}", "If-Match": initial_etag}
    )
    assert update2.status_code == 412
    body = update2.json()["detail"]
    assert body["error"] == "version_mismatch"
    assert body["your_etag"] == initial_etag


@pytest.mark.asyncio
async def test_update_without_if_match_returns_428(client, tenant_a_token, project_a_id):
    create = await client.post(
        "/tasks",
        json={"project_id": project_a_id, "title": "T", "status": "pending"},
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    task_id = create.json()["id"]

    response = await client.put(
        f"/tasks/{task_id}",
        json={"title": "X"},
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    assert response.status_code == 428

Test 5: bulk endpoint

# tests/test_tasks_bulk.py
@pytest.mark.asyncio
async def test_bulk_insert_idempotent(client, tenant_a_token, project_a_id):
    payload = [
        {"project_id": project_a_id, "external_id": f"ext-{i}",
         "title": f"T{i}", "status": "pending", "priority": (i % 5) + 1}
        for i in range(100)
    ]

    # First call: insert
    r1 = await client.post(
        "/tasks/bulk",
        json=payload,
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    assert r1.status_code == 200
    assert r1.json()["inserted"] == 100
    assert r1.json()["updated"] == 0

    # Second call: update
    r2 = await client.post(
        "/tasks/bulk",
        json=payload,
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    assert r2.json()["inserted"] == 0
    assert r2.json()["updated"] == 100


@pytest.mark.asyncio
async def test_bulk_partial_success(client, tenant_a_token, project_a_id):
    payload = [
        {"project_id": project_a_id, "external_id": "ok-1",
         "title": "T1", "status": "pending", "priority": 5},
        {"project_id": project_a_id, "external_id": "bad-1",
         "title": "T2", "status": "INVALID", "priority": 3},
        {"project_id": project_a_id, "external_id": "ok-2",
         "title": "T3", "status": "completed", "priority": 1},
    ]

    response = await client.post(
        "/tasks/bulk",
        json=payload,
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )

    body = response.json()
    assert body["inserted"] == 2
    assert body["skipped"] == 1
    assert len(body["errors"]) == 1
    assert body["errors"][0]["external_id"] == "bad-1"

Test 6: audit log

# tests/test_audit_log.py
@pytest.mark.asyncio
async def test_create_logs_to_audit(client, tenant_a_token, project_a_id):
    # Create
    create = await client.post(
        "/tasks",
        json={"project_id": project_a_id, "title": "T", "status": "pending"},
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    task_id = create.json()["id"]

    # History
    history = await client.get(
        f"/tasks/{task_id}/history",
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    log = history.json()["history"]
    assert len(log) == 1
    assert log[0]["action"] == "INSERT"


@pytest.mark.asyncio
async def test_update_creates_audit_entry(client, tenant_a_token, project_a_id):
    create = await client.post(
        "/tasks",
        json={"project_id": project_a_id, "title": "T", "status": "pending"},
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    task_id = create.json()["id"]
    etag = create.headers["ETag"]

    # Update
    await client.put(
        f"/tasks/{task_id}",
        json={"status": "completed"},
        headers={"Authorization": f"Bearer {tenant_a_token}", "If-Match": etag}
    )

    # History
    history = await client.get(
        f"/tasks/{task_id}/history",
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    log = history.json()["history"]
    assert len(log) == 2  # INSERT + UPDATE
    assert log[0]["action"] == "UPDATE"
    assert log[0]["old_data"]["status"] == "pending"
    assert log[0]["new_data"]["status"] == "completed"

Benchmarks

# benchmarks/bench_pagination.py
import asyncio
import time
import httpx


async def setup_data(token, project_id, n):
    """Create n tasks."""
    async with httpx.AsyncClient(base_url="http://localhost:8000", timeout=120) as client:
        # Bulk
        payload = [
            {"project_id": project_id, "external_id": f"ext-{i}",
             "title": f"Task {i}", "status": "pending", "priority": (i % 5) + 1}
            for i in range(n)
        ]
        response = await client.post(
            "/tasks/bulk",
            json=payload,
            headers={"Authorization": f"Bearer {token}"}
        )
        return response.json()


async def bench_cursor(token, page_size=20):
    """Cursor pagination — should be O(1)."""
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        # Page 1
        start = time.perf_counter()
        r = await client.get(
            f"/tasks?page_size={page_size}",
            headers={"Authorization": f"Bearer {token}"}
        )
        p1_ms = (time.perf_counter() - start) * 1000

        cursor = r.json()["next_cursor"]

        # Iterate to page ~50 (deep)
        for _ in range(50):
            if cursor is None:
                break
            start = time.perf_counter()
            r = await client.get(
                f"/tasks?page_size={page_size}&cursor={cursor}",
                headers={"Authorization": f"Bearer {token}"}
            )
            cursor = r.json()["next_cursor"]
        deep_ms = (time.perf_counter() - start) * 1000

    return {"page_1_ms": p1_ms, "page_50_ms": deep_ms}


async def main():
    # Set up token and project beforehand
    token = "..."
    project_id = "..."

    # Set up 100k tasks
    await setup_data(token, project_id, 100_000)

    # Bench
    cursor_results = await bench_cursor(token, page_size=20)
    print(f"Cursor pagination: page 1 = {cursor_results['page_1_ms']:.1f}ms, "
          f"page 50 = {cursor_results['page_50_ms']:.1f}ms")
    # Expected: both similar (O(1))


asyncio.run(main())

Typical result:

Cursor pagination: page 1 = 8.3ms, page 50 = 9.1ms

Minimal difference — confirms O(1).

Run the tests

# All tests
pytest -v

# Only isolation tests
pytest tests/test_rls_isolation.py -v

# With coverage
pytest --cov=app tests/

# Benchmarks
python benchmarks/bench_pagination.py
python benchmarks/bench_bulk.py

Expected output:

tests/test_rls_isolation.py::test_tenant_a_cannot_read_tenant_b_tasks PASSED
tests/test_rls_isolation.py::test_tenant_a_cannot_get_tenant_b_task_directly PASSED
tests/test_rls_isolation.py::test_tenant_a_cannot_update_tenant_b_task PASSED
tests/test_rls_isolation.py::test_rls_blocks_raw_select_without_filter PASSED
tests/test_rls_isolation.py::test_no_tenant_id_set_blocks_all PASSED
tests/test_tasks_pagination.py::test_cursor_pagination_consistency PASSED
tests/test_tasks_soft_delete.py::test_soft_delete_disappears_from_list PASSED
tests/test_tasks_optimistic_lock.py::test_concurrent_update_second_returns_412 PASSED
tests/test_tasks_optimistic_lock.py::test_update_without_if_match_returns_428 PASSED
tests/test_tasks_bulk.py::test_bulk_insert_idempotent PASSED
tests/test_tasks_bulk.py::test_bulk_partial_success PASSED
tests/test_audit_log.py::test_create_logs_to_audit PASSED
tests/test_audit_log.py::test_update_creates_audit_entry PASSED

13 passed in 12.34s

Pitfalls and common mistakes

1. Tests without real postgres (mocks).

RLS, triggers, COPY aren't mocked. For this module's patterns, testcontainers is mandatory.

2. testcontainers slow to set up.

Each session takes ~3-5s to start postgres. If you have few tests, you feel it. Solution: scope="session" on the fixture, reusable across tests.

3. Tests without cleanup between runs.

If tests insert data and don't roll back, the next run fails on unique constraints. Use the transaction-per-test pattern, or truncate.

4. Hardcoded JWT secret in tests.

Tests should use the same secret as the app. settings.jwt_secret is read from an env var, so setting JWT_SECRET=test in the pytest config works.

5. Tests that depend on order.

pytest -n (parallel) can run tests in a different order. Tests must be independent.

6. Tests that forget to await async.

asyncio_mode = "auto" in pyproject.toml helps. Verify that every async test has await correctly.

7. RLS test that fails because of the superuser.

If your test connects as postgres (superuser), RLS doesn't apply → tests "pass" but security doesn't exist. Verify that tests use app_user.

8. No teardown of testcontainers.

If pytest crashes, the containers stay running. docker ps -a to check and docker rm -f $(docker ps -aq) to clean up.

Summary and next step

What you have now:

  • A complete test suite with real Postgres (testcontainers).
  • 5 aggressive RLS isolation tests.
  • Tests for each pattern: cursor, soft delete, optimistic lock, bulk, audit log.
  • Benchmarks with reproducible numbers.
  • pytest -v with all tests passing.

Commit:

git add .
git commit -m "test: complete integration tests with testcontainers + benchmarks"

In the next capsule we close out the module and the whole guide with the final documentation: BENCHMARKS.md with numbers, MULTITENANCY.md with justification of the architectural decision, RUNBOOK-MIGRATION.md with actionable steps. And the guide's wrap-up with "what's missing for real production" and connections to the path's upcoming guides.

Resources

  1. pytest-asyncio — reference.
  2. testcontainers-python — real Postgres in tests.
  3. httpx — async client — reference.
  4. pytest fixtures — patterns.
  5. GitLab — Test pyramid for backend — testing strategy.

Capsule 07 of 08 — Module 8 — SQL Patterns for Production APIs Guide

← PreviousBulk endpoint + live zero-downtime migrationNext →Final documentation + guide wrap-up