Mocking External Dependencies in PySpark Tests

Mocking Nedir ve Neden Hayatidir?

Mocking, bir test sırasında kodunuzun bir parçasını (genellikle bir fonksiyon veya nesneyi) geçici olarak sahte bir versiyonuyla değiştirmektir. Bu sahte versiyon, sizin tam olarak kontrolünüz altındadır.

Peki neden mock kullanırız?

Videonun da vurguladığı gibi, mock'lar sayesinde:

• İzolasyon sağlarız: Testimiz, sadece test ettiğimiz kodun mantığına odaklanır, harici sistemin (örneğin veritabanının) ayakta olup olmamasından etkilenmez.
• Hız kazanırız: Gerçek bir ağ isteği yapmak, veritabanına bağlanmak veya büyük bir dosyayı okumak yavaştır. Mock'lar ise anında sonuç döner.
• Kontrol elde ederiz: Harici bir sistemin farklı durumlarını (örneğin "veritabanı bağlantı hatası" veya "dosya bulunamadı") kolayca simüle edebiliriz.

Projemizi Mocking İçin Hazırlama

Mocking tekniğini etkili bir şekilde uygulayabilmek için kodumuzu test edilebilir bir şekilde yapılandırmamız gerekir. Bunun en iyi yolu, yan etkileri (dosya okuma/yazma, API isteği gibi) olan işlemleri kendi fonksiyonları içine hapsetmektir.

Önceki dersteki src/jobs/user_reporting.py dosyamızı aşağıdaki gibi güncelleyelim. Artık generate_user_report fonksiyonu DataFrame'leri parametre olarak almak yerine, onları okuyan ve yazan fonksiyonları çağıracak.

# src/jobs/user_reporting.py

from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.functions import sum as _sum, col




# Yan Etki 1: Dosya okuma işlemi izole bir fonksiyona alındı.
def read_data(spark: SparkSession, path: str, format: str = "parquet") -> DataFrame:
    """Belirtilen yoldan ve formattan veri okur."""
    print(f"Gerçek okuma fonksiyonu çalışıyor: {path}") # Bu satır testte GÖZÜKMEMELİ
    return spark.read.format(format).load(path)




# Yan Etki 2: Dosya yazma işlemi izole bir fonksiyona alındı.
def write_data(df: DataFrame, path: str, format: str = "parquet"):
    """DataFrame'i belirtilen yola ve formata yazar."""
    print(f"Gerçek yazma fonksiyonu çalışıyor: {path}") # Bu satır testte GÖZÜKMEMELİ
    df.write.mode("overwrite").format(format).save(path)




# Ana ETL mantığı artık yan etkilerden arındırılmış durumda.
def generate_user_report(users_df: DataFrame, purchases_df: DataFrame) -> DataFrame:
    """
    Kullanıcılar ve harcamalarını birleştirip her kullanıcı için toplam harcamayı hesaplar.
    """
    enriched_purchases = purchases_df.join(users_df, "user_id", "inner")
    user_spending = enriched_purchases.groupBy("user_id", "full_name") \
                                      .agg(_sum("price").alias("total_spent"))
    final_report = user_spending.filter(col("total_spent") > 100.0)
    return final_report




# Tüm adımları birleştiren ana fonksiyon (entrypoint)
def run_job(spark: SparkSession, users_path: str, purchases_path: str, output_path: str):
    """ETL işinin ana çalışma fonksiyonu."""



    # 1. Veriyi oku
    users_df = read_data(spark, users_path)
    purchases_df = read_data(spark, purchases_path)




    # 2. Raporu oluştur
    report_df = generate_user_report(users_df, purchases_df)




    # 3. Raporu yaz
    write_data(report_df, output_path)

Bu yapı değişikliği, read_data ve write_data fonksiyonlarını kolayca "mock" etmemize olanak tanıyacak.

pytest-mock ile Okuma İşlemini Mock Etme

pytest ile mock yapmak için pytest-mock eklentisi kullanılır. Eğer yüklü değilse, projenizin terminalinde pip install pytest-mock komutu ile yükleyebilirsiniz. Bu eklenti, test fonksiyonlarımıza mocker adında özel bir fixture enjekte eder.

mocker'ın en güçlü aracı patch'tir. patch, belirttiğiniz bir fonksiyonun yerine geçici olarak bir "mock nesnesi" koyar.

Şimdi tests/test_user_reporting_job.py dosyasını run_job fonksiyonunu test edecek şekilde güncelleyelim.

Bu bilgileri kullanarak testimizi yazalım.

# tests/test_user_reporting_job.py

import pytest
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, DoubleType
from unittest.mock import call  # call nesnesini import ediyoruz




# Test edilecek ana fonksiyonu import et
from src.jobs.user_reporting import run_job




# Önceki dersten gelen kalite kontrol fonksiyonları
from tests.quality_checks import assert_schema_matches




# Okuma işlemini mock'ladığımızda döndürülecek sahte verileri oluşturan bir fixture
@pytest.fixture
def mock_input_data(spark_session: SparkSession):
    """Okuma fonksiyonları mock'landığında döndürülecek sahte DataFrame'leri sağlar."""
    users_schema = StructType([
        StructField("user_id", StringType(), False),
        StructField("full_name", StringType(), True)
    ])
    users_data = [("u1", "Alice Smith"), ("u2", "Bob Johnson")]
    users_df = spark_session.createDataFrame(users_data, users_schema)

    purchases_schema = StructType([
        StructField("purchase_id", StringType(), False),
        StructField("user_id", StringType(), False),
        StructField("price", DoubleType(), True)
    ])
    purchases_data = [("p1", "u1", 200.50), ("p2", "u2", 80.0)]
    purchases_df = spark_session.createDataFrame(purchases_data, purchases_schema)

    return {"users": users_df, "purchases": purchases_df}


