pytest ile Asenkron Kod Testi: asyncio Entegrasyonu

Yazar:komut.io ekibi
0 Yorum

Asenkron kod yazmak bir süre sonra alışkanlık haline geliyor. Ama o kodu test etmek? İşte orada çoğu kişi tökezliyor. “await şunu yaptım, test geçti” diyorsunuz ama aslında test hiçbir şey yapmamış. Bu yazıda pytest ile asyncio entegrasyonunu, gerçek senaryolar üzerinden, acı çekerek öğrendiğim yollarla anlatacağım.

Neden Asenkron Test Zordur

Senkron kodda pytest’in işi basittir: fonksiyonu çağır, sonucu al, assert et. Ama asenkron fonksiyonlar bir coroutine objesi döndürür, doğrudan çalıştırılmazlar. Şöyle bir şey yazarsanız:

def test_veri_cek():
    sonuc = veri_cek_async()  # Bu bir coroutine objesi döndürür
    assert sonuc is not None  # Bu her zaman geçer, yanlış bir test!

Buradaki tehlike şu: test gerçekten “geçiyor” çünkü sonuc bir coroutine objesi ve None değil. Ama asenkron kod hiç çalışmadı. Production’da hata alırsınız, testte görmezsiniz. Bu tam anlamıyla “yanlış güven” senaryosu.

pytest-asyncio Kurulumu ve Temel Yapı

İlk adım pytest-asyncio paketini kurmak:

pip install pytest-asyncio
# veya pyproject.toml / requirements-dev.txt'e ekleyin
pip install pytest>=7.0 pytest-asyncio>=0.21

Kurulumdan sonra pytest.ini veya pyproject.toml dosyanıza asyncio modunu tanımlamanız gerekiyor. Bu adımı atlarsanız her test için ayrı ayrı dekoratör kullanmak zorunda kalırsınız:

# pytest.ini
[pytest]
asyncio_mode = auto

# ya da pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"

asyncio_mode = auto ayarı, async def ile tanımlanmış tüm test fonksiyonlarını otomatik olarak asyncio event loop’u içinde çalıştırır. Bu, özellikle büyük projelerde hayat kurtarır.

Eğer auto modunu kullanmak istemiyorsanız, her test fonksiyonuna @pytest.mark.asyncio dekoratörü eklemeniz gerekir:

import pytest

@pytest.mark.asyncio
async def test_basit_asenkron():
    import asyncio
    await asyncio.sleep(0.1)
    assert True

Gerçek Dünya Senaryosu: HTTP İstemcisi Testi

Bir izleme servisi yazıyorsunuz diyelim. Dış API’lara istek atıyor ve yanıtları veritabanına kaydediyor. Bu servisi test etmek için dış API’ya gerçekten istek atmak istemiyorsunuz: hem yavaş, hem güvenilmez, hem de rate limit problemi yaratır.

Önce servis koduna bakalım:

# servis.py
import httpx
import asyncio
from typing import Optional

class IzlemeServisi:
    def __init__(self, base_url: str, timeout: int = 30):
        self.base_url = base_url
        self.timeout = timeout
        self._client: Optional[httpx.AsyncClient] = None

    async def __aenter__(self):
        self._client = httpx.AsyncClient(timeout=self.timeout)
        return self

    async def __aexit__(self, *args):
        if self._client:
            await self._client.aclose()

    async def sunucu_durumu_kontrol(self, endpoint: str) -> dict:
        yanit = await self._client.get(f"{self.base_url}/{endpoint}")
        yanit.raise_for_status()
        return {
            "durum": yanit.status_code,
            "yanit_suresi": yanit.elapsed.total_seconds(),
            "veri": yanit.json()
        }

    async def toplu_kontrol(self, endpointler: list) -> list:
        gorevler = [self.sunucu_durumu_kontrol(ep) for ep in endpointler]
        return await asyncio.gather(*gorevler, return_exceptions=True)

Şimdi bunu test edelim. httpx‘in kendi AsyncMock desteğini ve pytest-httpx gibi araçları kullanabilirsiniz, ama burada unittest.mock ile nasıl yapılacağını gösterelim:

# test_servis.py
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from servis import IzlemeServisi

@pytest.fixture
async def izleme_servisi():
    async with IzlemeServisi("https://api.example.com") as servis:
        yield servis

