পেশাদার API টেস্ট করা

                 ▲  slow, brittle, expensive
                ╱ ╲
               ╱E2E╲          a few: the deployed system, real browser or client
              ╱─────╲
             ╱ integ-╲        some: app + real DB / real HTTP / real provider sandbox
            ╱ ration  ╲
           ╱───────────╲
          ╱  API tests  ╲     many: the whole app through TestClient, outside world faked
         ╱───────────────╲
        ╱   unit tests    ╲   most: pure functions and classes, no I/O at all
       ╱───────────────────╲
                 ▼  fast, stable, cheap

mkdir -p ~/projects/course-reviews && cd ~/projects/course-reviews
uv init -q --bare --name course-reviews
uv add -q "fastapi[standard]" sqlalchemy
uv add -q --dev pytest httpx2 respx
source .venv/bin/activate
mkdir -p app tests && touch app/__init__.py tests/__init__.py
# pip instead of uv:
#   python -m venv .venv && source .venv/bin/activate
#   pip install -q "fastapi[standard]" sqlalchemy pytest httpx2 respx

# file: app/main.py
"""Course Reviews API: the service this page tests."""
import hashlib
import hmac
import json
import os
import time
from collections.abc import Iterator
from contextlib import asynccontextmanager
from typing import Annotated, Protocol

import httpx
from fastapi import Depends, FastAPI, Header, HTTPException, Request
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

PRACTICE_API_URL = os.environ.get("PRACTICE_API_URL", "http://localhost:8800")
DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///reviews.db")
WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "dev-only-secret").encode()
TOKENS = {"token-nadia": "nadia", "token-tanvir": "tanvir", "token-admin": "admin"}  # demo only
ADMINS = {"admin"}


class Base(DeclarativeBase):
    pass


class Review(Base):
    __tablename__ = "reviews"
    id: Mapped[int] = mapped_column(primary_key=True)
    author: Mapped[str] = mapped_column(String(50))
    course_id: Mapped[int]
    rating: Mapped[int]
    comment: Mapped[str] = mapped_column(String(500))


class ReviewIn(BaseModel):
    course_id: int = Field(ge=1)
    rating: int = Field(ge=1, le=5)
    comment: str = Field(min_length=3, max_length=500)


class ReviewOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    author: str
    course_id: int
    rating: int
    comment: str


engine = create_engine(DATABASE_URL)


@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(engine)
    app.state.http = httpx.Client(base_url=PRACTICE_API_URL, timeout=httpx.Timeout(3.0, connect=1.0))
    yield
    app.state.http.close()


app = FastAPI(title="Course Reviews", lifespan=lifespan)


# ---------- dependencies: each one is a seam that tests can replace
def get_session() -> Iterator[Session]:
    with Session(engine) as session:
        yield session


def get_http(request: Request) -> httpx.Client:
    return request.app.state.http


class LLM(Protocol):
    def complete(self, prompt: str) -> str: ...
    def stream(self, prompt: str) -> Iterator[str]: ...


class PracticeLLM:
    """Calls the Practice API's fake LLM; a real provider adapter would sit here."""

    def __init__(self, http: httpx.Client):
        self.http = http

    def _request(self, prompt: str, stream: bool) -> dict:
        return {"headers": {"Authorization": "Bearer token-nadia-001"}, "timeout": 30,
                "json": {"messages": [{"role": "user", "content": prompt}], "stream": stream}}

    def complete(self, prompt: str) -> str:
        resp = self.http.post("/v1/chat", **self._request(prompt, stream=False))
        resp.raise_for_status()
        return resp.json()["message"]["content"]

    def stream(self, prompt: str) -> Iterator[str]:
        with self.http.stream("POST", "/v1/chat", **self._request(prompt, stream=True)) as resp:
            resp.raise_for_status()
            for line in resp.iter_lines():
                if line.startswith("data: ") and "delta" in (event := json.loads(line[6:])):
                    yield event["delta"]


def get_llm(http: Annotated[httpx.Client, Depends(get_http)]) -> LLM:
    return PracticeLLM(http)


class RateLimiter:
    """Sliding-window limiter kept in memory (use Redis in production, as on the previous page)."""

    def __init__(self, limit: int, window: float, clock=time.monotonic):
        self.limit, self.window, self.clock = limit, window, clock
        self.hits: dict[str, list[float]] = {}

    def allow(self, key: str) -> bool:
        now = self.clock()
        recent = [t for t in self.hits.get(key, []) if now - t < self.window]
        allowed = len(recent) < self.limit
        if allowed:
            recent.append(now)
        self.hits[key] = recent
        return allowed


