Python ile Integration Test Yazımı ve Veritabanı Testleri

Yazar:komut.io ekibi
0 Yorum

Prodüksiyonda bir şey patladığında genellikle ilk suçlanan birim testleridir. “Testler geçiyor ama sistem çalışmıyor” cümlesi her sysadmin’in ve backend geliştiricinin kâbusudur. Bu noktada integration testlerin önemi ortaya çıkıyor. Birim testler tek bir fonksiyonu izole ederek test eder, integration testler ise birden fazla bileşenin birlikte doğru çalışıp çalışmadığını sorgular. Veritabanı, mesaj kuyruğu, harici API, dosya sistemi… Hepsinin bir arada düzgün davranması gerekir.

Bu yazıda pytest ekosistemi üzerine odaklanacağız. Gerçek dünyadan örnekler, gerçek hata senaryoları ve “bunu neden böyle yapıyoruz” açıklamalarıyla dolu bir rehber olacak. Teorik değil, pratik.

Integration Test Nedir, Birim Testten Farkı Ne?

Birim test yazarken dışarıyı mocklamak standart bir pratik. Veritabanı çağrısını mocklarsın, HTTP isteğini mocklarsın, işin bitsin. Ama bu yaklaşımın ciddi bir kör noktası var: mock’ların gerçek sistemi doğru temsil ettiğini varsayıyorsun.

Diyelim ki bir fonksiyon PostgreSQL’e bir kayıt ekliyor ve RETURNING id kullanıyor. Birim testinde mock bu değeri döndürüyor, her şey güzel. Ama production’da yanlış bir column constraint var ve insert başarısız oluyor. Mock bunu yakalamazdı.

Integration testlerin özü şu: gerçek bileşenleri kullan, gerçek hatalar ortaya çıksın. Test veritabanı, test mesaj kuyruğu, test API endpoint’i… Bunlar gerçek sistemlerin birebir kopyaları olmalı.

Ortam Hazırlığı: pytest ve Gerekli Araçlar

Önce temel kurulumu yapalım. Bir proje yapısı oluşturuyoruz:

pip install pytest pytest-asyncio sqlalchemy psycopg2-binary alembic factory-boy faker pytest-postgresql testcontainers

Proje dizin yapısı şöyle olsun:

myapp/
├── models/
│   ├── __init__.py
│   └── user.py
├── repositories/
│   ├── __init__.py
│   └── user_repository.py
├── services/
│   ├── __init__.py
│   └── user_service.py
├── tests/
│   ├── conftest.py
│   ├── integration/
│   │   ├── __init__.py
│   │   ├── test_user_repository.py
│   │   └── test_user_service.py
│   └── unit/
│       └── test_user_logic.py
├── pytest.ini
└── requirements-dev.txt

pytest.ini dosyasında integration testleri ayrı işaretleyelim:

[pytest]
markers =
    integration: Integration testleri, veritabanı bağlantısı gerektirir
    slow: Yavaş çalışan testler
asyncio_mode = auto
testpaths = tests

Bu marker sistemi çok işe yarıyor. CI/CD’de hızlı feedback için sadece unit testleri, gece build’de tüm testleri çalıştırabiliyorsun:

# Sadece unit testler
pytest -m "not integration"

# Sadece integration testler
pytest -m integration

# Tüm testler
pytest

conftest.py: Fixture’ların Kalbi

Integration testlerin gücü fixture’lardan geliyor. conftest.py doğru kurulmazsa test suite’iniz kaotik bir hale gelir. Benim gördüğüm en yaygın hata: her test dosyasında ayrı ayrı veritabanı bağlantısı açmak. Bu hem yavaşlatır hem de izolasyon sorunları yaratır.

# tests/conftest.py

import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from testcontainers.postgres import PostgresContainer

from myapp.models import Base
from myapp.models.user import User


@pytest.fixture(scope="session")
def postgres_container():
    """Session boyunca tek bir PostgreSQL container ayağa kaldır."""
    with PostgresContainer("postgres:15-alpine") as postgres:
        yield postgres


@pytest.fixture(scope="session")
def db_engine(postgres_container):
    """Test engine'i oluştur, tabloları kur."""
    engine = create_engine(postgres_container.get_connection_url())
    Base.metadata.create_all(engine)
    yield engine
    Base.metadata.drop_all(engine)
    engine.dispose()


@pytest.fixture(scope="function")
def db_session(db_engine):
    """Her test için temiz bir session, transaction rollback ile izolasyon."""
    connection = db_engine.connect()
    transaction = connection.begin()
    Session = sessionmaker(bind=connection)
    session = Session()

    yield session

    session.close()
    transaction.rollback()
    connection.close()

