End-to-End Testing

Chapter Pipeline Testing pada section Data Pipeline Engineering menutup dengan satu test end-to-end: jalankan mini-pipeline penuh lalu smoke query ke tabel hasil. Pendekatan cepat itu sudah menangkap galat alur, tetapi pada pipeline produksi, end-to-end testing adalah disiplin tersendiri. Kita perlu memilih data apa yang layak dipakai, menentukan hasil apa yang dianggap benar, dan mengelola kenyataan bahwa test ini paling lambat sekaligus paling dipercaya. Chapter ini membahas strategi di baliknya: lingkungan uji terisolasi, smoke dataset versi tetap, perbandingan golden output, dan ambang penerimaan.

Empat Keputusan Sebelum Menulis Test #

Keputusan pertama adalah lingkungan uji. Test end-to-end menjalankan alur sungguhan, sehingga wajib bekerja di ruang kerja terisolasi yang dibuat segar per run — tidak berbagi direktori, tidak berbagi database, dan tidak menyisakan berkas di mesin pengembang. Fixture tmp_path pytest memenuhi ketiganya dengan satu baris.

Keputusan kedua adalah smoke dataset: kumpulan data kecil yang isinya tetap dan ikut berversi bersama kode. Dataset yang baik tidak sekadar lima baris rapi — dataset di chapter ini sengaja memuat status pending yang harus dikarantina, dua tanggal berbeda, dan diskon bernilai nol, karena kasus tepi inilah yang membedakan alur sungguhan dari happy path. Keputusan ketiga adalah golden output: hasil run yang sudah disetujui pemilik data, disimpan sebagai referensi, lalu dibandingkan secara persis dengan hasil setiap run. Keputusan keempat adalah ambang penerimaan — untuk smoke dataset kita memakai perbandingan persis, tetapi untuk data yang isinya berubah tiap run, ambang berbasis toleransi lebih realistis: jumlah baris minimal, atau selisih total penjualan terhadap run sebelumnya tidak melebihi persentase tertentu. Ambang semacam ini sekeluarga dengan aturan di chapter Data Validation, dengan satu beda utama: di sini aturan dievaluasi pada data uji terkendali saat pengembangan, bukan pada data produksi saat runtime.

Pipeline Mini: dari CSV Mentah ke Mart #

Objek uji adalah pipeline tiga tahap pada satu berkas: extraction membaca CSV ekspor kasir, transformation membersihkan tiap baris dan mengarantina status pending, load menulis mart penjualan harian ke SQLite.

"""Mini pipeline TokoKita: CSV mentah -> pembersihan -> mart SQLite."""
import csv
import sqlite3


def bersihkan_pesanan(baris):
    """Transformation: validasi satu baris; kembalikan None bila dikarantina."""
    if baris["status"].strip().lower() == "pending":
        return None  # status belum final: karantina, jangan masuk mart
    total = int(baris["qty"]) * int(baris["harga"]) - int(baris["diskon"] or 0)
    return (baris["waktu"][:10], total)


def jalankan_pipeline(berkas_csv, berkas_db):
    """End-to-end: extraction -> transformation -> load, kembalikan ringkasan."""
    with open(berkas_csv, newline="", encoding="utf-8") as berkas:
        mentah = list(csv.DictReader(berkas))  # extraction: baca CSV mentah
    bersih = [b for b in (bersihkan_pesanan(r) for r in mentah) if b is not None]

    harian = {}
    for tanggal, total in bersih:
        jumlah, total_harian = harian.get(tanggal, (0, 0))
        harian[tanggal] = (jumlah + 1, total_harian + total)

    koneksi = sqlite3.connect(berkas_db)
    koneksi.execute(
        "CREATE TABLE IF NOT EXISTS mart_penjualan_harian ("
        "tanggal TEXT PRIMARY KEY, jumlah_pesanan INTEGER, total_penjualan INTEGER)"
    )
    koneksi.execute("DELETE FROM mart_penjualan_harian")
    koneksi.executemany(
        "INSERT INTO mart_penjualan_harian VALUES (?, ?, ?)",
        [(t, v[0], v[1]) for t, v in sorted(harian.items())],
    )
    koneksi.commit()
    koneksi.close()
    return {"baris_mentah": len(mentah), "baris_dimuat": len(bersih)}

Ringkasan run sengaja dibawa keluar dari pipeline: jumlah baris mentah dan jumlah baris yang dimuat memberi angka cepat untuk diperiksa tanpa membuka tabel. Mart ditulis dengan pola replace penuh — DELETE sebelum INSERT — sehingga hasil run selalu bergantung pada input saja, bukan pada isi mart sebelumnya.

Test End-to-End dengan Golden Output #

