FastAPI Endpoint Testleri pytest ile Nasıl Yazılır

Üretime aldığımız ilk FastAPI servisinde test yoktu. “Küçük bir proje, ne olacak” dedik. Üç ay sonra o servis 12 endpoint’e çıkmıştı ve bir şeyi değiştirdiğimizde başka bir şey kırılıyordu. Sabah 3’te production’da debug yapmak nasıl bir his, yaşayanlar bilir. O günden sonra her FastAPI projesinde test yazmak bizim için opsiyonel değil, zorunlu hale geldi.

Bu yazıda FastAPI endpoint’lerini pytest ile nasıl test edeceğinizi, fixture’ları nasıl organize edeceğinizi, database bağlantılarını test ortamında nasıl izole edeceğinizi ve gerçek dünyada karşılaştığım edge case’leri ele alacağım.

Temel Kurulum ve Bağımlılıklar

Önce gerekli paketleri kuralım:

pip install fastapi pytest pytest-asyncio httpx anyio
pip install sqlalchemy alembic pytest-mock
# Opsiyonel ama hayatı kolaylaştırır
pip install factory-boy faker

Proje yapımız şöyle olsun:

myproject/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── schemas.py
│   ├── database.py
│   └── routers/
│       ├── users.py
│       └── products.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_users.py
│   └── test_products.py
└── pytest.ini

pytest.ini dosyamız:

[pytest]
asyncio_mode = auto
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
filterwarnings =
    ignore::DeprecationWarning

TestClient ile İlk Adımlar

FastAPI’nin en güzel yanlarından biri httpx tabanlı bir test client sunması. Starlette’den gelen bu yapı sayesinde gerçek HTTP isteği atmadan endpoint’leri test edebiliyorsunuz.

Önce basit bir uygulamamız olsun:

# app/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, List

app = FastAPI()

# Geçici in-memory storage (test için ideal)
fake_db = {}

class User(BaseModel):
    id: Optional[int] = None
    username: str
    email: str
    is_active: bool = True

@app.get("/users", response_model=List[User])
def get_users():
    return list(fake_db.values())

@app.get("/users/{user_id}", response_model=User)
def get_user(user_id: int):
    if user_id not in fake_db:
        raise HTTPException(status_code=404, detail="User not found")
    return fake_db[user_id]

@app.post("/users", response_model=User, status_code=201)
def create_user(user: User):
    user_id = len(fake_db) + 1
    user.id = user_id
    fake_db[user_id] = user
    return user

@app.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int):
    if user_id not in fake_db:
        raise HTTPException(status_code=404, detail="User not found")
    del fake_db[user_id]

Şimdi bu endpoint’ler için temel testleri yazalım:

# tests/test_users.py
from fastapi.testclient import TestClient
from app.main import app, fake_db

client = TestClient(app)

def setup_function():
    """Her test fonksiyonu öncesinde çalışır"""
    fake_db.clear()

def test_get_users_empty():
    response = client.get("/users")
    assert response.status_code == 200
    assert response.json() == []

def test_create_user_success():
    payload = {
        "username": "ahmet",
        "email": "[email protected]"
    }
    response = client.post("/users", json=payload)
    assert response.status_code == 201
    data = response.json()
    assert data["username"] == "ahmet"
    assert data["email"] == "[email protected]"
    assert data["id"] is not None
    assert data["is_active"] is True

def test_get_user_not_found():
    response = client.get("/users/999")
    assert response.status_code == 404
    assert response.json()["detail"] == "User not found"

def test_delete_user_success():
    # Önce kullanıcı oluştur
    create_response = client.post("/users", json={
        "username": "mehmet",
        "email": "[email protected]"
    })
    user_id = create_response.json()["id"]
    
    # Sonra sil
    delete_response = client.delete(f"/users/{user_id}")
    assert delete_response.status_code == 204
    
    # Gerçekten silinmiş mi kontrol et
    get_response = client.get(f"/users/{user_id}")
    assert get_response.status_code == 404

conftest.py ile Fixture Yönetimi

Testlerin büyüyüp karmaşıklaştığı noktada conftest.py hayat kurtarır. Her test dosyasında tekrar tekrar aynı setup kodunu yazmak hem bakımı zorlaştırır hem de bir değişiklikte onlarca dosyayı güncellemenizi gerektirir.

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

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

# Test için in-memory SQLite kullanıyoruz
SQLALCHEMY_TEST_DATABASE_URL = "sqlite://"

@pytest.fixture(scope="session")
def test_engine():
    engine = create_engine(
        SQLALCHEMY_TEST_DATABASE_URL,
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,  # In-memory için şart
    )
    Base.metadata.create_all(bind=engine)
    yield engine
    Base.metadata.drop_all(bind=engine)