Burada kritik bir nokta var: scope="session" ile container bir kez ayağa kalkar, scope="function" ile her test kendi transaction’ını alır ve test bitince rollback yapılır. Bu sayede testler birbirini kirletmez ve her test temiz bir veritabanı üzerinde çalışır.

Testcontainers kullanmak production’a yakın bir ortam sağlar. Docker üzerinde gerçek bir PostgreSQL çalıştırırsın, SQLite gibi “neredeyse aynı ama değil” bir alternatif kullanmazsın. PostgreSQL’e özgü özellikler (JSON operatörleri, diziler, full-text search) gerçekten test edilmiş olur.

Model ve Repository Tanımları

Test yazmadan önce neyi test edeceğimizi tanımlayalım:

# myapp/models/user.py

from datetime import datetime
from sqlalchemy import Column, Integer, String, Boolean, DateTime, func
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, autoincrement=True)
    email = Column(String(255), unique=True, nullable=False, index=True)
    username = Column(String(100), unique=True, nullable=False)
    is_active = Column(Boolean, default=True, nullable=False)
    created_at = Column(DateTime, server_default=func.now(), nullable=False)
    updated_at = Column(DateTime, onupdate=func.now())

    def __repr__(self):
        return f"<User(id={self.id}, email={self.email})>"
# myapp/repositories/user_repository.py

from typing import Optional, List
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError

from myapp.models.user import User


class UserRepository:
    def __init__(self, session: Session):
        self.session = session

    def create(self, email: str, username: str) -> User:
        user = User(email=email, username=username)
        self.session.add(user)
        try:
            self.session.flush()
        except IntegrityError as e:
            self.session.rollback()
            raise ValueError(f"Kullanıcı oluşturulamadı: {e.orig}") from e
        return user

    def get_by_email(self, email: str) -> Optional[User]:
        return self.session.query(User).filter(User.email == email).first()

    def get_active_users(self) -> List[User]:
        return self.session.query(User).filter(User.is_active == True).all()

    def deactivate(self, user_id: int) -> bool:
        user = self.session.query(User).filter(User.id == user_id).first()
        if not user:
            return False
        user.is_active = False
        self.session.flush()
        return True

    def bulk_create(self, users_data: List[dict]) -> List[User]:
        users = [User(**data) for data in users_data]
        self.session.bulk_save_objects(users, return_defaults=True)
        self.session.flush()
        return users

Repository Integration Testleri

Asıl meseleye geldik. Repository testleri yazarken şu soruyu sormak gerekiyor: “Bu test, gerçek bir veritabanı davranışını test ediyor mu?” Eğer cevap evet ise, integration testimiz doğru yoldadır.

# tests/integration/test_user_repository.py

import pytest
from myapp.repositories.user_repository import UserRepository


@pytest.mark.integration
class TestUserRepository:

    def test_kullanici_olustur_ve_geri_al(self, db_session):
        repo = UserRepository(db_session)
        
        user = repo.create(email="[email protected]", username="ali_veli")
        
        assert user.id is not None, "ID atanmış olmalı"
        assert user.email == "[email protected]"
        assert user.is_active is True

    def test_ayni_email_ile_ikinci_kullanici_olusturulamaz(self, db_session):
        repo = UserRepository(db_session)
        repo.create(email="[email protected]", username="ilk_kullanici")
        
        with pytest.raises(ValueError, match="Kullanıcı oluşturulamadı"):
            repo.create(email="[email protected]", username="ikinci_kullanici")

    def test_email_ile_kullanici_bul(self, db_session):
        repo = UserRepository(db_session)
        repo.create(email="[email protected]", username="bulunan_user")
        
        bulunan = repo.get_by_email("[email protected]")
        olmayan = repo.get_by_email("[email protected]")
        
        assert bulunan is not None
        assert bulunan.email == "[email protected]"
        assert olmayan is None

    def test_aktif_kullanicilari_listele(self, db_session):
        repo = UserRepository(db_session)
        repo.create(email="[email protected]", username="aktif_bir")
        repo.create(email="[email protected]", username="aktif_iki")
        user3 = repo.create(email="[email protected]", username="pasif_biri")
        repo.deactivate(user3.id)
        
        aktifler = repo.get_active_users()
        
        assert len(aktifler) == 2
        emailler = [u.email for u in aktifler]
        assert "[email protected]" in emailler
        assert "[email protected]" in emailler
        assert "[email protected]" not in emailler

    def test_toplu_kullanici_olustur(self, db_session):
        repo = UserRepository(db_session)
        veri = [
            {"email": f"user{i}@ornek.com", "username": f"user_{i}"}
            for i in range(50)
        ]
        
        olusturulanlar = repo.bulk_create(veri)
        
        assert len(olusturulanlar) == 50
        db_sayisi = db_session.query(
            __import__('myapp.models.user', fromlist=['User']).User
        ).count()
        assert db_sayisi == 50