review_limiter = RateLimiter(limit=3, window=60)


def get_limiter() -> RateLimiter:
    return review_limiter


def current_user(authorization: Annotated[str | None, Header()] = None) -> str:
    scheme, _, token = (authorization or "").partition(" ")
    user = TOKENS.get(token) if scheme.lower() == "bearer" else None
    if user is None:
        raise HTTPException(401, "Not authenticated", headers={"WWW-Authenticate": "Bearer"})
    return user


SessionDep = Annotated[Session, Depends(get_session)]
HttpDep = Annotated[httpx.Client, Depends(get_http)]
LLMDep = Annotated[LLM, Depends(get_llm)]
LimiterDep = Annotated[RateLimiter, Depends(get_limiter)]
UserDep = Annotated[str, Depends(current_user)]


# ---------- business logic
def fetch_course(http: httpx.Client, course_id: int, attempts: int = 3) -> dict | None:
    """Look a course up in the course service, retrying 503s."""
    for attempt in range(1, attempts + 1):
        try:
            resp = http.get(f"/v1/courses/{course_id}")
        except httpx.TimeoutException:
            raise HTTPException(504, "Course service timed out")
        if resp.status_code == 404:
            return None
        if resp.status_code == 503 and attempt < attempts:
            continue                       # a real client also backs off here
        if resp.status_code != 200:
            raise HTTPException(502, "Course service error")
        return resp.json()


def verify_signature(secret: bytes, timestamp: str, body: bytes, signature: str,
                     now: float, tolerance: int = 300) -> bool:
    if not timestamp.isdigit() or abs(now - int(timestamp)) > tolerance:
        return False                       # missing, malformed or too old: replay protection
    expected = hmac.new(secret, timestamp.encode() + b"." + body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)


# ---------- routes
@app.get("/health")
def health() -> dict:
    return {"status": "ok"}


@app.post("/reviews", status_code=201)
def create_review(review: ReviewIn, user: UserDep, session: SessionDep,
                  http: HttpDep, limiter: LimiterDep) -> ReviewOut:
    if not limiter.allow(user):
        raise HTTPException(429, "Too many reviews, slow down", headers={"Retry-After": "60"})
    if fetch_course(http, review.course_id) is None:
        raise HTTPException(422, f"Course {review.course_id} does not exist")
    row = Review(author=user, **review.model_dump())
    session.add(row)
    session.commit()
    return ReviewOut.model_validate(row)


@app.get("/reviews/{review_id}")
def get_review(review_id: int, session: SessionDep) -> ReviewOut:
    row = session.get(Review, review_id)
    if row is None:
        raise HTTPException(404, "Review not found")
    return ReviewOut.model_validate(row)


@app.delete("/reviews/{review_id}", status_code=204)
def delete_review(review_id: int, user: UserDep, session: SessionDep) -> None:
    row = session.get(Review, review_id)
    if row is None or (row.author != user and user not in ADMINS):
        raise HTTPException(404, "Review not found")     # don't reveal other people's reviews
    session.delete(row)
    session.commit()


def course_prompt(course_id: int, session: SessionDep) -> tuple[str, int]:
    """A dependency, so a 404 is raised before any streaming starts."""
    comments = session.scalars(select(Review.comment).where(Review.course_id == course_id)).all()
    if not comments:
        raise HTTPException(404, "No reviews for this course yet")
    return "Summarise these course reviews in one sentence:\n" + "\n".join(comments), len(comments)


PromptDep = Annotated[tuple[str, int], Depends(course_prompt)]


@app.post("/courses/{course_id}/summary")
def summarise(course_id: int, prompt: PromptDep, llm: LLMDep) -> dict:
    text, count = prompt
    return {"course_id": course_id, "reviews": count, "summary": llm.complete(text)}


@app.get("/courses/{course_id}/summary/stream", response_class=EventSourceResponse)
def summarise_stream(prompt: PromptDep, llm: LLMDep) -> Iterator[ServerSentEvent]:
    text, count = prompt
    for delta in llm.stream(text):
        yield ServerSentEvent(data={"delta": delta})
    yield ServerSentEvent(event="done", data={"reviews": count})


