Pipeline Testing

Chapter Pipeline Testing pada section Data Pipeline Engineering memperkenalkan tiga level pengujian — unit, integration, dan end-to-end — lewat satu berkas test tunggal. Bab-bab sebelumnya di section ini kemudian membedah tiap level: Unit Testing untuk logika field, Integration Testing untuk sambungan antar komponen, End-to-End Testing untuk alur penuh dengan golden output, dan Data Transformation Testing untuk asersi data tiga level. Pertanyaan yang tersisa adalah metodologi: bagaimana keempatnya disusun menjadi suite yang sehat untuk pipeline produksi. Chapter ini menyusun jawabannya — komposisi piramida, daftar periksa per stage, strategi data uji, penanganan flaky test, dan gerbang continuous integration.

Piramida Test untuk Pipeline #

Komposisi yang sehat berbentuk piramida: sangat banyak unit test di dasar, cukup banyak integration test di tengah, dan sedikit end-to-end test di puncak. Setiap lapisan menumpuk di atas lapisan di bawahnya — end-to-end yang sedikit berdiri di atas integration yang cukup banyak, dan integration berdiri di atas unit yang sangat banyak.

Dua alasan menjaga bentuk ini. Pertama, kecepatan: dasar piramida berjalan dalam milidetik sehingga pengembang menjalankannya puluhan kali sehari, sedangkan puncak yang mahal dijalankan sesekali. Kedua, lokasi kegagalan: test unit yang gagal menunjuk satu fungsi dalam hitungan detik, sedangkan test end-to-end yang gagal hanya menyisakan jejak alur yang harus ditelusuri. Angka komposisi seperti 70-20-10 adalah orientasi, bukan target ketat — yang penting gradiennya: makin ke atas makin sedikit dan makin mahal. Ada satu penyesuaian khas data: asersi data level himpunan dari chapter Data Transformation Testing berjalan cepat tanpa I/O, sehingga asersi tersebut termasuk wajar dihimpun di dasar piramida.

Apa yang Diuji di Setiap Stage #

Dua baris tabel itu sering terlewat. Extraction jarang diuji padahal kontraknya patah paling sering — perubahan di pihak sumber datang tanpa permisi; cara mengujinya dibahas tuntas di chapter Contract Testing. Jalur alerting juga wajib diuji: pipeline yang gagal tetapi tidak mengirim sinyal sama berbahayanya dengan pipeline yang salah hitung diam-diam, dan perilaku ini menyambung ke materi Metrics dan Alerting.

Strategi Data Uji #

Ketiganya dipakai berlapis: sample kecil berversi sebagai smoke dataset seperti pada chapter End-to-End Testing, synthetic untuk volume dan kasus tepi, seeded saat test perlu ribuan baris yang sama di setiap mesin. Pembahasan lengkap pembuatan dan pengelolaannya ada di chapter Test Fixtures dan Sample Data.

Flaky Test pada Konteks Data #

Flaky test adalah test yang lulus dan gagal bergantian tanpa perubahan kode. Pipeline data punya sumber flakiness sendiri: urutan baris yang tidak deterministik dari paralelisasi atau iterasi himpunan, timestamp dan timezone yang ikut berubah setiap run, serta data eksternal yang isinya memang berubah. Mitigasinya lugas — urutkan hasil sebelum membandingkan (ORDER BY di SQL, sort_values() di pandas), bekukan waktu dengan tanggal tetap, dan isolasi setiap test di tmp_path seperti pada bab-bab sebelumnya.

Aturan mainnya tegas: flaky test ditandai, diselidiki, dan diperbaiki — bukan dijalankan ulang sampai lulus. Kebiasaan rerun-sampai-lulus mengajari tim bahwa warna merah boleh diabaikan, dan tepat pada saat itu test yang benar-benar rusak ikut terabaikan.

Gerbang CI dan Smoke Test Idempotensi #

Gerbang penggabungan kode disusun berjenjang: suite cepat (unit dan integration) wajib lulus sebelum setiap merge, end-to-end dijalankan sekali per pull request dan sekali per malam untuk regression penuh. Setelah pipeline berjalan di produksi, smoke test membaca ringkasan run dan beberapa asersi cepat — bila smoke gagal, tabel hasil ditahan agar job downstream tidak membaca data yang patut dicurigai. Otomasi gerbang ini dibahas tuntas di chapter Continuous Integration; contoh berikut menunjukkan smoke test yang memeriksa dua hal sekaligus: ringkasan angka dan idempotensi load.

"""Mini pipeline untuk smoke test: CSV pesanan -> mart SQLite yang idempoten."""
import csv
import sqlite3