@pytest.mark.asyncio
async def test_sunucu_durumu_kontrol_basarili():
    mock_yanit = MagicMock()
    mock_yanit.status_code = 200
    mock_yanit.elapsed.total_seconds.return_value = 0.123
    mock_yanit.json.return_value = {"durum": "aktif", "versiyon": "1.0"}

    with patch("httpx.AsyncClient.get", new_callable=AsyncMock) as mock_get:
        mock_get.return_value = mock_yanit

        async with IzlemeServisi("https://api.example.com") as servis:
            sonuc = await servis.sunucu_durumu_kontrol("health")

        assert sonuc["durum"] == 200
        assert sonuc["yanit_suresi"] == 0.123
        assert sonuc["veri"]["durum"] == "aktif"

AsyncMock ile Dikkat Edilmesi Gerekenler

unittest.mock.AsyncMock Python 3.8 ile geldi. Ondan önce farklı yaklaşımlar kullanılıyordu. Eğer hala 3.7 kullanan bir projeniz varsa (ki 2025’te varsa ciddi bir sorun bu), ayrı bir async mock tanımlamanız gerekir.

AsyncMock’un senkron Mock’tan farkı şu: çağrıldığında bir coroutine döndürür. Yani await mock_func() doğru çalışır. Eğer normal MagicMock kullansaydınız await beklediğiniz gibi çalışmaz ve garip hatalar alırdınız.

import pytest
from unittest.mock import AsyncMock, patch

async def veritabani_kaydet(veri: dict) -> bool:
    # Normalde gerçek DB işlemi
    pass

async def veri_isle_ve_kaydet(ham_veri: str) -> bool:
    islenmis = {"deger": ham_veri.upper(), "uzunluk": len(ham_veri)}
    return await veritabani_kaydet(islenmis)

@pytest.mark.asyncio
async def test_veri_isle_ve_kaydet():
    with patch("__main__.veritabani_kaydet", new_callable=AsyncMock) as mock_db:
        mock_db.return_value = True

        sonuc = await veri_isle_ve_kaydet("merhaba")

        assert sonuc is True
        mock_db.assert_called_once_with({"deger": "MERHABA", "uzunluk": 7})

Fixture’larda Async Kullanımı

Async fixture’lar özellikle veritabanı bağlantıları, Redis bağlantıları veya test ortamı kurulumu için kritik. Şöyle düşünün: her test öncesinde test veritabanını temizlemek ve sonrasında bağlantıyı düzgünce kapatmak istiyorsunuz.

import pytest
import asyncpg

@pytest.fixture(scope="session")
async def veritabani_havuzu():
    """Session boyunca tek bir connection pool kullan"""
    havuz = await asyncpg.create_pool(
        host="localhost",
        database="test_db",
        user="test_user",
        password="test_pass",
        min_size=2,
        max_size=10
    )
    yield havuz
    await havuz.close()

@pytest.fixture(autouse=True)
async def temiz_veritabani(veritabani_havuzu):
    """Her test öncesinde tabloları temizle"""
    async with veritabani_havuzu.acquire() as baglanti:
        await baglanti.execute("TRUNCATE TABLE kullanicilar CASCADE")
        await baglanti.execute("TRUNCATE TABLE loglar CASCADE")
    yield
    # Test sonrası ek temizlik gerekiyorsa buraya ekle

@pytest.fixture
async def ornek_kullanici(veritabani_havuzu):
    async with veritabani_havuzu.acquire() as baglanti:
        kullanici_id = await baglanti.fetchval(
            "INSERT INTO kullanicilar (ad, email) VALUES ($1, $2) RETURNING id",
            "Test Kullanici",
            "[email protected]"
        )
    return {"id": kullanici_id, "ad": "Test Kullanici", "email": "[email protected]"}

Burada scope="session" kullanımına dikkat edin. Bağlantı havuzu pahalı bir kaynak, her test için yeniden oluşturmak testlerinizi gereksiz yere yavaşlatır. Ama tabloları temizleyen fixture her test için çalışmalı, autouse=True bunu sağlıyor.

Eş Zamanlılık Testleri: Race Condition’ları Yakalamak

Bu kısım çoğu test kılavuzunda eksik kalıyor. Asenkron kodun en kritik test senaryosu, eş zamanlı erişim durumlarıdır. Bir cache mekanizması yazdığınızı düşünün:

import asyncio
import pytest
from typing import Dict, Optional

class AkilliCache:
    def __init__(self):
        self._depo: Dict[str, str] = {}
        self._yukleniyor: Dict[str, asyncio.Event] = {}

    async def getir(self, anahtar: str, yukleyici) -> str:
        if anahtar in self._depo:
            return self._depo[anahtar]

        if anahtar in self._yukleniyor:
            # Başka bir coroutine zaten yüklüyor, bekle
            await self._yukleniyor[anahtar].wait()
            return self._depo[anahtar]

        event = asyncio.Event()
        self._yukleniyor[anahtar] = event

        try:
            deger = await yukleyici(anahtar)
            self._depo[anahtar] = deger
            return deger
        finally:
            event.set()
            del self._yukleniyor[anahtar]