@app.post("/webhooks/payments")
async def payment_webhook(request: Request, x_timestamp: Annotated[str, Header()],
                          x_signature: Annotated[str, Header()]) -> dict:
    body = await request.body()
    if not verify_signature(WEBHOOK_SECRET, x_timestamp, body, x_signature, time.time()):
        raise HTTPException(401, "Invalid signature")
    return {"received": json.loads(body)["event"]}

cat >> pyproject.toml <<'EOF'

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "--strict-markers"
markers = ["integration: talks to real services (deselect with -m 'not integration')"]
EOF
export PRACTICE_API_URL=http://localhost:8800

# file: tests/conftest.py
"""Shared fixtures: a rolled-back database, fake HTTP, a fake LLM and a fresh rate limiter."""
import httpx
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, event
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool

from app import main
from app.main import Base, RateLimiter, Review, app

COURSES = {1: {"id": 1, "title": "How AI Works"}, 8: {"id": 8, "title": "RAG Systems"}}


class FakeLLM:
    """Stands in for the LLM provider: instant, free, predictable, and it records prompts."""

    def __init__(self):
        self.prompts = []

    def complete(self, prompt):
        self.prompts.append(prompt)
        return "Learners find it clear."

    def stream(self, prompt):
        self.prompts.append(prompt)
        yield from ["Learners ", "find ", "it ", "clear."]


def fake_course_service(request: httpx.Request) -> httpx.Response:
    course_id = int(request.url.path.rsplit("/", 1)[-1])
    if course_id in COURSES:
        return httpx.Response(200, json=COURSES[course_id])
    return httpx.Response(404, json={"detail": "Course not found"})


@pytest.fixture(scope="session")
def engine():
    """One in-memory database for the whole test run."""
    engine = create_engine("sqlite://", poolclass=StaticPool,
                           connect_args={"check_same_thread": False})

    @event.listens_for(engine, "connect")
    def _no_driver_transactions(dbapi_connection, record):
        dbapi_connection.isolation_level = None    # let SQLAlchemy emit BEGIN/SAVEPOINT itself

    @event.listens_for(engine, "begin")
    def _begin(connection):
        connection.exec_driver_sql("BEGIN")

    Base.metadata.create_all(engine)
    yield engine
    engine.dispose()


@pytest.fixture
def db_session(engine):
    """Each test runs inside a transaction that is rolled back afterwards."""
    with engine.connect() as connection:
        transaction = connection.begin()
        session = Session(bind=connection, join_transaction_mode="create_savepoint")
        yield session
        session.close()
        transaction.rollback()


@pytest.fixture
def llm():
    return FakeLLM()


@pytest.fixture
def course_service():
    """What the fake course service does; a test may swap the handler."""
    return {"handler": fake_course_service, "calls": 0}


@pytest.fixture
def client(db_session, llm, course_service):
    def transport_handler(request):
        course_service["calls"] += 1
        return course_service["handler"](request)

    http = httpx.Client(base_url="http://courses.test", transport=httpx.MockTransport(transport_handler))
    limiter = RateLimiter(limit=3, window=60)       # a fresh limiter per test
    app.dependency_overrides[main.get_session] = lambda: db_session
    app.dependency_overrides[main.get_http] = lambda: http
    app.dependency_overrides[main.get_llm] = lambda: llm
    app.dependency_overrides[main.get_limiter] = lambda: limiter
    yield TestClient(app)
    app.dependency_overrides.clear()
    http.close()


@pytest.fixture
def make_review(db_session):
    """A factory fixture: call it to create reviews with sensible defaults."""
    def make(author="nadia", course_id=1, rating=5, comment="Very clear examples"):
        review = Review(author=author, course_id=course_id, rating=rating, comment=comment)
        db_session.add(review)
        db_session.commit()
        return review
    return make


NADIA = {"Authorization": "Bearer token-nadia"}
TANVIR = {"Authorization": "Bearer token-tanvir"}
ADMIN = {"Authorization": "Bearer token-admin"}

# file: tests/test_reviews.py
import pytest

from tests.conftest import NADIA

GOOD = {"course_id": 1, "rating": 5, "comment": "Very clear examples"}


def test_create_review(client):
    resp = client.post("/reviews", json=GOOD, headers=NADIA)
    assert resp.status_code == 201
    assert resp.json() == {"id": 1, "author": "nadia", **GOOD}


def test_database_is_clean_for_every_test(client):
    assert client.get("/reviews/1").status_code == 404


