Bagian 9: Menguji Workflow

Seluruh bangunan sudah berdiri: DAG dengan lima task, jadwal, retry, backfill, dan pemantauan. Pertanyaan penutup proyek: bagaimana memastikan semuanya tetap benar setelah minggu depan seseorang mengubah aturan pembersihan atau menambah task? Jawabannya adalah strategi pengujian yang menyadari keterbatasan kita: logika diuji sungguhan dengan pytest, struktur DAG diverifikasi lewat ast, dan perilaku berisiko tinggi (backfill) disimulasikan eksplisit. Bagian ini merangkai ketiganya lalu menutup proyek.

Strategi: Uji Logika, Verifikasi Struktur #

Keputusan arsitektur pada Bagian 2 — logika di src/, deklarasi di dags/ — terbayar penuh di sini. Empat berkas pengujian membagi tanggung jawab:

Pembagian ini bukan formalitas; setiap lapisan menangkap kelas galat yang berbeda; kesalahan parsing rupiah tertangkap test_tasks, watermark yang tidak sengaja mundur tertangkap test_state, wiring yang terhapus tertangkap test_dag_structure, dan logika ulang-rentang yang meleset tertangkap test_backfill_sim. Pendekatan bertingkat seperti ini dibahas pada chapter Unit Testing dan Pipeline Testing.

Fondasi: Fixture yang Menumbuhkan Data #

Seluruh uji berangkat dari fixture bersama di tests/conftest.py:

    """Direktori data sementara: raw/, gold/, dan jalur state/state.json."""
    raw = tmp_path / "raw"
    gold = tmp_path / "gold"
    raw.mkdir()
    gold.mkdir()
    return {"raw": raw, "gold": gold, "state": tmp_path / "state" / "state.json"}

Fixture zona memakai tmp_path bawaan pytest: setiap uji mendapat direktori sementara segar yang dihapus otomatis setelah selesai. Tidak ada uji yang menyentuh berkas proyek, tidak ada uji yang saling meninggalkan sisa, dan tidak ada urutan eksekusi yang memengaruhi hasil — tiga sifat yang membuat suite tetap bisa dipercaya berbulan-bulan kemudian. Pola fixture seperti ini dibahas pada chapter Test Fixtures dan Sample Data.

Uji Fungsi Murni #

Karena fungsi di src/ tidak mengimpor Airflow, uji unitnya adalah pytest biasa yang selesai dalam hitungan detik. Contohnya, uji idempotensi eksekusi:

def test_jalankan_etl_idempoten_dilewati(zona):
    tulis_partisi(zona["raw"], "2026-11-02", BARIS_11_02)
    tasks.jalankan_etl(zona["raw"], zona["gold"], zona["state"], "2026-11-02")
    isi_gold = (zona["gold"] / "summary_harian_2026-11-02.csv").read_text(encoding="utf-8")
    hasil = tasks.jalankan_etl(zona["raw"], zona["gold"], zona["state"], "2026-11-02")
    assert hasil.status == "dilewati"
    assert hasil.watermark == "2026-11-02"
    assert hasil.jumlah_valid == 4  # dibaca dari state, bukan dari raw
    # berkas gold tidak tersentuh
    assert (zona["gold"] / "summary_harian_2026-11-02.csv").read_text(encoding="utf-8") == isi_gold

Uji tersebut menumbuhkan partisi, menjalankan ETL dua kali, lalu memastikan eksekusi kedua dilewati dan berkas gold tidak berubah — perilaku yang menjadi fondasi keamanan retry pada Bagian 6. Delapan belas uji sejenis menjaga parse_rupiah, penolakan baris dengan alasan dan nomor baris, agregasi per kota, penulisan gold, hingga format laporan. Sebelas uji di test_state.py menjaga sisi lainnya: roundtrip simpan-baca, penolakan struktur state asing, watermark yang tidak mundur, dan rentang interval yang terbalik.

Uji Struktur DAG tanpa Airflow #

Pengujian DAG biasa dilakukan dengan memuat DAG di environment Airflow sungguhan. Tanpa Airflow, kita mengambil jalan yang lebih sederhana namun tetap berguna: mengurai berkas DAG sebagai pohon sintaks dan menuntut kontrak structural terpenuhi:

