Testing

# app/main.py
from fastapi import FastAPI

app = FastAPI(title="Bookshelf API")


@app.get("/")
async def read_root() -> dict[str, str]:
    return {"name": "Bookshelf API"}


@app.get("/books/{book_id}")
def read_book(book_id: int) -> dict[str, int | str]:
    return {"id": book_id, "title": "The Left Hand of Darkness"}

# tests/test_main.py
from fastapi.testclient import TestClient

from app.main import app

client = TestClient(app)


def test_read_root() -> None:
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"name": "Bookshelf API"}


def test_read_book_validates_the_path_parameter() -> None:
    assert client.get("/books/3").json()["id"] == 3
    response = client.get("/books/three")
    assert response.status_code == 422
    assert response.json()["detail"][0]["type"] == "int_parsing"

pytest -q

..                                                                       [100%]
=============================== warnings summary ===============================
../../../venv/lib/python3.13/site-packages/fastapi/testclient.py:1
  .venv/lib/python3.13/site-packages/fastapi/testclient.py:1: StarletteDeprecationWarning: Using `httpx` with `starlette.testclient` is deprecated; install `httpx2` instead.
    from starlette.testclient import TestClient as TestClient  # noqa

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
2 passed, 1 warning in 0.80s

bookshelf/
├── pyproject.toml
├── app/
│   ├── __init__.py
│   ├── config.py        # pydantic-settings, cached get_settings()
│   ├── db.py            # engine, get_session dependency
│   ├── models.py        # SQLModel tables and request models
│   ├── security.py      # OAuth2 password flow, JWT, current user
│   ├── service.py       # pure business logic
│   ├── isbn.py          # client for an external service
│   ├── rooms.py         # WebSocket rooms
│   └── main.py          # the FastAPI app and its path operations
└── tests/
    ├── conftest.py      # fixtures: settings, session, client, auth
    ├── test_books.py
    ├── test_auth.py
    ├── test_overrides.py
    ├── test_lifespan.py
    ├── test_websockets.py
    ├── test_review_feed.py
    ├── test_async.py
    ├── test_database.py
    ├── test_service.py
    └── ...

[tool.pytest.ini_options]
testpaths = ["tests"]   # where pytest looks when run without arguments
pythonpath = ["."]      # the project root is importable: "from app.main import app"

pytest                                                # everything under testpaths
pytest tests/test_auth.py                             # one file
pytest tests/test_books.py::test_unknown_book_is_404  # one test, by node id
pytest -k "expired or wrong" tests/test_auth.py       # tests whose name matches an expression
pytest -x --lf                                        # stop at the first failure, rerun the last failures
pytest -v tests/test_auth.py                          # one line per test

tests/test_auth.py::test_login_returns_a_token_that_works PASSED         [ 16%]
tests/test_auth.py::test_wrong_password_is_401 PASSED                    [ 33%]
tests/test_auth.py::test_missing_token_is_401 PASSED                     [ 50%]
tests/test_auth.py::test_expired_token_is_401 PASSED                     [ 66%]
tests/test_auth.py::test_token_signed_with_another_key_is_401 PASSED     [ 83%]
tests/test_auth.py::test_override_the_user_instead_of_logging_in PASSED  [100%]
============================== 6 passed in 0.21s ===============================

..                                                                       [100%]
2 passed, 4 deselected in 0.13s

from functools import lru_cache

from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="BOOKSHELF_", env_file=".env")

    database_url: str = "sqlite:///bookshelf.db"
    jwt_secret: SecretStr
    token_minutes: int = 30


@lru_cache
def get_settings() -> Settings:
    return Settings()

from sqlmodel import Field, SQLModel


class AuthorCreate(SQLModel):
    name: str = Field(min_length=1, max_length=100)


class Author(AuthorCreate, table=True):
    id: int | None = Field(default=None, primary_key=True)


class BookCreate(SQLModel):
    title: str = Field(min_length=1, max_length=200)
    author_id: int
    year: int | None = Field(default=None, ge=1400, le=2100)


class Book(BookCreate, table=True):
    id: int | None = Field(default=None, primary_key=True)
    author_id: int = Field(foreign_key="author.id")