@pytest.fixture(scope="function")
def db_session(test_engine):
    """Her test için temiz bir session"""
    TestingSessionLocal = sessionmaker(
        autocommit=False, 
        autoflush=False, 
        bind=test_engine
    )
    session = TestingSessionLocal()
    try:
        yield session
    finally:
        session.rollback()
        session.close()

@pytest.fixture(scope="function")
def client(db_session):
    """Database dependency'yi override eden test client"""
    def override_get_db():
        try:
            yield db_session
        finally:
            pass
    
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as test_client:
        yield test_client
    app.dependency_overrides.clear()

@pytest.fixture
def sample_user_data():
    return {
        "username": "testuser",
        "email": "[email protected]",
        "full_name": "Test Kullanici"
    }

@pytest.fixture
def created_user(client, sample_user_data):
    """Veritabanında hazır bir kullanıcı"""
    response = client.post("/users", json=sample_user_data)
    assert response.status_code == 201
    return response.json()

Burada dikkat edilmesi gereken nokta scope parametresi. session scope’lu fixture sadece test oturumunda bir kez çalışır, function scope’lu ise her test için yeniden oluşturulur. Database engine’i bir kez oluşturmak mantıklı ama her testin temiz bir transaction ile başlaması kritik.

Authentication Testleri

Gerçek dünya uygulamalarının büyük çoğunluğunda kimlik doğrulama var. JWT token’larını test ederken sık yapılan hata, gerçek token üretip onunla test etmeye çalışmak. Bunun yerine dependency injection’ı override etmek çok daha temiz:

# tests/conftest.py'a eklenecek
from app.auth import get_current_user
from app.models import User

@pytest.fixture
def mock_current_user():
    return User(
        id=1,
        username="testadmin",
        email="[email protected]",
        is_active=True,
        role="admin"
    )

@pytest.fixture
def authenticated_client(client, mock_current_user):
    """Admin yetkili test client"""
    app.dependency_overrides[get_current_user] = lambda: mock_current_user
    yield client
    # client fixture'ı zaten temizliyor ama explicit olalım
    if get_current_user in app.dependency_overrides:
        del app.dependency_overrides[get_current_user]

@pytest.fixture  
def regular_user_client(client):
    """Normal kullanıcı yetkili test client"""
    regular_user = User(
        id=2,
        username="normaluser", 
        email="[email protected]",
        is_active=True,
        role="user"
    )
    app.dependency_overrides[get_current_user] = lambda: regular_user
    yield client

Kullanımı:

# tests/test_admin_endpoints.py
def test_admin_only_endpoint_with_admin(authenticated_client):
    response = authenticated_client.get("/admin/users")
    assert response.status_code == 200

def test_admin_only_endpoint_with_regular_user(regular_user_client):
    response = regular_user_client.get("/admin/users")
    assert response.status_code == 403

def test_protected_endpoint_without_auth(client):
    """Auth override yok, 401 bekliyoruz"""
    response = client.get("/profile")
    assert response.status_code == 401

Harici Servis ve Mock Kullanımı

Ödeme sistemi, e-posta servisi, SMS gateway… Bunları gerçekten test ortamında çağıramazsınız. pytest-mock bu noktada devreye girer:

# tests/test_payment_endpoint.py
import pytest
from unittest.mock import AsyncMock, patch

def test_payment_endpoint_success(authenticated_client, created_user, mocker):
    """Ödeme servisi başarılı döndüğünde"""
    mock_payment = mocker.patch(
        "app.services.payment.PaymentService.charge",
        return_value={
            "transaction_id": "TXN123456",
            "status": "success",
            "amount": 99.99
        }
    )
    
    response = authenticated_client.post("/payments", json={
        "user_id": created_user["id"],
        "amount": 99.99,
        "card_token": "tok_test_123"
    })
    
    assert response.status_code == 200
    data = response.json()
    assert data["transaction_id"] == "TXN123456"
    assert data["status"] == "success"
    
    # Servis gerçekten çağrıldı mı?
    mock_payment.assert_called_once_with(
        amount=99.99,
        card_token="tok_test_123"
    )

def test_payment_endpoint_service_failure(authenticated_client, created_user, mocker):
    """Ödeme servisi hata döndüğünde graceful handling"""
    mocker.patch(
        "app.services.payment.PaymentService.charge",
        side_effect=Exception("Payment gateway timeout")
    )
    
    response = authenticated_client.post("/payments", json={
        "user_id": created_user["id"],
        "amount": 99.99,
        "card_token": "tok_test_123"
    })
    
    assert response.status_code == 502
    assert "payment" in response.json()["detail"].lower()