def test_run_job_mocks_read_and_write(spark_session, mocker, mock_input_data):
    """
    run_job'un entegrasyon testi.
    Dosya okuma ve yazma işlemleri mock'lanmıştır.
    """



    # 1. GIVEN (Arrange): Mock'ları ve beklenen sonucu ayarla




    # a) `read_data` fonksiyonunu mock'la
    # Bu fonksiyon iki kez çağrılacak: biri kullanıcılar, diğeri harcamalar için.
    # `side_effect` ile her çağrıda farklı bir değer döndürmesini sağlayabiliriz.
    mock_read = mocker.patch(
        "src.jobs.user_reporting.read_data",
        side_effect=[mock_input_data["users"], mock_input_data["purchases"]]
    )




    # b) `write_data` fonksiyonunu mock'la. Bu fonksiyonun hiçbir şey yapmamasını istiyoruz.
    mock_write = mocker.patch("src.jobs.user_reporting.write_data")




    # c) Beklenen çıktıyı hazırla
    expected_schema = StructType([
        StructField("user_id", StringType(), False),
        StructField("full_name", StringType(), True),
        StructField("total_spent", DoubleType(), True)
    ])
    expected_data = [("u1", "Alice Smith", 200.50)]
    expected_df = spark_session.createDataFrame(expected_data, expected_schema)





    # 2. WHEN (Act): ETL işini çalıştır
    users_path = "s3://bucket/users"
    purchases_path = "s3://bucket/purchases"
    output_path = "s3://bucket/report"

    run_job(spark_session, users_path, purchases_path, output_path)





    # 3. THEN (Assert): Mock'ların doğru çağrıldığını ve sonucun doğru olduğunu doğrula




    # a) `read_data` doğru parametrelerle mi çağrıldı?
    expected_read_calls = [
        call(spark_session, users_path),
        call(spark_session, purchases_path)
    ]
    mock_read.assert_has_calls(expected_read_calls, any_order=False)




    # b) `write_data` doğru parametrelerle mi çağrıldı?
    # `assert_called_once_with` doğrudan DataFrame karşılaştırması yapamaz.
    # Bunun yerine, argümanları yakalayıp kendimiz karşılaştıracağız.
    mock_write.assert_called_once() # Sadece bir kez çağrıldığını kontrol et
    



    # Yazma fonksiyonuna gönderilen DataFrame'i al
    # call_args[0] -> pozisyonel argümanlar tuple'ı
    # call_args[0][0] -> ilk pozisyonel argüman (bizim durumumuzda DataFrame)
    actual_output_df = mock_write.call_args[0][0] 
    



    # Şemasını ve içeriğini beklenenle karşılaştır
    assert_schema_matches(actual_output_df, expected_df.schema)
    assert sorted(actual_output_df.collect()) == sorted(expected_df.collect())

Bu testi terminalden pytest ile çalıştırdığınızda, run_job içindeki print ifadelerinin konsolda görünmediğini fark edeceksiniz. Çünkü gerçek read_data ve write_data fonksiyonları asla çalışmadı! Onların yerine bizim kontrolümüzdeki sahte "mock" nesneleri devreye girdi.

Bu yaklaşım, kodunuzun I/O (Giriş/Çıkış) operasyonlarına dokunan kısımlarını, iş mantığınızı (business logic) işleyen saf dönüşüm kısımlarından ayırarak test etmenizi sağlar. Sizin gibi, sürdürülebilir ve kurumsal ölçekte veri akışları tasarlayan bir mühendis için bu ayrım, test edilebilir ve bakımı kolay kod yazmanın temelidir.

Sonuç

Bu derste, testlerimizi harici sistemlerden izole ederek daha hızlı, daha güvenilir ve daha sağlam hale getiren mocking tekniğini öğrendik. Bu, özellikle CI/CD işlem hatları kurarken vazgeçilmez bir yetenektir, çünkü testlerinizin ortamdan bağımsız ve tekrarlanabilir olmasını sağlar.

Temel Çıkarımlar:

• İzolasyon Prensibi: Testler, dosya sistemleri, veritabanları veya API'ler gibi harici bağımlılıklardan izole edilmelidir.
• Test Edilebilir Tasarım: Yan etkileri (I/O işlemleri) olan kod parçalarını ayrı fonksiyonlara ayırmak, onları mock'lamayı kolaylaştırır.
• mocker.patch: pytest-mock kütüphanesinin bu aracı, bir fonksiyonu veya nesneyi test süresince sahte bir versiyonuyla değiştirmemizi sağlar.
• Okuma (Input) Mock'lama: return_value veya side_effect kullanarak, mock'lanan bir fonksiyonun test için hazırladığımız sahte veriyi döndürmesini sağlarız.
• Yazma (Output) Mock'lama: Bir fonksiyonun bir şey döndürmesini beklemediğimizde, assert_called_once() veya assert_called_with() gibi metodlarla doğru parametrelerle çağrılıp çağrılmadığını doğrularız.

Gelecek Dersin Önizlemesi:
Test altyapımızı sağlam bir şekilde kurduk: birim testleri, entegrasyon testleri ve mock'larla harici bağımlılıkların yönetimi. Artık kodumuzun sadece doğru çalışmasını değil, aynı zamanda belirli bir kalite standardına ve stile uymasını da sağlamaya odaklanabiliriz. Bir sonraki dersimizde, kodumuzu otomatik olarak analiz edip biçimlendirmek için güçlü bir araç olan ruff'ı yapılandıracağız.