def jalankan_pipeline(berkas_csv, berkas_db):
    """Muat CSV pesanan ke mart; jalankan ulang harus memberi state identik."""
    koneksi = sqlite3.connect(berkas_db)
    koneksi.execute(
        "CREATE TABLE IF NOT EXISTS mart_pesanan ("
        "pesanan_id INTEGER PRIMARY KEY, tanggal TEXT, pelanggan TEXT, total INTEGER)"
    )
    with open(berkas_csv, newline="", encoding="utf-8") as berkas:
        for baris in csv.DictReader(berkas):
            total = int(baris["qty"]) * int(baris["harga"]) - int(baris["diskon"] or 0)
            koneksi.execute(
                "INSERT OR REPLACE INTO mart_pesanan VALUES (?, ?, ?, ?)",
                (int(baris["pesanan_id"]), baris["waktu"][:10], baris["pelanggan"], total),
            )
    koneksi.commit()
    (jumlah, total_semua) = koneksi.execute(
        "SELECT COUNT(*), COALESCE(SUM(total), 0) FROM mart_pesanan"
    ).fetchone()
    koneksi.close()
    return {"jumlah_baris": jumlah, "total_penjualan": total_semua}

"""Smoke test pipeline: ringkasan stabil dan idempoten setelah muat ulang."""
import sqlite3

from pipeline_harian import jalankan_pipeline

SMOKE_CSV = """pesanan_id,waktu,pelanggan,qty,harga,diskon
1,2026-07-01T09:00:00,Budi,2,240000,20000
2,2026-07-01T11:00:00,Siti,1,230000,0
3,2026-07-02T08:30:00,Dewi,1,310000,10000
4,2026-07-02T10:45:00,Rina,2,120000,0
"""


def test_smoke_ringkasan_dan_idempotensi(tmp_path):
    berkas_csv = tmp_path / "pesanan.csv"
    berkas_csv.write_text(SMOKE_CSV, encoding="utf-8")
    berkas_db = tmp_path / "mart.db"

    ringkasan_awal = jalankan_pipeline(berkas_csv, berkas_db)
    ringkasan_ulang = jalankan_pipeline(berkas_csv, berkas_db)

    diharapkan = {"jumlah_baris": 4, "total_penjualan": 1230000}
    assert ringkasan_awal == diharapkan
    assert ringkasan_ulang == diharapkan  # muat ulang tidak menduplikasi baris


def test_smoke_asersi_data_cepat(tmp_path):
    berkas_csv = tmp_path / "pesanan.csv"
    berkas_csv.write_text(SMOKE_CSV, encoding="utf-8")
    berkas_db = tmp_path / "mart.db"
    jalankan_pipeline(berkas_csv, berkas_db)

    koneksi = sqlite3.connect(berkas_db)
    (total_terkecil,) = koneksi.execute("SELECT MIN(total) FROM mart_pesanan").fetchone()
    (jumlah_tanggal,) = koneksi.execute(
        "SELECT COUNT(DISTINCT tanggal) FROM mart_pesanan"
    ).fetchone()
    koneksi.close()
    assert total_terkecil > 0
    assert jumlah_tanggal == 2

python3 -m pytest -v --no-header test_smoke_pipeline.py

Output yang Dihasilkan:

============================= test session starts ==============================
collecting ... collected 2 items

test_smoke_pipeline.py::test_smoke_ringkasan_dan_idempotensi PASSED      [ 50%]
test_smoke_pipeline.py::test_smoke_asersi_data_cepat PASSED              [100%]

============================== 2 passed in 0.11s ==============================

Mari kita telaah baris-baris kode di atas:

1. Baris 8-19 di pipeline (INSERT OR REPLACE): muat ulang berkas yang sama tidak menduplikasi baris karena primary key pesanan_id menimpa baris lama — inilah yang membuat idempotensi load bisa diuji, bukan sekadar diharapkan.
2. Baris 21-23 (ringkasan): COUNT dan SUM dibaca setelah commit sehingga ringkasan mencerminkan state akhir tabel, bukan antrean yang belum selesai ditulis.
3. Baris 14-24 (test idempotensi): pipeline dijalankan dua kali pada berkas dan database yang sama; dua ringkasan harus identik dengan angka yang dihitung manual — empat baris, total 1230000.
4. Baris 27-40 (test asersi cepat): dua query ringkas memeriksa sifat data — tidak ada total nol atau negatif, dan tanggal yang muncul memang dua hari sesuai dataset.

Metodologi ini berdiri di atas satu fondasi yang belum kita bedah: kesepakatan eksplisit dengan setiap sumber data tentang bentuk yang boleh datang. Chapter Contract Testing membahas cara menguji kesepakatan tersebut sebelum perubahan pihak sumber menjadi insiden di pipeline kita.