Test berikut menjalankan pipeline dari berkas CSV sungguhan sampai tabel mart sungguhan, lalu membandingkan dua hal: isi mart terhadap golden output, dan ringkasan run terhadap angka yang dihitung manual.

"""End-to-end test: smoke dataset tetap -> pipeline penuh -> golden output."""
import sqlite3

import pytest

from pipeline_toko import jalankan_pipeline

SMOKE_CSV = """pesanan_id,waktu,pelanggan,produk,qty,harga,diskon,status
1,2026-05-01T09:10:00,Budi,Kemeja Batik,2,240000,20000,lunas
2,2026-05-01T11:30:00,Siti,Kain Songket,1,850000,0,lunas
3,2026-05-01T15:45:00,Dewi,Rok Tenun,1,310000,10000,BELUM LUNAS
4,2026-05-02T08:05:00,Rina,Blouse Katun,2,120000,0,dikirim
5,2026-05-02T13:15:00,Budi,Seluar Linen,3,175000,25000,lunas
6,2026-05-02T18:20:00,Siti,Kemeja Batik,1,240000,0,pending
"""

GOLDEN_MART = [
    ("2026-05-01", 3, 1610000),
    ("2026-05-02", 2, 740000),
]


@pytest.fixture
def smoke_dataset(tmp_path):
    """Tulis smoke dataset versi tetap ke ruang kerja terisolasi."""
    berkas_csv = tmp_path / "pesanan_mentah.csv"
    berkas_csv.write_text(SMOKE_CSV, encoding="utf-8")
    return berkas_csv


def baca_mart(berkas_db):
    koneksi = sqlite3.connect(berkas_db)
    baris = koneksi.execute(
        "SELECT tanggal, jumlah_pesanan, total_penjualan "
        "FROM mart_penjualan_harian ORDER BY tanggal"
    ).fetchall()
    koneksi.close()
    return baris


def test_e2e_mart_sama_dengan_golden(smoke_dataset, tmp_path):
    jalankan_pipeline(smoke_dataset, tmp_path / "mart.db")
    assert baca_mart(tmp_path / "mart.db") == GOLDEN_MART


def test_e2e_ringkasan_mencatat_karantina(smoke_dataset, tmp_path):
    ringkasan = jalankan_pipeline(smoke_dataset, tmp_path / "mart.db")
    assert ringkasan == {"baris_mentah": 6, "baris_dimuat": 5}

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

Output yang Dihasilkan:

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

test_e2e.py::test_e2e_mart_sama_dengan_golden PASSED                     [ 50%]
test_e2e.py::test_e2e_ringkasan_mencatat_karantina PASSED                [100%]

============================== 2 passed in 0.12s ==============================

Mari kita telaah baris-baris kode di atas:

1. Baris 8-15 (SMOKE_CSV): enam baris pesanan berisi kasus tepi yang disengaja — dua tanggal, diskon nol, status kapital campur, dan satu baris pending yang wajib dikarantina. Isinya tetap, sehingga hasil golden stabil selamanya.
2. Baris 17-20 (GOLDEN_MART): golden output berupa daftar baris mart yang disetujui. Pada pipeline lebih besar, golden disimpan sebagai berkas tersendiri di version control dan dimuat oleh test — pola yang dibahas di chapter Test Fixtures dan Sample Data.
3. Baris 23-28 (fixture smoke_dataset): smoke dataset ditulis ke tmp_path, jadi setiap test end-to-end memperoleh berkas CSV dan database miliknya sendiri; dua test bisa berjalan paralel tanpa saling menimpa.
4. Baris 41-48 (dua test): test pertama membandingkan seluruh isi mart terhadap golden — tanggal, jumlah pesanan, dan total penjualan sekaligus. Test kedua membandingkan ringkasan run: enam baris mentah, lima baris dimuat, yang membuktikan karantina benar-benar terjadi di jalur nyata.

Trade-off End-to-End #

End-to-End test lambat karena menjalankan semuanya: pembacaan berkas, pembersihan, agregasi, sampai tulis tabel. Kepercayaan yang diberikan juga proporsional — hanya test inilah yang membuktikan ketiga tahap benar-benar tersambung dari ujung ke ujung. Karena itu komposisinya dibalik: sedikit end-to-end yang luas cakupannya, banyak unit dan integration yang sempit tapi cepat.

End-to-End test memverifikasi alur, tetapi asersi di dalamnya masih berupa angka ringkas: jumlah baris dan total penjualan. Bab berikutnya membedah asersi data itu sendiri — bagaimana tabel input dan tabel expected output dijadikan bahan uji per baris dan per himpunan — pada chapter Data Transformation Testing.