Factory Boy ile Test Verisi Üretimi

Testlerde elle veri yazmak zahmetlidir ve test büyüdükçe sürdürülemez hale gelir. factory_boy bu sorunu çözer:

# tests/factories.py

import factory
from factory.alchemy import SQLAlchemyModelFactory
from faker import Faker

from myapp.models.user import User

fake = Faker("tr_TR")


class UserFactory(SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session_persistence = "flush"

    email = factory.LazyFunction(lambda: fake.unique.email())
    username = factory.LazyFunction(lambda: fake.unique.user_name())
    is_active = True


class InactiveUserFactory(UserFactory):
    is_active = False

Factory’leri conftest’e ekleyelim:

# tests/conftest.py'e eklenecek bölüm

from tests.factories import UserFactory, InactiveUserFactory

@pytest.fixture(autouse=True)
def set_factory_session(db_session):
    """Factory'lerin test session'ını kullanmasını sağla."""
    UserFactory._meta.sqlalchemy_session = db_session
    InactiveUserFactory._meta.sqlalchemy_session = db_session
    yield

Artık testler çok daha temiz yazılabiliyor:

# Factory kullanarak yazılmış test

@pytest.mark.integration
def test_servis_yalnizca_aktif_kullanicilara_email_gonder(db_session):
    from tests.factories import UserFactory, InactiveUserFactory
    from myapp.services.user_service import UserService
    
    aktif_kullanici = UserFactory.create_batch(3)
    pasif_kullanici = InactiveUserFactory.create_batch(2)
    
    servis = UserService(session=db_session)
    hedefler = servis.get_email_targets()
    
    hedef_emailler = [u.email for u in hedefler]
    
    for kullanici in aktif_kullanici:
        assert kullanici.email in hedef_emailler
    
    for kullanici in pasif_kullanici:
        assert kullanici.email not in hedef_emailler

Alembic Migration Testleri

Migration’ları test etmek genellikle ihmal edilen ama kritik bir alan. Özellikle ekip büyüdükçe “migration çalışıyor mu” sorusu hayati önem taşır.

# tests/integration/test_migrations.py

import pytest
from alembic.config import Config
from alembic import command
from sqlalchemy import create_engine, inspect


@pytest.mark.integration
def test_migration_basa_alinip_ilerletilebilir(postgres_container):
    """Migration'ların sıfırdan uygulanabildiğini ve geri alınabildiğini test et."""
    engine = create_engine(postgres_container.get_connection_url())
    
    alembic_cfg = Config("alembic.ini")
    alembic_cfg.set_main_option(
        "sqlalchemy.url", 
        postgres_container.get_connection_url()
    )
    
    # Tüm migration'ları uygula
    command.upgrade(alembic_cfg, "head")
    
    inspector = inspect(engine)
    tablolar = inspector.get_table_names()
    
    assert "users" in tablolar, "Users tablosu oluşturulmuş olmalı"
    
    kolonlar = {k["name"] for k in inspector.get_columns("users")}
    beklenen_kolonlar = {"id", "email", "username", "is_active", "created_at"}
    assert beklenen_kolonlar.issubset(kolonlar)
    
    # Bir önceki versiyona dön
    command.downgrade(alembic_cfg, "-1")
    
    # Tekrar en üste çık
    command.upgrade(alembic_cfg, "head")
    
    engine.dispose()

Servis Katmanı Integration Testleri

Repository testleri veritabanı etkileşimini test eder. Servis katmanı testleri ise iş mantığının birden fazla bileşenle doğru çalışıp çalışmadığını sorgular. Bazen bir HTTP isteğini de dahil etmek gerekir:

# tests/integration/test_user_service.py

import pytest
import responses
from myapp.services.user_service import UserService
from tests.factories import UserFactory


@pytest.mark.integration
class TestUserServiceIntegration:

    @responses.activate
    def test_kullanici_kaydi_bildirim_gonderir(self, db_session):
        """Kullanıcı kaydolduğunda bildirim servisine istek atıldığını doğrula."""
        responses.add(
            responses.POST,
            "https://bildirim.internal/api/notify",
            json={"status": "ok"},
            status=200
        )
        
        servis = UserService(
            session=db_session,
            notification_url="https://bildirim.internal/api/notify"
        )
        
        yeni_kullanici = servis.register_user(
            email="[email protected]",
            username="yeni_kullanici"
        )
        
        assert yeni_kullanici.id is not None
        assert len(responses.calls) == 1
        
        istek_body = responses.calls[0].request.body
        assert "[email protected]" in istek_body

    def test_toplu_deaktivasyon_tum_kullanicilari_etkiler(self, db_session):
        """Belirli kriterdeki tüm kullanıcıların deaktive edildiğini test et."""
        import factory
        hedefler = UserFactory.create_batch(5)
        diger = UserFactory.create_batch(3)
        
        hedef_idler = [u.id for u in hedefler]
        
        servis = UserService(session=db_session)
        etkilenen_sayi = servis.bulk_deactivate(user_ids=hedef_idler)
        
        assert etkilenen_sayi == 5
        
        from myapp.repositories.user_repository import UserRepository
        repo = UserRepository(db_session)
        aktifler = repo.get_active_users()
        aktif_idler = [u.id for u in aktifler]
        
        for uid in hedef_idler:
            assert uid not in aktif_idler
        
        for kullanici in diger:
            assert kullanici.id in aktif_idler

Paralel Test Çalıştırma ve Performans

Test suite büyüdükçe çalışma süresi sorun olmaya başlar. pytest-xdist ile paralel çalıştırabilirsiniz:

pip install pytest-xdist

# 4 paralel worker ile çalıştır
pytest -n 4 -m integration

# CPU sayısına göre otomatik ayarla
pytest -n auto -m integration

Ama paralel testlerde dikkatli olmak gerekiyor. Her worker kendi veritabanı bağlantısına ihtiyaç duyar. conftest.py‘deki fixture’ları buna göre düzenlemek şart:

# Paralel test için worker-aware fixture

@pytest.fixture(scope="session")
def worker_id(tmp_path_factory):
    """xdist worker ID'sini al, yoksa 'main' döndür."""
    import os
    return os.environ.get("PYTEST_XDIST_WORKER", "main")


@pytest.fixture(scope="session")
def postgres_container(worker_id):
    """Her worker için ayrı container."""
    with PostgresContainer("postgres:15-alpine") as postgres:
        yield postgres

Test Coverage ve Raporlama

Sadece test yazmak yetmez, neleri test ettiğinizi de bilmeniz gerekir:

pip install pytest-cov

# Coverage raporu ile çalıştır
pytest --cov=myapp --cov-report=html --cov-report=term-missing -m integration

# Minimum coverage threshold belirle
pytest --cov=myapp --cov-fail-under=80

# XML rapor (CI/CD için)
pytest --cov=myapp --cov-report=xml:coverage.xml

pytest.ini‘ye kalıcı coverage ayarları ekleyin:

[pytest]
markers =
    integration: Integration testleri
    slow: Yavaş testler
addopts = --cov=myapp --cov-report=term-missing
testpaths = tests
asyncio_mode = auto

CI/CD Entegrasyonu

GitLab CI örneği, integration testleri ayrı stage’de çalıştırma:

# .gitlab-ci.yml

stages:
  - test-unit
  - test-integration

unit-tests:
  stage: test-unit
  image: python:3.11-slim
  script:
    - pip install -r requirements-dev.txt
    - pytest -m "not integration" --cov=myapp --cov-report=xml
  artifacts:
    reports:
      coverage_report:
        coverage_format: cobertura
        path: coverage.xml

integration-tests:
  stage: test-integration
  image: python:3.11-slim
  services:
    - docker:dind
  variables:
    DOCKER_HOST: tcp://docker:2376
    DOCKER_TLS_CERTDIR: "/certs"
  script:
    - pip install -r requirements-dev.txt
    - pytest -m integration --timeout=120
  only:
    - main
    - merge_requests

Sonuç

Integration test yazmak birim testten daha zahmetlidir: ortam kurmak gerekir, daha yavaş çalışır, daha fazla bakım ister. Ama bunu yapmamak daha pahalıya patlar. “Testler geçti ama production’da patlıyor” durumunun büyük bölümü tam olarak integration testlerin eksikliğinden kaynaklanır.

Önerdiğim yaklaşım: birim testlerle iş mantığını, integration testlerle bileşen etkileşimlerini test edin. İkisine de ihtiyaç var, biri diğerinin yerini tutmuyor.

Testcontainers’ı kesinlikle deneyin. SQLite ile “integration test yazıyorum” demek kendinizi kandırmak. Gerçek PostgreSQL, gerçek Redis, gerçek RabbitMQ kullanın. Fixture scope’larını doğru ayarlayın: container session seviyesinde, transaction her test için. Factory boy kullanın, elle test verisi yazmak ölçeklenmiyor.

Ve son olarak: coverage rakamını takip edin ama ona tapmayın. Yüzde seksen coverage ile kritik kodun test edilmemiş olması, yüzde yüz coverage ile yüzeysel testlerden daha tehlikelidir. Hangi senaryoların kritik olduğuna odaklanın, ona göre test yazın.

Bir yanıt yazın

E-posta adresiniz yayınlanmayacak. Gerekli alanlar * ile işaretlenmişlerdir