آزمون خودکار API با pytest

۱. چه چیزی را آزمون کنیم

هر آزمون سه بخش دارد: آماده‌سازی، اجرا و بررسی. و قاعده این فصل: حالت‌های ناموفق به همان اندازه مهم‌اند. ورودی بد باید دقیقاً همان‌طور که قول داده‌ایم شکست بخورد.

۲. pytest، ساده

import pytest

from pricing import PricingError, apply_discount, installments


def test_discount_reduces_the_price() -> None:
    assert apply_discount(1_000_000, 10) == 900_000


def test_no_discount_keeps_the_price() -> None:
    assert apply_discount(1_000_000, 0) == 1_000_000


@pytest.mark.parametrize(
    ('price', 'percent', 'expected'),
    [
        (100, 50, 50),
        (99, 50, 49),  # rounded down
        (4_800_000, 100, 0),
    ],
)
def test_discount_table(price: int, percent: int, expected: int) -> None:
    assert apply_discount(price, percent) == expected


@pytest.mark.parametrize('percent', [-1, 101, 1000])
def test_invalid_discount_is_rejected(percent: int) -> None:
    with pytest.raises(PricingError, match='between 0 and 100'):
        apply_discount(1000, percent)


def test_installments_add_up_to_the_total() -> None:
    parts = installments(1_000_000, 3)

    assert sum(parts) == 1_000_000
    assert parts == [333_334, 333_333, 333_333]


@pytest.fixture
def big_price() -> int:
    """A fixture: reusable test data, created fresh for every test that asks for it."""
    return 12_345_678


def test_installments_with_fixture(big_price: int) -> None:
    assert len(installments(big_price, 12)) == 12


def test_zero_installments_fail() -> None:
    with pytest.raises(PricingError):
        installments(100, 0)

assert، parametrize، raises و fixture: 01_pytest_basics/test_pricing.py

۳. TestClient

TestClient(app) درخواست‌های واقعی HTTP را درون همین فرایند به برنامه می‌فرستد: بدون سرور و بدون شبکه. همان رابط کلاینت HTTPX را دارد:

import pytest
from fastapi.testclient import TestClient

from app import COURSES, app


@pytest.fixture
def client() -> TestClient:
    COURSES.clear()  # every test starts with an empty "database"
    return TestClient(app)


def test_create_course(client: TestClient) -> None:
    response = client.post('/courses', json={'title': 'FastAPI', 'price': 4_800_000})

    assert response.status_code == 201
    assert response.json() == {'id': 1, 'title': 'FastAPI', 'price': 4_800_000}


def test_read_back_what_was_created(client: TestClient) -> None:
    created = client.post('/courses', json={'title': 'FastAPI', 'price': 1}).json()

    response = client.get(f'/courses/{created["id"]}')

    assert response.status_code == 200
    assert response.json()['title'] == 'FastAPI'


def test_missing_course_is_404(client: TestClient) -> None:
    response = client.get('/courses/999')

    assert response.status_code == 404
    assert response.json() == {'detail': 'Course not found'}


@pytest.mark.parametrize(
    ('body', 'bad_field'),
    [
        ({'title': 'ab', 'price': 1}, 'title'),
        ({'title': 'FastAPI', 'price': -5}, 'price'),
        ({'title': 'FastAPI'}, 'price'),
    ],
)
def test_validation_errors(client: TestClient, body: dict, bad_field: str) -> None:
    response = client.post('/courses', json=body)

    assert response.status_code == 422
    assert response.json()['detail'][0]['loc'] == ['body', bad_field]


def test_wrong_path_type_is_422(client: TestClient) -> None:
    assert client.get('/courses/abc').status_code == 422

۲۰۱، ۴۰۴ و ۴۲۲: 02_testclient/test_app.py

۴. جایگزینی وابستگی‌ها

from collections.abc import Iterator

import pytest
from fastapi.testclient import TestClient

from app import app, get_current_user, get_db


class FakeDatabase:
    def list_courses(self) -> list[str]:
        return ['Python', 'FastAPI']


@pytest.fixture
def client() -> Iterator[TestClient]:
    app.dependency_overrides[get_db] = FakeDatabase
    yield TestClient(app)
    app.dependency_overrides.clear()


def login_as(role: str) -> None:
    app.dependency_overrides[get_current_user] = lambda: {'email': f'{role}@x.com', 'role': role}


def test_student_sees_their_courses(client: TestClient) -> None:
    login_as('student')

    response = client.get('/my-courses')

    assert response.status_code == 200
    assert response.json() == {'user': 'student@x.com', 'courses': ['Python', 'FastAPI']}


def test_instructor_is_forbidden(client: TestClient) -> None:
    login_as('instructor')

    assert client.get('/my-courses').status_code == 403


def test_without_override_the_real_auth_runs(client: TestClient) -> None:
    assert client.get('/my-courses').status_code == 401

پایگاه داده جعلی و کاربر جعلی: 03_dependency_overrides/test_app.py