@pytest.mark.asyncio
async def test_ayni_anda_coklu_istek():
    """Aynı anahtara eş zamanlı istekler geldiğinde yükleyici sadece bir kez çağrılmalı"""
    cache = AkilliCache()
    cagri_sayisi = 0

    async def pahali_yukleyici(anahtar: str) -> str:
        nonlocal cagri_sayisi
        cagri_sayisi += 1
        await asyncio.sleep(0.05)  # Pahalı işlem simülasyonu
        return f"deger_{anahtar}"

    # 10 eş zamanlı istek gönder
    gorevler = [cache.getir("test_anahtari", pahali_yukleyici) for _ in range(10)]
    sonuclar = await asyncio.gather(*gorevler)

    assert all(s == "deger_test_anahtari" for s in sonuclar)
    assert cagri_sayisi == 1, f"Yükleyici {cagri_sayisi} kez çağrıldı, 1 kez çağrılmalıydı"

Bu test, cache mekanizmanızın “thundering herd” problemini çözüp çözmediğini doğrular. Senkron testle bunu yakalamak neredeyse imkansızdı.

Timeout Testleri

Asenkron kodda timeout yönetimi kritik. Bir istek sonsuza kadar bekleyebilir, bunu test etmeniz gerekiyor:

import asyncio
import pytest

async def zaman_asimi_olan_islem(sure: float) -> str:
    try:
        async with asyncio.timeout(2.0):  # Python 3.11+
            await asyncio.sleep(sure)
            return "tamamlandi"
    except asyncio.TimeoutError:
        return "zaman_asimi"

@pytest.mark.asyncio
async def test_normal_islem_tamamlanir():
    sonuc = await zaman_asimi_olan_islem(0.5)
    assert sonuc == "tamamlandi"

@pytest.mark.asyncio
async def test_uzun_islem_zaman_asimina_ugrarar():
    sonuc = await zaman_asimi_olan_islem(3.0)
    assert sonuc == "zaman_asimi"

@pytest.mark.asyncio
async def test_islem_gercekten_iptal_edildi():
    """asyncio.CancelledError düzgün handle ediliyor mu?"""
    iptal_edildi = False

    async def iptal_izleme():
        nonlocal iptal_edildi
        try:
            await asyncio.sleep(10)
        except asyncio.CancelledError:
            iptal_edildi = True
            raise

    gorev = asyncio.create_task(iptal_izleme())
    await asyncio.sleep(0.1)
    gorev.cancel()

    try:
        await gorev
    except asyncio.CancelledError:
        pass

    assert iptal_edildi is True

Event Loop Scope Yönetimi

pytest-asyncio 0.21 sonrasında event loop scope’u daha açık biçimde kontrol edebilirsiniz. Bu önemli çünkü bazı durumlarda session-scoped fixture’ların function-scoped event loop ile çakışması sorun yaratır:

# conftest.py
import pytest
import asyncio

# Tüm session için tek event loop kullan
@pytest.fixture(scope="session")
def event_loop():
    """Session boyunca tek event loop"""
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

Ya da pyproject.toml‘da:

[tool.pytest.ini_options]
asyncio_mode = "auto"
# pytest-asyncio 0.23+ için
asyncio_default_fixture_loop_scope = "session"

Bu ayar olmadan, session-scoped async fixture kullandığınızda şöyle bir uyarı görürsünüz: PytestUnraisableExceptionWarning: Exception ignored in... Bu uyarıyı görmezden gelmek yerine, event loop scope’unu doğru ayarlayın.

Test Performansı: Paralel Async Testler

Binlerce async testiniz varsa ve her biri gerçek I/O bekliyorsa (mock kullanmadan entegrasyon testi yapıyorsanız), testleriniz yavaşlayabilir. pytest-xdist ile paralel çalıştırabilirsiniz ama dikkatli olun: her worker kendi event loop’unu kullanır, paylaşılan kaynaklar (veritabanı, Redis) için dikkatli izolasyon gerekir.

Daha pratik bir yaklaşım, async testlerin içinde asyncio.gather kullanmak:

import pytest
import asyncio
import httpx