@pytest.mark.parametrize("field, value", [
    ("rating", 0), ("rating", 6), ("comment", "ok"), ("course_id", 0), ("comment", "x" * 501),
])
def test_invalid_reviews_are_rejected(client, field, value):
    resp = client.post("/reviews", json={**GOOD, field: value}, headers=NADIA)
    assert resp.status_code == 422
    assert resp.json()["detail"][0]["loc"] == ["body", field]


def test_unknown_course_is_rejected(client):
    resp = client.post("/reviews", json={**GOOD, "course_id": 99}, headers=NADIA)
    assert resp.status_code == 422
    assert resp.json() == {"detail": "Course 99 does not exist"}

pytest -q

........                                                                                     [100%]
8 passed in 0.04s

pytest -q --setup-show tests/test_reviews.py::test_create_review

SETUP    S engine
        SETUP    F db_session (fixtures used: engine)
        SETUP    F llm
        SETUP    F course_service
        SETUP    F client (fixtures used: course_service, db_session, llm)
        tests/test_reviews.py::test_create_review (fixtures used: client, course_service, db_session, engine, llm) .
        TEARDOWN F client
        TEARDOWN F course_service
        TEARDOWN F llm
        TEARDOWN F db_session
TEARDOWN S engine
1 passed in 0.02s

# file: tests/test_auth.py
import pytest

from tests.conftest import ADMIN, NADIA, TANVIR


@pytest.mark.parametrize("headers", [{}, {"Authorization": "Bearer wrong"}, {"Authorization": "token-nadia"}])
def test_writing_needs_a_valid_token(client, headers):
    resp = client.post("/reviews", json={"course_id": 1, "rating": 5, "comment": "Nice"}, headers=headers)
    assert resp.status_code == 401
    assert resp.headers["WWW-Authenticate"] == "Bearer"


def test_owner_can_delete(client, make_review):
    review = make_review(author="nadia")
    assert client.delete(f"/reviews/{review.id}", headers=NADIA).status_code == 204
    assert client.get(f"/reviews/{review.id}").status_code == 404


def test_other_users_cannot_delete(client, make_review):
    """Security regression test for BOLA: Tanvir must not delete Nadia's review."""
    review = make_review(author="nadia")
    assert client.delete(f"/reviews/{review.id}", headers=TANVIR).status_code == 404
    assert client.get(f"/reviews/{review.id}").status_code == 200


def test_admin_can_delete_any_review(client, make_review):
    review = make_review(author="tanvir")
    assert client.delete(f"/reviews/{review.id}", headers=ADMIN).status_code == 204

# file: tests/test_limits_and_upstream.py
import httpx

from app.main import RateLimiter
from tests.conftest import NADIA, TANVIR

GOOD = {"course_id": 1, "rating": 4, "comment": "Good pace"}


def test_fourth_review_in_a_minute_is_limited(client):
    codes = [client.post("/reviews", json=GOOD, headers=NADIA).status_code for _ in range(4)]
    assert codes == [201, 201, 201, 429]
    assert client.post("/reviews", json=GOOD, headers=TANVIR).status_code == 201


def test_limiter_window_slides():
    now = [0.0]
    limiter = RateLimiter(limit=2, window=60, clock=lambda: now[0])
    assert [limiter.allow("nadia") for _ in range(3)] == [True, True, False]
    now[0] = 60.0                                   # time passes without sleeping
    assert limiter.allow("nadia") is True


def test_course_service_503s_are_retried(client, course_service):
    replies = iter([httpx.Response(503), httpx.Response(503), httpx.Response(200, json={"id": 1})])
    course_service["handler"] = lambda request: next(replies)
    assert client.post("/reviews", json=GOOD, headers=NADIA).status_code == 201
    assert course_service["calls"] == 3


def test_course_service_keeps_failing(client, course_service):
    course_service["handler"] = lambda request: httpx.Response(503)
    resp = client.post("/reviews", json=GOOD, headers=NADIA)
    assert (resp.status_code, course_service["calls"]) == (502, 3)


def test_course_service_timeout_becomes_504(client, course_service):
    def hang(request):
        raise httpx.ReadTimeout("timed out", request=request)

    course_service["handler"] = hang
    resp = client.post("/reviews", json=GOOD, headers=NADIA)
    assert resp.status_code == 504
    assert course_service["calls"] == 1             # timeouts are not retried here

# file: tests/test_llm.py
from tests.conftest import NADIA


def test_summary_uses_the_reviews(client, llm, make_review):
    make_review(comment="Very clear examples")
    make_review(comment="Too fast at the end", rating=3)
    resp = client.post("/courses/1/summary")
    assert resp.json() == {"course_id": 1, "reviews": 2, "summary": "Learners find it clear."}
    assert "Too fast at the end" in llm.prompts[0]


