ClickHouse kullanan Python kodunu nasıl test ederiz?

Selamlar, bu yazımda ClickHouse'a bağlı Python kodunu nasıl test ettiğimize bakacağız. Konu kulağa basit gelse de iş başa düşünce 'şu sorgu prod'da gerçekten doğru sonucu mu döndürüyor?' sorusu epey sinir bozucu olabiliyor. Lafı uzatmadan başlayayım.

ClickHouse, OLAP tarafında biz Python'cuların sevdiği bir veritabanı. Ama test yazma kültürü hâlâ çoğunlukla 'mock'la geç, hayatına devam et' tarafında. Bence bu eksik bir yaklaşım. Çünkü ClickHouse'un asıl değerli olduğu yerler (aggregate fonksiyonlar, materialized view'lar, uniq, quantile gibi yapılar) mock'larla test edildiğinde aslında hiç test edilmemiş oluyor.

Üç katmanlı bir yaklaşım

Bana sorarsanız ClickHouse bağımlı kodu üç farklı seviyede test etmek lazım:

Birim testler: ClickHouse istemcisini mock'larız, sadece iş mantığını test ederiz. Hızlı ama sorgunun kendisini garanti etmez.
Entegrasyon testleri: Testcontainers ile gerçek bir ClickHouse instance'ı kaldırırız. Yavaş ama dürüst.
Kontrat testleri: Sorgu çıktısının beklenen şemaya uyduğunu doğrularız.

Üçü birbirini tamamlıyor; birini diğerine tercih etmeyin.

Mock ile birim test

Diyelim ki günlük aktif kullanıcı sayısını hesaplayan basit bir fonksiyonumuz var:

# app.py
def get_daily_active_users(client, date: str) -> int:
    result = client.query(
        f"SELECT uniq(user_id) FROM events WHERE toDate(event_time) = '{date}'"
    )
    return result.first_row[0]

Test tarafında istemciyi MagicMock ile sahteliyoruz:

# test_app.py
from unittest.mock import MagicMock
from app import get_daily_active_users

def test_daily_active_users_returns_int():
    mock_client = MagicMock()
    mock_result = MagicMock()
    mock_result.first_row = [42]
    mock_client.query.return_value = mock_result

    count = get_daily_active_users(mock_client, '2026-03-31')

    assert count == 42
    assert '2026-03-31' in mock_client.query.call_args[0][0]

Burada test ettiğimiz şey çok dar: fonksiyon first_row[0]'ı doğru döndürüyor mu, tarihi sorguya gerçekten yerleştiriyor mu? Bu kadar. uniq fonksiyonunun doğru çalışıp çalışmadığını test etmiyoruz. Açıkçası ben bu seviyeyi çoğunlukla SQL injection riskini görmek için kullanıyorum.

Testcontainers ile gerçek ClickHouse

Şimdi işin gerçek kısmı. testcontainers paketiyle her test oturumu için tek bir ClickHouse container'ı kaldırıyoruz:

pip install testcontainers[clickhouse] clickhouse-connect pytest

conftest.py içinde fixture'larımızı tanımlıyoruz:

# conftest.py
import pytest
import clickhouse_connect
from testcontainers.clickhouse import ClickHouseContainer

@pytest.fixture(scope='session')
def ch_client():
    with ClickHouseContainer('clickhouse/clickhouse-server:latest') as ch:
        client = clickhouse_connect.get_client(
            host=ch.get_container_host_ip(),
            port=int(ch.get_exposed_port(8123)),
        )
        client.command('''
            CREATE TABLE events (
                user_id UInt64,
                event_time DateTime DEFAULT now(),
                event_type String
            ) ENGINE = MergeTree() ORDER BY (user_id, event_time)
        ''')
        yield client

scope='session' olması kritik. Her test için container kaldırırsanız test süitiniz yarım saatte biter; oturum başına bir kez kaldırırsanız 5-10 saniyelik bir sabit maliyetiniz olur, sonrası akar gider.

Testin kendisi sade:

# test_integration.py
def test_insert_and_count(ch_client):
    ch_client.insert(
        'events',
        [[1, 'login'], [2, 'click']],
        column_names=['user_id', 'event_type'],
    )
    result = ch_client.query('SELECT count() FROM events')
    assert result.first_row[0] == 2

Materialized view'ları test etmek

İşin sinsi tarafı burası. Materialized view'lar arka planda merge bekliyor; insert sonrası hemen sorgu atarsanız beklenen sonucu göremeyebilirsiniz. OPTIMIZE ... FINAL ile merge'ü zorlamak gerekiyor:

def test_event_counts_materialized_view(ch_client):
    ch_client.command('''
        CREATE MATERIALIZED VIEW event_counts
        ENGINE = SummingMergeTree() ORDER BY event_type
        AS SELECT event_type, count() AS cnt
        FROM events GROUP BY event_type
    ''')
    ch_client.insert('events', [[3, 'login']], column_names=['user_id', 'event_type'])
    ch_client.command('OPTIMIZE TABLE event_counts FINAL')

    result = ch_client.query(
        "SELECT cnt FROM event_counts WHERE event_type = 'login'"
    )
    assert result.first_row[0] >= 1

Sık karşılaşılan hatalar

Her test için yeni container kaldırmak: scope='function' kullanırsanız süit dakikalarca akar. Oturum bazlı fixture kullanın, test başında TRUNCATE TABLE ile temizleyin.
Materialized view'da merge'ü beklememek: Insert sonrası hemen SELECT atarsanız sonuç boş gelebilir. OPTIMIZE TABLE <View> FINAL çağırmayı unutmayın.
f-string ile sorgu kurmak: Yukarıdaki örnekte tarihi f-string ile yerleştirdik; gerçek hayatta parameters= argümanını kullanın, yoksa SQL injection kapısı açıyorsunuz.
Sadece mock ile yetinmek: uniq, quantile, argMax gibi ClickHouse'a özgü fonksiyonların doğru sonuç verdiğini ancak gerçek bir instance ile doğrulayabilirsiniz.

Kapanış

Bu yazıda ClickHouse'a bağımlı Python kodunu mock'tan Testcontainers'a, oradan da materialized view senaryolarına kadar üç farklı seviyede nasıl test ettiğimize baktık. Şahsi kanaatim, mock'lara fazla yaslanmadan en azından kritik sorguları gerçek bir container'a karşı çalıştırmak şart; oturum kapsamlı fixture'larla bunun maliyeti de görünenden çok daha düşük. Umarım faydalı olur, bir sonraki yazıda görüşmek üzere.

Üç katmanlı bir yaklaşım#

Mock ile birim test#

Testcontainers ile gerçek ClickHouse#

Materialized view'ları test etmek#

Sık karşılaşılan hatalar#

Kapanış#