Test & Harden TaskFlow for Production

Objective

Make TaskFlow production-ready by writing a comprehensive test suite and adding production hardening layers. By the end of this lab, you will have at least 15 passing tests and 5 production middleware/features.

This lab has two parts: Testing (60 points) and Production Hardening (40 points).

---

Part 1: Testing (60 points)

1.1 — Test Configuration (tests/conftest.py)

Create the test infrastructure with these fixtures:

• Test database: Create an async SQLAlchemy engine pointing to a taskflow_test database
• Database setup: A session-scoped fixture that creates all tables before tests and drops them after
• Async session: A function-scoped fixture that provides a database session with rollback after each test
• Test client: An httpx.AsyncClient using ASGITransport with the FastAPI app and database dependency override
• Auth helper: A fixture or helper function that registers a user and returns an auth header dict {"Authorization": "Bearer <token>"}

# tests/conftest.py — Required structure
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

from app.main import app
from app.database import Base, get_db

TEST_DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/taskflow_test"

engine_test = create_async_engine(TEST_DATABASE_URL, echo=False)
async_session_test = async_sessionmaker(engine_test, class_=AsyncSession, expire_on_commit=False)

# Implement fixtures: setup_database, db_session, client, auth_headers

Also create a pytest.ini or pyproject.toml section:

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

1.2 — Auth Tests (tests/test_auth.py) — 4+ test cases

Write tests for the authentication flow:

1. Register: POST /api/v1/auth/register with valid data returns 201 and user object (no password in response)
2. Login: POST /api/v1/auth/login with valid credentials returns 200 and access_token
3. Protected route access: GET /api/v1/projects with valid token returns 200
4. Invalid token rejection: GET /api/v1/projects with Authorization: Bearer invalidtoken returns 401

1.3 — CRUD Tests (tests/test_projects.py, tests/test_tasks.py) — 6+ test cases

Write tests for project and task CRUD operations:

1. Create project: POST /api/v1/projects returns 201 with project data
2. List projects: GET /api/v1/projects returns 200 with list of projects for the authenticated user
3. Get project by ID: GET /api/v1/projects/{id} returns 200 with correct project
4. Update project: PATCH /api/v1/projects/{id} returns 200 with updated fields
5. Delete project: DELETE /api/v1/projects/{id} returns 204
6. Create task in project: POST /api/v1/projects/{id}/tasks returns 201 with task data

1.4 — Permission Tests (tests/test_permissions.py) — 3+ test cases

Write tests verifying authorization boundaries:

1. Non-member cannot access project: Register a second user, try to GET a project owned by the first user — expect 403 or 404
2. Non-owner cannot delete project: Second user tries to DELETE first user's project — expect 403
3. Non-member cannot create task: Second user tries to POST a task to first user's project — expect 403 or 404

1.5 — Pagination Tests (tests/test_pagination.py) — 2+ test cases

Write tests for pagination parameters:

1. Default pagination: Create 5 projects, GET /api/v1/projects returns all with total count
2. Custom page/size: GET /api/v1/projects?page=1&size=2 returns exactly 2 items and correct total

Test Requirements Summary

• Minimum 15 test cases total across all test files
• All tests use @pytest.mark.anyio for async
• All tests use the client fixture (no manual app setup)
• Each test is independent (no test depends on another test's side effects)

---

Part 2: Production Hardening (40 points)

2.1 — Redis Caching (app/cache.py) — Cache-Aside Pattern

Implement Redis caching for read-heavy endpoints:

• Create an async Redis client using redis.asyncio.Redis
• Implement get_cached(key) and set_cached(key, value, ttl) helper functions
• Implement invalidate_cache(pattern) to delete keys by glob pattern
• Add caching to GET /api/v1/projects with a 60-second TTL
• Add caching to GET /api/v1/projects/{id}/tasks with a 60-second TTL
• Invalidate relevant cache on any POST, PATCH, or DELETE operation

# app/cache.py — Required interface
from redis.asyncio import Redis

redis_client = Redis(host="localhost", port=6379, db=0, decode_responses=True)

async def get_cached(key: str) -> dict | None: ...
async def set_cached(key: str, value: dict, ttl: int = 60): ...
async def invalidate_cache(pattern: str): ...

2.2 — Global Error Handler Middleware (app/middleware/error_handler.py)

Create middleware that catches all unhandled exceptions and returns a consistent JSON format:

{
  "error": "internal_server_error",
  "message": "An unexpected error occurred",
  "detail": null
}

• Catch all Exception types in the middleware dispatch
• Log the full traceback using Python logging
• In debug mode, include error details in the detail field; in production, set it to null
• Return status code 500 for unhandled exceptions

2.3 — Request Logging Middleware (app/middleware/logging.py)

Create middleware that logs every HTTP request:

• Log: HTTP method, URL path, response status code, and duration in milliseconds
• Use Python's logging module with a named logger (e.g., taskflow.access)
• Format: GET /api/v1/projects status=200 duration=12.3ms
• Measure duration using time.perf_counter()

2.4 — Rate Limiting on Auth Endpoints

Add rate limiting to prevent brute-force login attempts:

• Limit POST /api/v1/auth/login to 5 requests per minute per IP
• Use slowapi with get_remote_address as the key function
• Return a 429 status code with a clear error message when rate limited
• Register the rate limiter error handler in main.py

2.5 — CORS Middleware

Configure CORS for production:

• Allow origins from an environment variable (ALLOWED_ORIGINS, comma-separated)
• Allow methods: GET, POST, PATCH, DELETE, OPTIONS
• Allow headers: Content-Type, Authorization
• Allow credentials: true

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.ALLOWED_ORIGINS.split(","),
    allow_credentials=True,
    allow_methods=["GET", "POST", "PATCH", "DELETE", "OPTIONS"],
    allow_headers=["Content-Type", "Authorization"],
)

2.6 — Health Check Endpoint (GET /health)

Create an endpoint that checks the status of all dependencies:

• Check database connectivity by executing SELECT 1
• Check Redis connectivity by calling redis_client.ping()
• Return 200 with {"api": "healthy", "database": "healthy", "redis": "healthy"} when all checks pass
• Return 503 with unhealthy status for any failed check

---

What to Submit

Your submission should contain 10 file sections in the editor below. Each section begins with a # FILE N: header.

---

Hints

• Use httpx.ASGITransport(app=app) to create the async test client — this is the FastAPI-recommended approach
• Remember to override get_db in your test client fixture so tests use the test database
• For permission tests, create two separate users with two separate auth tokens
• Redis scan_iter with match=pattern is the async-safe way to find keys for invalidation
• Use time.perf_counter() (not time.time()) for accurate duration measurement
• The health check should return 503 (Service Unavailable) if any dependency is down, not 500