def test_no_reviews_means_404_and_no_llm_call(client, llm):
    assert client.post("/courses/8/summary").status_code == 404
    assert llm.prompts == []                        # no tokens spent


def test_summary_stream(client, make_review):
    make_review()
    with client.stream("GET", "/courses/1/summary/stream") as resp:
        assert resp.headers["content-type"].startswith("text/event-stream")
        lines = [line for line in resp.iter_lines() if line]
    assert lines == [
        'data: {"delta": "Learners "}', 'data: {"delta": "find "}',
        'data: {"delta": "it "}', 'data: {"delta": "clear."}',
        "event: done", 'data: {"reviews": 1}',
    ]

# file: tests/test_webhooks.py
import hashlib
import hmac
import json
import time

import pytest

from app.main import WEBHOOK_SECRET, verify_signature


def signed(body: dict, secret: bytes = WEBHOOK_SECRET, timestamp: int | None = None):
    raw = json.dumps(body).encode()
    ts = str(timestamp if timestamp is not None else int(time.time()))
    sig = hmac.new(secret, ts.encode() + b"." + raw, hashlib.sha256).hexdigest()
    return raw, {"X-Timestamp": ts, "X-Signature": sig, "Content-Type": "application/json"}


def test_valid_webhook_is_accepted(client):
    raw, headers = signed({"event": "payment.succeeded"})
    resp = client.post("/webhooks/payments", content=raw, headers=headers)
    assert resp.json() == {"received": "payment.succeeded"}


@pytest.mark.parametrize("case", ["wrong secret", "tampered body", "replayed"])
def test_bad_webhooks_are_rejected(client, case):
    if case == "wrong secret":
        raw, headers = signed({"event": "payment.succeeded"}, secret=b"guess")
    elif case == "tampered body":
        _, headers = signed({"event": "payment.succeeded", "amount": 500})
        raw = json.dumps({"event": "payment.succeeded", "amount": 5}).encode()
    else:
        raw, headers = signed({"event": "payment.succeeded"}, timestamp=int(time.time()) - 3600)
    assert client.post("/webhooks/payments", content=raw, headers=headers).status_code == 401


def test_signature_check_is_a_pure_function():
    """A unit test: no app, no HTTP, a fixed clock."""
    raw = b'{"event": "x"}'
    sig = hmac.new(b"k", b"1000." + raw, hashlib.sha256).hexdigest()
    assert verify_signature(b"k", "1000", raw, sig, now=1100)
    assert not verify_signature(b"k", "1000", raw, sig, now=1400)     # 400 s later: too old
    assert not verify_signature(b"k", "abc", raw, sig, now=1000)

pytest -q

...........................                                                                  [100%]
27 passed in 0.11s

# file: tests/test_contract.py
"""Contract test: the OpenAPI schema is the promise you made to clients."""
import json
import os
from pathlib import Path

from app.main import app

SNAPSHOT = Path(__file__).parent / "openapi.snapshot.json"


def test_openapi_matches_snapshot():
    current = app.openapi()
    if os.environ.get("UPDATE_SNAPSHOT") == "1" or not SNAPSHOT.exists():
        SNAPSHOT.write_text(json.dumps(current, indent=2, sort_keys=True) + "\n")
    saved = json.loads(SNAPSHOT.read_text())
    assert sorted(current["paths"]) == sorted(saved["paths"]), "endpoints were added or removed"
    for name, schema in saved["components"]["schemas"].items():
        now = current["components"]["schemas"].get(name, {})
        assert now.get("properties") == schema.get("properties"), f"schema {name} changed"
        assert now.get("required") == schema.get("required"), f"required fields of {name} changed"

pytest -q tests/test_contract.py
ls tests/openapi.snapshot.json

.                                                                                            [100%]
1 passed in 0.01s
tests/openapi.snapshot.json

sed -i 's/    comment: str = Field(min_length=3, max_length=500)/    text: str = Field(min_length=3, max_length=500)/' app/main.py
pytest -q --tb=short tests/test_contract.py

F                                                                                            [100%]
============================================= FAILURES =============================================
__________________________________ test_openapi_matches_snapshot ___________________________________
tests/test_contract.py:20: in test_openapi_matches_snapshot
    assert now.get("properties") == schema.get("properties"), f"schema {name} changed"