def test_email_notification_sent_after_registration(client, mocker):
    """Kayıt sonrası e-posta gönderildi mi?"""
    mock_send_email = mocker.patch(
        "app.services.email.EmailService.send_welcome_email",
        return_value=True
    )
    
    response = client.post("/users/register", json={
        "username": "yenikullanici",
        "email": "[email protected]",
        "password": "Gizli123!"
    })
    
    assert response.status_code == 201
    mock_send_email.assert_called_once_with("[email protected]")

Async Endpoint Testleri

FastAPI’nin asıl gücü async endpoint’lerde. pytest-asyncio ile bunları da rahatlıkla test edebilirsiniz:

# tests/test_async_endpoints.py
import pytest
import asyncio
from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_async_bulk_operation():
    """Birden fazla isteği paralel gönderme testi"""
    async with AsyncClient(app=app, base_url="http://test") as ac:
        # Paralel istek gönder
        tasks = [
            ac.get(f"/users/{i}") 
            for i in range(1, 6)
        ]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Hepsi 404 dönmeli (boş db)
        for response in responses:
            if not isinstance(response, Exception):
                assert response.status_code == 404

@pytest.mark.asyncio
async def test_async_endpoint_response_time():
    """Endpoint belirli sürede yanıt vermeli"""
    import time
    
    async with AsyncClient(app=app, base_url="http://test") as ac:
        start = time.monotonic()
        response = await ac.get("/health")
        elapsed = time.monotonic() - start
        
        assert response.status_code == 200
        assert elapsed < 0.5, f"Yanıt çok geç geldi: {elapsed:.3f}s"

@pytest.mark.asyncio  
async def test_websocket_endpoint():
    """WebSocket endpoint testi"""
    from starlette.testclient import TestClient
    
    with TestClient(app) as client:
        with client.websocket_connect("/ws/notifications") as ws:
            ws.send_json({"type": "subscribe", "channel": "alerts"})
            data = ws.receive_json()
            assert data["type"] == "subscribed"

Parameterize ile Kapsamlı Test Senaryoları

Aynı endpoint’e farklı input’larla test yazmak için pytest.mark.parametrize kullanın. Bu özellikle validasyon testlerinde çok işe yarıyor:

# tests/test_validation.py
import pytest

invalid_user_payloads = [
    # (payload, expected_status, error_field)
    ({"username": "", "email": "[email protected]"}, 422, "username"),
    ({"username": "ab", "email": "[email protected]"}, 422, "username"),  # çok kısa
    ({"username": "validuser", "email": "gecersiz-email"}, 422, "email"),
    ({"username": "validuser", "email": ""}, 422, "email"),
    ({}, 422, None),  # body yok
    ({"username": "a" * 51, "email": "[email protected]"}, 422, "username"),  # çok uzun
]

@pytest.mark.parametrize("payload,expected_status,error_field", invalid_user_payloads)
def test_user_creation_validation(client, payload, expected_status, error_field):
    response = client.post("/users", json=payload)
    assert response.status_code == expected_status
    
    if error_field:
        errors = response.json()["detail"]
        error_fields = [err["loc"][-1] for err in errors]
        assert error_field in error_fields, 
            f"'{error_field}' alanında hata bekliyorduk ama gelmedi. Gelen: {errors}"

# HTTP method testleri
@pytest.mark.parametrize("method,path,expected_status", [
    ("get", "/users/abc", 422),      # integer bekliyor string geldi
    ("get", "/users/-1", 404),       # negatif id
    ("get", "/users/0", 404),        # sıfır id  
    ("delete", "/users/999", 404),   # olmayan kullanıcı
    ("put", "/users/999", 404),      # olmayan kullanıcı update
])
def test_edge_cases(client, method, path, expected_status):
    response = getattr(client, method)(path)
    assert response.status_code == expected_status

Integration Test: Gerçekçi Senaryo

Birim testler güzel ama bazen baştan sona bir iş akışını test etmek gerekir. Sipariş alma sürecini ele alalım:

# tests/test_order_flow.py
import pytest