app.dependency_overrides[get_db] = FakeDatabase یعنی هر endpointی که به get_db وابسته است حالا نسخه جعلی را می‌گیرد. این همان سود تزریق وابستگی است: کد endpoint تغییر نمی‌کند، فقط چیزی که تزریق می‌کنیم عوض می‌شود. جایگزینی‌ها را همیشه بعد از آزمون پاک کنید؛ fixture با yield این را حتی وقتی آزمون شکست بخورد تضمین می‌کند.

۵. آزمون ناهمگام با HTTPX

وقتی خود آزمون باید ناهمگام باشد، مثلاً برای await کردن کمک‌کننده‌های پایگاه داده یا فرستادن درخواست‌های هم‌زمان، httpx.AsyncClient را به برنامه وصل می‌کنیم:

pytestmark = pytest.mark.anyio


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


@pytest.fixture
async def client() -> AsyncIterator[httpx.AsyncClient]:
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url='http://test') as async_client:
        yield async_client


async def test_one_request(client: httpx.AsyncClient) -> None:
    response = await client.get('/courses/5/seats')

    assert response.status_code == 200
    assert response.json() == {'course_id': 5, 'seats': 15}


async def test_many_requests_at_once(client: httpx.AsyncClient) -> None:
    responses = await asyncio.gather(*(client.get(f'/courses/{i}/seats') for i in range(20)))

    assert all(r.status_code == 200 for r in responses)
    assert [r.json()['seats'] for r in responses] == [10 + i for i in range(20)]


async def test_a_coroutine_directly() -> None:
    assert await fetch_seats(1) == 11

AsyncClient و ASGITransport: 04_async_tests/test_app.py

pytest.mark.anyio آزمون را در یک حلقه رویداد اجرا می‌کند؛ افزونه AnyIO همراه FastAPI نصب می‌شود. ASGITransport چرخه lifespan را اجرا نمی‌کند، که در پروژه این را در نظر گرفته‌ایم.

۶. پروژه: پنجاه و یک آزمون

@pytest.fixture
def settings(tmp_path: Path) -> Settings:
    return Settings(
        database_url=f'sqlite+aiosqlite:///{tmp_path / "test.db"}',
        jwt_secret_key='test-secret-key-that-is-long-enough-0123456789',
        upload_dir=tmp_path / 'uploads',
        max_upload_mb=1,
    )


@pytest.fixture
async def session_factory(settings: Settings) -> AsyncIterator[async_sessionmaker[AsyncSession]]:
    engine = create_async_engine(settings.database_url)

    @event.listens_for(engine.sync_engine, 'connect')
    def foreign_keys_on(connection, _record) -> None:  # type: ignore[no-untyped-def]
        connection.cursor().execute('PRAGMA foreign_keys=ON')

    async with engine.begin() as connection:
        await connection.run_sync(Base.metadata.create_all)
    yield async_sessionmaker(engine, expire_on_commit=False)
    await engine.dispose()


@pytest.fixture
async def client(
    settings: Settings, session_factory: async_sessionmaker[AsyncSession]
) -> AsyncIterator[httpx.AsyncClient]:
    async def test_db() -> AsyncIterator[AsyncSession]:
        async with session_factory() as session:
            yield session

    app.dependency_overrides[get_db] = test_db
    app.dependency_overrides[get_settings] = lambda: settings
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url='http://test') as async_client:
        yield async_client
    app.dependency_overrides.clear()

پایگاه داده آزمون و جایگزینی وابستگی‌ها: 05_project/tests/conftest.py

هر آزمون این‌ها را می‌گیرد:

• یک فایل SQLite تازه در یک پوشه موقت؛ آزمون‌ها داده یکدیگر را نمی‌بینند و هرگز به training.db دست نمی‌زنند.
• get_db و get_settings جایگزین شده با نسخه آزمونی.
• یک کلاینت ناهمگام HTTPX که درون فرایند با برنامه صحبت می‌کند.
• کارخانه‌هایی برای ساختن کاربر و دوره مستقیم در پایگاه داده، و auth(user) که بدون صدا زدن /auth/token هدر مجوز می‌سازد.

async def test_enroll_many_is_all_or_nothing(
    client: AsyncClient, make_user: UserFactory, make_course: CourseFactory, auth: Auth
) -> None:
    teacher = await make_user('instructor')
    open_course = await make_course(teacher, title='Open')
    full_course = await make_course(teacher, title='Full', capacity=1)
    await client.post(
        f'/courses/{full_course.id}/enrollments', headers=auth(await make_user('student'))
    )
    student = await make_user('student')

    response = await client.post(
        '/users/me/enrollments',
        json={'course_ids': [open_course.id, full_course.id]},
        headers=auth(student),
    )
    mine = await client.get('/users/me/enrollments', headers=auth(student))

    assert response.status_code == 409
    assert mine.json() == []  # the first enrollment was rolled back

آزمون «همه یا هیچ»: 05_project/tests/test_enrollments.py

cd 05_project
pytest
...................................................                      [100%]
51 passed in 5.27s

۷. خلاصه فصل