@pytest.mark.asyncio
async def test_coklu_endpoint_kontrol():
    """Tüm endpointleri paralel test et"""
    async def endpoint_kontrol(client: httpx.AsyncClient, url: str) -> tuple:
        yanit = await client.get(url)
        return url, yanit.status_code

    test_urller = [
        "https://httpbin.org/get",
        "https://httpbin.org/ip",
        "https://httpbin.org/user-agent"
    ]

    async with httpx.AsyncClient(timeout=10.0) as client:
        gorevler = [endpoint_kontrol(client, url) for url in test_urller]
        sonuclar = await asyncio.gather(*gorevler)

    for url, durum in sonuclar:
        assert durum == 200, f"{url} beklenen 200, alınan {durum}"

Bu şekilde üç ayrı HTTP isteği paralel gider ve toplam süre en yavaş isteğin süresi kadar olur, üç katı değil.

Hata Senaryoları ve Exception Testleri

Async kodda exception’ların doğru fırlatıldığını test etmek de biraz farklı:

import pytest
import asyncio

class VeriCekmeHatasi(Exception):
    def __init__(self, url: str, durum_kodu: int):
        self.url = url
        self.durum_kodu = durum_kodu
        super().__init__(f"{url} adresinden veri çekilemedi: HTTP {durum_kodu}")

async def guvenli_veri_cek(url: str) -> dict:
    # Basitleştirilmiş örnek
    if "hata" in url:
        raise VeriCekmeHatasi(url, 500)
    return {"veri": "ok"}

@pytest.mark.asyncio
async def test_hata_durumu_dogru_exception():
    with pytest.raises(VeriCekmeHatasi) as exc_info:
        await guvenli_veri_cek("https://hata.example.com/api")

    assert exc_info.value.durum_kodu == 500
    assert "hata.example.com" in exc_info.value.url

@pytest.mark.asyncio
async def test_gather_hatalari_toplar():
    """return_exceptions=True ile hatalar exception yerine sonuç olarak döner"""
    async def basarili(): return "ok"
    async def basarisiz(): raise ValueError("bir hata")

    sonuclar = await asyncio.gather(
        basarili(),
        basarisiz(),
        basarili(),
        return_exceptions=True
    )

    assert sonuclar[0] == "ok"
    assert isinstance(sonuclar[1], ValueError)
    assert str(sonuclar[1]) == "bir hata"
    assert sonuclar[2] == "ok"

Pratik Öneriler ve Sık Yapılan Hatalar

Yıllar içinde gördüğüm yaygın hatalar ve çözümleri:

  • Event loop’u elle kapatmayın: pytest-asyncio bunu yönetir. loop.close() çağırmak testlerin geri kalanını bozar.
  • asyncio.run() test içinde kullanmayın: Test zaten bir event loop içinde çalışıyor, iç içe event loop çalışmaz.
  • Mock’ları coroutine döndürür yapın: MagicMock() yerine AsyncMock() kullanın. Bu hata saatlerce hata ayıklamaya yol açabilir.
  • async for ve async with mock’lamak: __aiter__, __anext__, __aenter__, __aexit__ metodlarını da mock’lamanız gerekir. AsyncMock bunu otomatik halleder.
  • Fixture scope’larını tutarlı tutun: Session-scoped async fixture, function-scoped event loop ile çalışmaz.
  • Gerçek zamanlama testlerinden kaçının: asyncio.sleep(0.1) ve sonra 0.05‘ten fazla sürdü mü diye kontrol etmeyin. CI sunucuları yavaş olabilir.
  • anyio alternatifini değerlendirin: trio veya farklı async backend’ler kullanıyorsanız, anyio ve pytest-anyio daha esnek bir çözüm sunabilir.

Sonuç

Async kod testi başlangıçta karmaşık görünebilir ama temelde birkaç ilkeye dayanıyor: event loop’u pytest-asyncio’ya bırakın, AsyncMock’u doğru kullanın, fixture scope’larına dikkat edin ve eş zamanlılık senaryolarını ihmal etmeyin.

En büyük kazanım şu: async testleri yazmaya başladıktan sonra, senkron kodda fark etmediğiniz pek çok race condition ve hata yönetimi sorununu önceden yakalarsınız. Bir servisin production’da “zaman zaman çöküyor” şikayetini almak yerine, o senaryoyu test ortamında yeniden üretip düzeltebilirsiniz.

Kendi projelerimde şu yapıyı standart olarak kullanıyorum: asyncio_mode = auto, session-scoped event loop, veritabanı bağlantıları için session-scoped havuz, her test için temizlik fixture’ı. Bu yapı kurulduktan sonra async test yazmak senkron test yazmaktan daha zor değil.

Son bir not: pytest-asyncio sürekli gelişen bir kütüphane. 0.21 ile gelen değişiklikler, 0.23 ile gelenler önemli. CHANGELOG‘u takip edin ve sürüm kısıtlamalarınızı requirements-dev.txt‘te net belirtin.

Bir yanıt yazın

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