def test_argumen_dag_lengkap():
    kwargs = _kwargs_dekorator(_fungsi(_pohon(), "etl_pesanan_tokokita"))
    assert _konstanta(kwargs["schedule"]) == "30 2 * * *"
    assert _konstanta(kwargs["catchup"]) is True
    assert _konstanta(kwargs["max_active_runs"]) == 1
    assert "pendulum" in _konstanta(kwargs["start_date"])
    assert "Asia/Jakarta" in _konstanta(kwargs["start_date"])
    bawaan = kwargs["default_args"]
    assert isinstance(bawaan, ast.Dict)
    isi = {_konstanta(k): _konstanta(v) for k, v in zip(bawaan.keys, bawaan.values)}
    assert isi["retries"] >= 1
    assert "timedelta" in isi["retry_delay"]
    assert "timedelta" in isi["execution_timeout"]

Pendekatan ini jujur tentang batasnya: ast membuktikan bahwa argumen, dekorator, lima task, wiring >>, dan instansiasi ada — bukan bahwa DAG berjalan benar di scheduler. Manfaatnya nyata untuk regresi: hapus catchup=True atau satu sisi >>, dan suite langsung merah. Ketika lingkungan Airflow tersedia, uji pemuatan DAG sungguhan (misalnya lewat pustaka pengujian Airflow) bisa ditambahkan di atas fondasi ini.

Simulasi Backfill sebagai Uji Perilaku #

Berkas terakhir menguji perilaku yang melintasi beberapa tanggal — wilayah yang paling rawan kesalahan off-by-one:

def test_backfill_mengisi_celah_setelah_scheduler_mati(tiga_partisi):
    """Skenario: 02 sudah sukses, lalu scheduler mati satu hari."""
    zona = tiga_partisi
    tasks.jalankan_etl(zona["raw"], zona["gold"], zona["state"], "2026-11-02")
    hasil = tasks.jalankan_backfill(
        zona["raw"], zona["gold"], zona["state"], "2026-11-02", "2026-11-04"
    )
    assert [h.tanggal for h in hasil] == ["2026-11-03", "2026-11-04"]

Docstring dan nama berkasnya menegaskan status: ini simulasi lokal berlabel jelas, bukan eksekusi Airflow. Tujuh uji di berkas tersebut mencakup urutan tiga tanggal, catatan state per tanggal, eksekusi ulang yang tidak mengerjakan apa pun, force yang menghasilkan gold identik, kegagalan di tengah rentang, dan watermark yang maju selangkah per tanggal.

Suite Lengkap dan Pemeliharaan #

Seluruh pengujian dijalankan dari direktori proyek:

python3 -m pytest tests/ -q

Output yang Dihasilkan:

..............................................                           [100%]
46 passed in 0.28s

Angka 46 terbagi atas 18 uji fungsi ETL, 11 uji state, 10 uji struktur DAG, dan 7 uji simulasi backfill. Aturan pemeliharaannya bisa diringkas jadi tabel:

Rujukan lengkap ada di dokumentasi pytest serta modul ast Python .

Penutup Proyek #

Sembilan bagian membawa TokoKita dari pipeline manual menjadi workflow yang diorkestrasi: masalah operasional dipetakan menjadi DAG, logika ETL dipisah agar teruji cepat, dependency dan scheduling dinyatakan eksplisit, kegagalan ditangani retry dengan jaminan idempotensi, waktu yang terlewat dikejar backfill, dan seluruh eksekusi bisa dipantau dari state maupun UI. Kontrak verifikasi yang jujur — nyatakan apa yang dijalankan, jelaskan apa yang tidak — membuat proyek ini tetap dapat dipelihara meskipun ditulis tanpa Airflow terpasang.

Hingga di sini data kita masih bergerak harian: satu partisi per hari, diproses satu kali per malam. Dunia nyata sering menuntut lebih cepat dari itu — pesanan masuk dan harus direaksi dalam hitungan detik. Proyek berikutnya, Proyek: Streaming Data dengan Kafka, memindahkan TokoKita ke paradigma itu: data mengalir sebagai stream ke topic Kafka, dan pola watermark yang kita kenal di sini bertemu konsep offset dan consumer group.