class TestOrderFlow:
    """Sipariş akışı end-to-end testi"""
    
    def test_complete_order_flow(self, authenticated_client):
        """
        Gerçek kullanım senaryosu:
        1. Kullanıcı ürünleri listeler
        2. Sepete ekler  
        3. Siparişi onaylar
        4. Sipariş durumunu kontrol eder
        """
        # Adım 1: Ürünleri listele
        products_response = authenticated_client.get("/products?category=elektronik")
        assert products_response.status_code == 200
        products = products_response.json()
        assert len(products) > 0
        product_id = products[0]["id"]
        
        # Adım 2: Sepete ekle
        cart_response = authenticated_client.post("/cart/items", json={
            "product_id": product_id,
            "quantity": 2
        })
        assert cart_response.status_code == 201
        cart_item = cart_response.json()
        assert cart_item["quantity"] == 2
        
        # Adım 3: Siparişi oluştur
        order_response = authenticated_client.post("/orders", json={
            "shipping_address": "Atatürk Cad. No:1 İstanbul",
            "payment_method": "credit_card"
        })
        assert order_response.status_code == 201
        order = order_response.json()
        order_id = order["id"]
        assert order["status"] == "pending"
        
        # Adım 4: Sipariş durumu
        status_response = authenticated_client.get(f"/orders/{order_id}")
        assert status_response.status_code == 200
        assert status_response.json()["id"] == order_id
    
    def test_out_of_stock_handling(self, authenticated_client, mocker):
        """Stok tükendikten sonra sipariş vermeye çalışma"""
        mocker.patch(
            "app.services.inventory.check_stock",
            return_value={"available": 0, "reserved": 5}
        )
        
        response = authenticated_client.post("/cart/items", json={
            "product_id": 1,
            "quantity": 1
        })
        
        assert response.status_code == 409
        assert "stok" in response.json()["detail"].lower() or 
               "stock" in response.json()["detail"].lower()

Test Coverage ve Raporlama

Test yazdık, güzel. Peki neyi kapsıyor bu testler? pytest-cov ile görelim:

# Coverage raporu al
pip install pytest-cov

# Terminal çıktısı
pytest --cov=app --cov-report=term-missing tests/

# HTML rapor (çok daha okunaklı)
pytest --cov=app --cov-report=html:coverage_report tests/
open coverage_report/index.html

# Minimum coverage zorunlu kılma (CI/CD için)
pytest --cov=app --cov-fail-under=80 tests/

# Belirli dosyaları dışla
pytest --cov=app --cov-omit="app/migrations/*,app/config.py" tests/

.coveragerc dosyası ile daha granüler kontrol:

# .coveragerc
[run]
source = app
omit = 
    app/migrations/*
    app/tests/*
    app/config.py
    */__init__.py

[report]
exclude_lines =
    pragma: no cover
    def __repr__
    raise NotImplementedError
    if TYPE_CHECKING:
    pass

[html]
directory = coverage_report

Bazı sysadmin’lerin yaptığı hata: coverage’ı %100’e çıkarmak için anlamsız testler yazmak. %80-85 gerçek test senaryolarıyla dolu coverage, %100 sahte testlerden iyidir. Özellikle kritik iş mantığını kapsayan kod bloklarına odaklanın.

CI/CD Pipeline Entegrasyonu

Testleri lokal çalıştırmak yetmez, her push’ta otomatik çalışmalı. GitHub Actions için minimal bir konfigurasyon:

# .github/workflows/test.yml
name: FastAPI Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    
    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_PASSWORD: testpass
          POSTGRES_DB: testdb
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Python kur
      uses: actions/setup-python@v4
      with:
        python-version: '3.11'
    
    - name: Bağımlılıkları kur
      run: |
        pip install -r requirements.txt
        pip install -r requirements-test.txt
    
    - name: Testleri çalıştır
      env:
        DATABASE_URL: postgresql://postgres:testpass@localhost/testdb
        TESTING: "true"
      run: |
        pytest --cov=app --cov-report=xml --cov-fail-under=75 -v
    
    - name: Coverage raporunu yükle
      uses: codecov/codecov-action@v3
      with:
        file: ./coverage.xml

Sonuç

FastAPI endpoint testleri yazarken en çok değer katan şey fixture organizasyonuna zaman harcamak. conftest.py‘ı düzgün kurarsanız her yeni test dosyası için neredeyse sıfır setup kodu yazarsınız.

Pratikte işe yarayan birkaç kuralı paylaşayım:

• Her endpoint için mutlaka: başarılı senaryo, 404/422 durumları, ve authentication testi yazın.
• Harici servisler için: gerçek API çağrısı yapmayın, mock kullanın. Test ortamınız production’ın ödeme servisine istek atmaya başlarsa ciddi sorunlar çıkabilir.
• Database izolasyonu: her test fonksiyonu kendi transaction’ında çalışmalı ve rollback yapmalı. Test sırası fark etmemeli.
• Test isimlendirmesi: test_user_can_delete_own_profile_but_not_others gibi açıklayıcı isimler kullanın. Altı ay sonra kodu okuyan siz veya başkası teşekkür edecek.
• Flaky testlerle: paralel çalışan testlerde shared state varsa aniden flaky olmaya başlarlar. scope="function" fixture’lar bu sorunu büyük ölçüde çözer.

FastAPI ekosistemi bu anlamda gerçekten olgun. Test altyapısı hazır, dependency injection mock’lamayı kolaylaştırıyor, async desteği mevcut. Geriye sadece oturup yazmak kalıyor. Başlamak için mükemmel bir an beklemeyin, var olan en kritik endpoint’i alın ve onun testini yazın. Geri kalanı kendiliğinden gelecektir.