class ReviewCreate(SQLModel):
    rating: int = Field(ge=1, le=5)
    text: str = Field(default="", max_length=2000)


class Review(ReviewCreate, table=True):
    id: int | None = Field(default=None, primary_key=True)
    book_id: int = Field(foreign_key="book.id")
    reviewer: str

from collections.abc import Iterator
from typing import Annotated

from fastapi import Depends
from sqlmodel import Session, create_engine

from app.config import get_settings

_engine = None


def get_engine():
    global _engine
    if _engine is None:
        _engine = create_engine(get_settings().database_url)
    return _engine


def get_session() -> Iterator[Session]:
    with Session(get_engine()) as session:
        yield session


SessionDep = Annotated[Session, Depends(get_session)]

from datetime import UTC, datetime, timedelta
from typing import Annotated

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from pwdlib import PasswordHash

from app.config import Settings, get_settings

password_hash = PasswordHash.recommended()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# username -> password hash; a real application reads these from the database
USERS = {"ada": password_hash.hash("ada-test-password")}


def create_access_token(username: str, settings: Settings) -> str:
    expires = datetime.now(UTC) + timedelta(minutes=settings.token_minutes)
    return jwt.encode(
        {"sub": username, "exp": expires},
        settings.jwt_secret.get_secret_value(),
        algorithm="HS256",
    )


def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    settings: Annotated[Settings, Depends(get_settings)],
) -> str:
    try:
        payload = jwt.decode(
            token, settings.jwt_secret.get_secret_value(), algorithms=["HS256"]
        )
    except jwt.InvalidTokenError:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            "Invalid or expired token",
            headers={"WWW-Authenticate": "Bearer"},
        ) from None
    return payload["sub"]


CurrentUser = Annotated[str, Depends(get_current_user)]

# app/service.py
def rating_summary(ratings: list[int]) -> dict[str, float | int | None]:
    """Pure business logic: no FastAPI, no database."""
    if not ratings:
        return {"count": 0, "average": None}
    return {"count": len(ratings), "average": round(sum(ratings) / len(ratings), 2)}

# app/isbn.py
import httpx


class IsbnLookup:
    """Talks to an external service: slow, rate limited, not for unit tests."""

    def __init__(self, base_url: str = "https://isbn.example.com") -> None:
        self.base_url = base_url

    def title_for(self, isbn: str) -> str | None:
        response = httpx.get(f"{self.base_url}/isbn/{isbn}", timeout=5)
        return response.json().get("title") if response.status_code == 200 else None


def get_isbn_lookup() -> IsbnLookup:
    return IsbnLookup()

# app/rooms.py
from collections import defaultdict

from fastapi import WebSocket


class Rooms:
    """Which sockets watch which book."""

    def __init__(self) -> None:
        self.by_book: dict[int, set[WebSocket]] = defaultdict(set)

    async def join(self, book_id: int, websocket: WebSocket) -> None:
        await websocket.accept()
        self.by_book[book_id].add(websocket)

    def leave(self, book_id: int, websocket: WebSocket) -> None:
        self.by_book[book_id].discard(websocket)

    async def broadcast(self, book_id: int, message: dict) -> None:
        for websocket in list(self.by_book[book_id]):
            await websocket.send_json(message)


rooms = Rooms()

from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import (
    Request,
    BackgroundTasks,
    Depends,
    FastAPI,
    HTTPException,
    WebSocket,
    WebSocketDisconnect,
    status,
)
from fastapi.security import OAuth2PasswordRequestForm
from sqlmodel import select

from app.config import Settings, get_settings
from app.db import SessionDep
from app.isbn import IsbnLookup, get_isbn_lookup
from app.models import Author, AuthorCreate, Book, BookCreate, Review, ReviewCreate
from app.rooms import rooms
from app.security import USERS, CurrentUser, create_access_token, password_hash
from app.service import rating_summary

audit_log: list[str] = []


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.genres = ["fiction", "history", "science"]  # startup: warm a cache
    yield
    app.state.genres = []  # shutdown: release it


app = FastAPI(title="Bookshelf API", lifespan=lifespan)