E   AssertionError: schema ReviewIn changed
E   assert {'course_id':...tle': 'Text'}} == {'comment': {...': 'integer'}}
E     
E     Omitting 2 identical items, use -vv to show
E     Left contains 1 more item:
E     {'text': {'maxLength': 500, 'minLength': 3, 'title': 'Text', 'type': 'string'}}
E     Right contains 1 more item:
E     {'comment': {'maxLength': 500,
E                  'minLength': 3,...
E     
E     ...Full output truncated (3 lines hidden), use '-vv' to show
===================================== short test summary info ======================================
FAILED tests/test_contract.py::test_openapi_matches_snapshot - AssertionError: schema ReviewIn ch...
1 failed in 0.04s

sed -i 's/    text: str = Field(min_length=3, max_length=500)/    comment: str = Field(min_length=3, max_length=500)/' app/main.py
pytest -q tests/test_contract.py

.                                                                                            [100%]
1 passed in 0.01s

# file: tests/test_integration.py
"""Tests that talk to real services. Slower; run them on purpose with -m integration."""
import httpx
import pytest
import respx
from fastapi.testclient import TestClient

from app import main

pytestmark = pytest.mark.integration


def test_real_course_service_lookup():
    with httpx.Client(base_url=main.PRACTICE_API_URL, timeout=5) as http:
        assert main.fetch_course(http, 2)["title"] == "Python for AI"
        assert main.fetch_course(http, 999) is None


def test_practice_llm_adapter_streams():
    with httpx.Client(base_url=main.PRACTICE_API_URL, timeout=5) as http:
        text = "".join(main.PracticeLLM(http).stream("What is MLOps?"))
    assert text.startswith("This is a practice answer about: What is MLOps.")


@respx.mock
def test_app_with_respx_and_lifespan(tmp_path, monkeypatch):
    """The whole app, real lifespan and real httpx client; respx answers the HTTP calls."""
    monkeypatch.setattr(main, "engine", main.create_engine(f"sqlite:///{tmp_path}/test.db"))
    route = respx.get(f"{main.PRACTICE_API_URL}/v1/courses/1").mock(
        return_value=httpx.Response(200, json={"id": 1}))
    with TestClient(main.app) as client:            # "with" runs the lifespan
        resp = client.post("/reviews", json={"course_id": 1, "rating": 5, "comment": "Great"},
                           headers={"Authorization": "Bearer token-nadia"})
    assert resp.status_code == 201
    assert route.call_count == 1

pytest -q -m "not integration"
pytest -q -m integration

............................                                                                 [100%]
28 passed, 3 deselected in 0.17s
...                                                                                          [100%]
3 passed, 28 deselected in 0.87s

# file: tests/test_async.py
import httpx
import pytest

from app.main import app


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


@pytest.mark.anyio
async def test_health_with_an_async_client():
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac:
        resp = await ac.get("/health")
    assert resp.json() == {"status": "ok"}

pytest -q tests/test_async.py

.                                                                                            [100%]
1 passed in 0.01s

uv add --dev pytest-cov                 # or: pip install pytest-cov
pytest --cov=app --cov-branch --cov-report=term-missing

# file: locustfile.py
from locust import HttpUser, between, task


class Learner(HttpUser):
    wait_time = between(1, 3)                       # think time between actions

    @task(5)
    def read_review(self):
        self.client.get("/reviews/1", name="/reviews/[id]")

    @task(1)
    def summary(self):
        self.client.post("/courses/1/summary")

uv run --with locust locust -f locustfile.py --host http://localhost:8000 \
    --users 50 --spawn-rate 5 --run-time 2m --headless

# file: tests/test_exercises.py
import httpx
import pytest

from tests.conftest import NADIA


# 1. Easy
@pytest.mark.parametrize("review_id, status", [("abc", 422), ("1.5", 422), ("12345", 404)])
def test_get_review_ids(client, review_id, status):
    assert client.get(f"/reviews/{review_id}").status_code == status


# 2. Medium
def test_stream_without_reviews(client, llm):
    assert client.get("/courses/8/summary/stream").status_code == 404
    assert llm.prompts == []


# 3. Hard
def test_500_is_not_retried(client, course_service):
    course_service["handler"] = lambda request: httpx.Response(500)
    resp = client.post("/reviews", json={"course_id": 1, "rating": 5, "comment": "Nice"}, headers=NADIA)
    assert (resp.status_code, course_service["calls"]) == (502, 1)

pytest -q tests/test_exercises.py

.....                                                                                        [100%]
5 passed in 0.03s