Logging & Testing#

uv add structlog

import structlog
import logging
import sys

def setup_logging(debug: bool = False):
    """Configure structlog for the entire application"""

    # Set log level
    log_level = logging.DEBUG if debug else logging.INFO
    logging.basicConfig(
        format="%(message)s",
        stream=sys.stdout,
        level=log_level,
    )

    structlog.configure(
        processors=[
            structlog.stdlib.filter_by_level,
            structlog.stdlib.add_logger_name,
            structlog.stdlib.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.stdlib.PositionalArgumentsFormatter(),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            # JSON in production, pretty-print in dev
            structlog.dev.ConsoleRenderer() if debug
            else structlog.processors.JSONRenderer(),
        ],
        wrapper_class=structlog.stdlib.BoundLogger,
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        cache_logger_on_first_use=True,
    )

import structlog
from logging_config import setup_logging

setup_logging(debug=True)
log = structlog.get_logger()

@app.get("/users/{user_id}")
async def get_user(user_id: int, request: Request):
    # Bind context for all log calls in this request
    bound_log = log.bind(user_id=user_id, method="GET", path="/users")

    bound_log.info("Fetching user")
    user = db.get_user(user_id)

    if not user:
        bound_log.warning("User not found")
        raise HTTPException(404, "User not found")

    bound_log.info("User fetched successfully", email=user.email)
    return user

{"timestamp": "2024-01-15T10:23:45Z", "level": "info", "event": "Fetching user", "user_id": 42}
{"timestamp": "2024-01-15T10:23:45Z", "level": "info", "event": "User fetched", "user_id": 42, "email": "[email protected]"}

uv add --dev pytest pytest-asyncio httpx

from fastapi.testclient import TestClient
from main import app

# TestClient doesn't need a running server
client = TestClient(app)

def test_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello, World!"}

def test_create_task():
    response = client.post("/tasks", json={"title": "Buy groceries"})
    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Buy groceries"
    assert "id" in data

def test_get_nonexistent_task():
    response = client.get("/tasks/fake-id")
    assert response.status_code == 404
    assert "not found" in response.json()["detail"].lower()

pytest                          # run all tests
pytest -v                       # verbose output
pytest tests/test_api.py -v     # specific file
pytest -k "test_create"         # run tests matching pattern

import pytest
from fastapi.testclient import TestClient
from main import app
from config import get_settings, Settings

# Override settings for tests
def get_test_settings():
    return Settings(
        secret_key="test-secret-key-that-is-long-enough-32chars",
        database_url="sqlite:///./test.db",
    )

app.dependency_overrides[get_settings] = get_test_settings

@pytest.fixture
def client():
    return TestClient(app)

@pytest.fixture
def auth_headers(client):
    """Get a valid JWT token for testing"""
    # Create a test user token directly (bypass OAuth flow)
    from jose import jwt
    from datetime import datetime, timedelta

    token = jwt.encode(
        {"sub": "123", "email": "[email protected]", "exp": datetime.utcnow() + timedelta(hours=1)},
        "test-secret-key-that-is-long-enough-32chars",
        algorithm="HS256",
    )
    return {"Authorization": f"Bearer {token}"}

def test_protected_without_auth(client):
    response = client.get("/protected")
    assert response.status_code == 401

def test_protected_with_auth(client, auth_headers):
    response = client.get("/protected", headers=auth_headers)
    assert response.status_code == 200
    assert "Hello" in response.json()["message"]

import pytest
import httpx
from httpx import AsyncClient
from main import app

@pytest.mark.anyio
async def test_async_endpoint():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/")
        assert response.status_code == 200

uv add --dev anyio[trio]

[tool.pytest.ini_options]
anyio_mode = "asyncio"

uv add --dev pytest-cov

# Run tests and measure coverage
pytest --cov=main --cov-report=term-missing

# Generate HTML report
pytest --cov=main --cov-report=html
# → Open htmlcov/index.html in browser

[tool.pytest.ini_options]
addopts = "--cov=. --cov-report=term-missing --cov-fail-under=80"

from fastapi.testclient import TestClient
from main import app

def test_websocket_echo():
    client = TestClient(app)
    with client.websocket_connect("/ws") as ws:
        ws.send_text("hello")
        data = ws.receive_text()
        assert data == "Echo: hello"