@app.get("/genres")
def list_genres(request: Request) -> list[str]:
    return request.app.state.genres


@app.post("/token")
def login(form: Annotated[OAuth2PasswordRequestForm, Depends()],
          settings: Annotated[Settings, Depends(get_settings)]) -> dict[str, str]:
    stored = USERS.get(form.username)
    if stored is None or not password_hash.verify(form.password, stored):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Incorrect username or password",
                            headers={"WWW-Authenticate": "Bearer"})
    return {"access_token": create_access_token(form.username, settings),
            "token_type": "bearer"}


@app.post("/authors", status_code=status.HTTP_201_CREATED)
def create_author(data: AuthorCreate, session: SessionDep, user: CurrentUser) -> Author:
    author = Author.model_validate(data)
    session.add(author)
    session.commit()
    session.refresh(author)
    return author


@app.post("/books", status_code=status.HTTP_201_CREATED)
def create_book(data: BookCreate, session: SessionDep, user: CurrentUser,
                tasks: BackgroundTasks) -> Book:
    if session.get(Author, data.author_id) is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Author not found")
    book = Book.model_validate(data)
    session.add(book)
    session.commit()
    session.refresh(book)
    tasks.add_task(audit_log.append, f"{user} created book {book.id}")
    return book


@app.get("/books")
def list_books(session: SessionDep, limit: int = 20, offset: int = 0) -> list[Book]:
    return list(session.exec(select(Book).offset(offset).limit(limit)))


@app.get("/books/{book_id}")
def read_book(book_id: int, session: SessionDep) -> Book:
    book = session.get(Book, book_id)
    if book is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Book not found")
    return book


@app.post("/books/{book_id}/reviews", status_code=status.HTTP_201_CREATED)
def add_review(book_id: int, data: ReviewCreate, session: SessionDep,
               user: CurrentUser, tasks: BackgroundTasks) -> Review:
    if session.get(Book, book_id) is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Book not found")
    review = Review.model_validate(data, update={"book_id": book_id, "reviewer": user})
    session.add(review)
    session.commit()
    session.refresh(review)
    tasks.add_task(rooms.broadcast, book_id, {"rating": review.rating, "reviewer": user})
    return review


@app.get("/books/{book_id}/rating")
def book_rating(book_id: int, session: SessionDep) -> dict:
    ratings = list(session.exec(select(Review.rating).where(Review.book_id == book_id)))
    return rating_summary(ratings)


@app.get("/isbn/{isbn}")
def title_by_isbn(isbn: str, lookup: Annotated[IsbnLookup, Depends(get_isbn_lookup)]) -> dict:
    title = lookup.title_for(isbn)
    if title is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Unknown ISBN")
    return {"isbn": isbn, "title": title}


@app.get("/boom")
def boom() -> None:
    raise RuntimeError("unexpected failure")


@app.websocket("/ws/books/{book_id}")
async def book_updates(websocket: WebSocket, book_id: int):
    await websocket.accept()
    try:
        while True:
            text = await websocket.receive_text()
            await websocket.send_json({"book_id": book_id, "echo": text})
    except WebSocketDisconnect:
        pass


@app.websocket("/ws/books/{book_id}/reviews")
async def review_feed(websocket: WebSocket, book_id: int):
    await rooms.join(book_id, websocket)
    try:
        while True:
            await websocket.receive_text()  # keep the socket open until the client leaves
    except WebSocketDisconnect:
        rooms.leave(book_id, websocket)

import secrets
from collections.abc import Iterator

import pytest
from fastapi.testclient import TestClient
from sqlalchemy.pool import StaticPool
from sqlmodel import Session, SQLModel, create_engine

from app.config import Settings, get_settings
from app.db import get_session
from app.main import app
from app.security import create_access_token


@pytest.fixture(name="settings")
def settings_fixture() -> Settings:
    # Built from explicit values: no .env file, no real environment, a throwaway key.
    return Settings(_env_file=None, jwt_secret=secrets.token_hex(32), database_url="sqlite://")


@pytest.fixture(name="session")
def session_fixture() -> Iterator[Session]:
    engine = create_engine(
        "sqlite://",  # in-memory database
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,  # one connection shared by every thread
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session
    engine.dispose()  # close the shared connection explicitly


@pytest.fixture(name="client")
def client_fixture(session: Session, settings: Settings) -> Iterator[TestClient]:
    app.dependency_overrides[get_session] = lambda: session
    app.dependency_overrides[get_settings] = lambda: settings
    with TestClient(app) as client:  # also runs the lifespan
        yield client
    app.dependency_overrides.clear()  # never leak overrides into other tests


@pytest.fixture(name="auth")
def auth_fixture(settings: Settings) -> dict[str, str]:
    token = create_access_token("ada", settings)
    return {"Authorization": f"Bearer {token}"}

pytest -q --setup-show tests/test_books.py::test_unknown_book_is_404

        SETUP    F session
        SETUP    F settings
        SETUP    F client (fixtures used: session, settings)
        TEARDOWN F client
        TEARDOWN F settings
        TEARDOWN F session
1 passed in 0.12s

from fastapi.testclient import TestClient


def test_create_and_read_book(client: TestClient, auth: dict[str, str]) -> None:
    author = client.post("/authors", json={"name": "Ursula K. Le Guin"}, headers=auth)
    assert author.status_code == 201

    created = client.post(
        "/books",
        json={"title": "The Dispossessed", "author_id": author.json()["id"], "year": 1974},
        headers=auth,
    )
    assert created.status_code == 201
    book = created.json()
    assert book["title"] == "The Dispossessed"

    fetched = client.get(f"/books/{book['id']}")
    assert fetched.status_code == 200
    assert fetched.json() == book


def test_unknown_book_is_404(client: TestClient) -> None:
    response = client.get("/books/999")
    assert response.status_code == 404
    assert response.json() == {"detail": "Book not found"}


def test_validation_error_is_422(client: TestClient, auth: dict[str, str]) -> None:
    response = client.post("/books", json={"title": "", "author_id": 1}, headers=auth)
    assert response.status_code == 422
    error = response.json()["detail"][0]
    assert error["loc"] == ["body", "title"]
    assert error["type"] == "string_too_short"

from unittest.mock import patch

import pytest
from fastapi.testclient import TestClient

from app.isbn import IsbnLookup, get_isbn_lookup
from app.main import app


class FakeIsbnLookup:
    def __init__(self, titles: dict[str, str]) -> None:
        self.titles = titles
        self.calls: list[str] = []

    def title_for(self, isbn: str) -> str | None:
        self.calls.append(isbn)
        return self.titles.get(isbn)


@pytest.fixture(name="isbn")
def isbn_fixture(client: TestClient) -> FakeIsbnLookup:
    fake = FakeIsbnLookup({"9780441013593": "Dune"})
    app.dependency_overrides[get_isbn_lookup] = lambda: fake
    return fake  # the client fixture clears every override at teardown


def test_known_isbn(client: TestClient, isbn: FakeIsbnLookup) -> None:
    response = client.get("/isbn/9780441013593")
    assert response.json() == {"isbn": "9780441013593", "title": "Dune"}
    assert isbn.calls == ["9780441013593"]


def test_unknown_isbn(client: TestClient, isbn: FakeIsbnLookup) -> None:
    assert client.get("/isbn/0000000000").status_code == 404


def test_same_thing_with_mock_patch(client: TestClient) -> None:
    # patch() replaces an attribute of the class: it works, but it ties the test to
    # where the code looks the name up. The override ties it to the public seam instead.
    with patch.object(IsbnLookup, "title_for", return_value="Kindred"):
        response = client.get("/isbn/123")
    assert response.json()["title"] == "Kindred"

from typing import Annotated

from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient


def current_user() -> str:
    raise RuntimeError("would call the real identity provider")


main = FastAPI()
admin = FastAPI()


@admin.get("/whoami")
def whoami(user: Annotated[str, Depends(current_user)]) -> str:
    return user


main.mount("/admin", admin)


def test_overrides_are_per_application() -> None:
    main.dependency_overrides[current_user] = lambda: "grace"  # wrong application
    client = TestClient(main)
    try:
        client.get("/admin/whoami")
    except RuntimeError as error:
        assert "real identity provider" in str(error)  # the override was not used

    admin.dependency_overrides[current_user] = lambda: "grace"  # the mounted app owns it
    assert client.get("/admin/whoami").json() == "grace"
    main.dependency_overrides.clear()
    admin.dependency_overrides.clear()

import jwt
from fastapi.testclient import TestClient

from app.config import Settings
from app.security import create_access_token, get_current_user


def test_login_returns_a_token_that_works(client: TestClient, settings: Settings) -> None:
    response = client.post("/token", data={"username": "ada", "password": "ada-test-password"})
    assert response.status_code == 200
    token = response.json()["access_token"]
    payload = jwt.decode(token, settings.jwt_secret.get_secret_value(), algorithms=["HS256"])
    assert payload["sub"] == "ada"

    created = client.post(
        "/authors", json={"name": "Octavia E. Butler"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert created.status_code == 201


def test_wrong_password_is_401(client: TestClient) -> None:
    response = client.post("/token", data={"username": "ada", "password": "wrong"})
    assert response.status_code == 401
    assert response.headers["www-authenticate"] == "Bearer"


def test_missing_token_is_401(client: TestClient) -> None:
    response = client.post("/authors", json={"name": "Anonymous"})
    assert response.status_code == 401
    assert response.json() == {"detail": "Not authenticated"}


def test_expired_token_is_401(client: TestClient, settings: Settings) -> None:
    expired = create_access_token("ada", settings.model_copy(update={"token_minutes": -1}))
    response = client.post(
        "/authors", json={"name": "Late"}, headers={"Authorization": f"Bearer {expired}"}
    )
    assert response.status_code == 401
    assert response.json() == {"detail": "Invalid or expired token"}


def test_token_signed_with_another_key_is_401(client: TestClient) -> None:
    forged = jwt.encode({"sub": "ada"}, "some-other-key-0123456789-0123456789", algorithm="HS256")
    response = client.get("/books", headers={"Authorization": f"Bearer {forged}"})
    assert response.status_code == 200  # /books is public: the header is simply ignored
    response = client.post(
        "/authors", json={"name": "Mallory"}, headers={"Authorization": f"Bearer {forged}"}
    )
    assert response.status_code == 401


def test_override_the_user_instead_of_logging_in(client: TestClient) -> None:
    from app.main import app

    app.dependency_overrides[get_current_user] = lambda: "grace"
    response = client.post("/authors", json={"name": "Anyone"})  # no Authorization header
    assert response.status_code == 201

from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.main import app

events: list[str] = []


@asynccontextmanager
async def recording_lifespan(app: FastAPI):
    events.append("startup")
    yield
    events.append("shutdown")


probe = FastAPI(lifespan=recording_lifespan)


@probe.get("/ping")
def ping() -> str:
    return "pong"


def test_lifespan_runs_only_inside_with() -> None:
    events.clear()

    client = TestClient(probe)  # constructed, never entered
    assert client.get("/ping").json() == "pong"
    assert events == []  # the request worked, the lifespan did not run

    with TestClient(probe) as entered:
        assert events == ["startup"]
        assert entered.get("/ping").json() == "pong"
    assert events == ["startup", "shutdown"]


def test_the_real_lifespan_warms_the_cache() -> None:
    with TestClient(app) as client:
        assert client.get("/genres").json() == ["fiction", "history", "science"]
    assert app.state.genres == []  # leaving the block ran the shutdown code

from fastapi.testclient import TestClient

from app.main import app


def test_echo() -> None:
    client = TestClient(app)
    with client.websocket_connect("/ws/books/7") as websocket:
        websocket.send_text("hello")
        assert websocket.receive_json() == {"book_id": 7, "echo": "hello"}

from fastapi.testclient import TestClient

from app.models import Author, Book


def test_every_open_socket_gets_the_new_review(client: TestClient, session, auth) -> None:
    author = Author(name="N. K. Jemisin")
    session.add(author)
    session.commit()
    book = Book(title="The Fifth Season", author_id=author.id)
    session.add(book)
    session.commit()

    # `client` was created with `with TestClient(app)`, so both connections share one
    # event loop and one portal. Without that, every websocket_connect() would start its own.
    with client.websocket_connect(f"/ws/books/{book.id}/reviews") as first:
        with client.websocket_connect(f"/ws/books/{book.id}/reviews") as second:
            response = client.post(
                f"/books/{book.id}/reviews", json={"rating": 5, "text": "Superb"}, headers=auth
            )
            assert response.status_code == 201
            assert first.receive_json() == {"rating": 5, "reviewer": "ada"}
            assert second.receive_json() == {"rating": 5, "reviewer": "ada"}

from fastapi.testclient import TestClient

from app.main import audit_log
from app.models import Author


def test_background_task_has_run_when_the_response_returns(client: TestClient, session, auth) -> None:
    audit_log.clear()
    author = Author(name="Terry Pratchett")
    session.add(author)
    session.commit()

    response = client.post("/books", json={"title": "Small Gods", "author_id": author.id}, headers=auth)

    assert response.status_code == 201
    assert audit_log == [f"ada created book {response.json()['id']}"]

import pytest
from httpx import ASGITransport, AsyncClient
from sqlmodel import Session, select

from app.main import app
from app.models import Author


@pytest.fixture
def anyio_backend() -> str:
    return "asyncio"  # run only on asyncio; add "trio" here if you support it


@pytest.mark.anyio
async def test_create_author_and_check_the_database(client, session: Session, auth) -> None:
    # `client` (the sync fixture) is only used here for its dependency overrides.
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
        response = await ac.post("/authors", json={"name": "Chinua Achebe"}, headers=auth)
    assert response.status_code == 201

    names = [author.name for author in session.exec(select(Author))]
    assert names == ["Chinua Achebe"]

import pytest
from asgi_lifespan import LifespanManager
from httpx import ASGITransport, AsyncClient

from app.main import app


@pytest.fixture
def anyio_backend() -> str:
    return "asyncio"


@pytest.mark.anyio
async def test_asyncclient_does_not_run_the_lifespan() -> None:
    app.state.genres = None
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
        assert app.state.genres is None  # no startup happened


@pytest.mark.anyio
async def test_lifespan_manager_runs_it() -> None:
    async with LifespanManager(app):
        async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
            response = await ac.get("/genres")
    assert response.json() == ["fiction", "history", "science"]
    assert app.state.genres == []  # shutdown ran when the manager exited

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import event
from sqlalchemy.exc import IntegrityError
from sqlalchemy.pool import StaticPool
from sqlmodel import Session, SQLModel, create_engine, select

from app.models import Author, Book, Review


def test_the_api_wrote_the_rows_we_expect(client: TestClient, session: Session, auth) -> None:
    client.post("/authors", json={"name": "Italo Calvino"}, headers=auth)
    author = session.exec(select(Author)).one()  # same session, same in-memory database
    client.post("/books", json={"title": "Invisible Cities", "author_id": author.id}, headers=auth)

    books = session.exec(select(Book)).all()
    assert [book.title for book in books] == ["Invisible Cities"]


def test_every_test_starts_from_an_empty_database(session: Session) -> None:
    assert session.exec(select(Book)).all() == []  # nothing left over from the test above


def test_sqlite_ignores_foreign_keys_by_default(session: Session) -> None:
    session.add(Review(book_id=999, rating=3, reviewer="ada"))  # book 999 does not exist
    session.commit()  # accepted: SQLite does not enforce foreign keys unless asked to


def test_enforcing_foreign_keys_like_production_does() -> None:
    engine = create_engine("sqlite://", connect_args={"check_same_thread": False},
                           poolclass=StaticPool)

    @event.listens_for(engine, "connect")
    def enforce_foreign_keys(dbapi_connection, _record) -> None:
        dbapi_connection.execute("PRAGMA foreign_keys=ON")

    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        session.add(Review(book_id=999, rating=3, reviewer="ada"))
        with pytest.raises(IntegrityError):
            session.commit()
    engine.dispose()

import pytest

from app.service import rating_summary


@pytest.mark.parametrize(
    ("ratings", "expected"),
    [
        ([], {"count": 0, "average": None}),
        ([5], {"count": 1, "average": 5.0}),
        ([5, 4, 4], {"count": 3, "average": 4.33}),
    ],
)
def test_rating_summary(ratings: list[int], expected: dict) -> None:
    assert rating_summary(ratings) == expected

..                                                                       [100%]
=============================== warnings summary ===============================
../../../venv/lib/python3.13/site-packages/fastapi/testclient.py:1
  .venv/lib/python3.13/site-packages/fastapi/testclient.py:1: StarletteDeprecationWarning: Using `httpx` with `starlette.testclient` is deprecated; install `httpx2` instead.
    from starlette.testclient import TestClient as TestClient  # noqa

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
2 passed, 1 warning in 0.80s

pip install httpx2
pytest -q -W error

....................................                                     [100%]
36 passed in 0.34s

import pytest
from fastapi.testclient import TestClient
from httpx2 import ASGITransport, AsyncClient

from app.main import app


def test_sync_client_without_a_warning() -> None:
    with TestClient(app) as client:
        assert client.get("/genres").json() == ["fiction", "history", "science"]


@pytest.fixture
def anyio_backend() -> str:
    return "asyncio"


@pytest.mark.anyio
async def test_async_client_from_httpx2() -> None:
    async with app.router.lifespan_context(app):  # starts and stops the lifespan, no extra package
        async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
            response = await ac.get("/genres")
    assert response.json() == ["fiction", "history", "science"]

[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["."]
filterwarnings = [
    "error",  # any other warning fails the run
    "ignore:Using `httpx` with `starlette.testclient` is deprecated:starlette.exceptions.StarletteDeprecationWarning",
]

..                                                                       [100%]
2 passed in 0.78s

.....................................                                    [100%]
37 passed in 0.31s

==================================== ERRORS ====================================
_____________________ ERROR collecting tests/test_main.py ______________________
E   ModuleNotFoundError: No module named 'httpx2'
E   starlette.exceptions.StarletteDeprecationWarning: Using `httpx` with `starlette.testclient` is deprecated; install `httpx2` instead.
ERROR tests/test_main.py - starlette.exceptions.StarletteDeprecationWarning: ...
!!!!!!!!!!!!!!!!!!!! Interrupted: 1 error during collection !!!!!!!!!!!!!!!!!!!!
1 error in 0.73s

def test_root_path_is_not_prepended_to_the_url() -> None:
    client = TestClient(app, root_path="/api/v1")
    response = client.get("/new")
    assert response.url == "http://testserver/new"  # the request URL has no prefix
    body = response.json()
    assert body["root_path"] == "/api/v1"  # but the ASGI scope carries it ...
    assert body["url_for"] == "http://testserver/api/v1/new"  # ... so generated URLs include it

[tool.coverage.run]
source = ["app"]   # measure the application package, not the tests
branch = true      # also report branches never taken

[tool.coverage.report]
show_missing = true
fail_under = 90    # the run fails below this total percentage

pip install pytest-cov
pytest --cov --cov-report=term-missing

.....................................                                    [100%]
================================ tests coverage ================================
______________ coverage: platform darwin, python 3.13.16-final-0 _______________

Name              Stmts   Miss Branch BrPart  Cover   Missing
-------------------------------------------------------------
app/__init__.py       0      0      0      0   100%
app/config.py        11      1      0      0    91%   17
app/db.py            14      5      2      0    56%   14-16, 20-21
app/isbn.py           9      2      0      0    78%   11-12
app/main.py          94      4     10      2    94%   67, 93, 104-105
app/models.py        19      0      0      0   100%
app/rooms.py         14      0      2      0   100%
app/security.py      20      0      0      0   100%
app/service.py        4      0      2      0   100%
-------------------------------------------------------------
TOTAL               185     12     16      2    92%
Required test coverage of 90.0% reached. Total coverage: 92.04%
37 passed in 0.44s

pytest --cov --cov-fail-under=95

TOTAL               185     12     16      2    92%
FAIL Required test coverage of 95% not reached. Total coverage: 92.04%
37 passed in 0.43s
exit code: 1