Data Transformation Testing

Chapter Unit Testing menguji fungsi field satu baris; chapter ini naik satu tingkat: tabel masuk, tabel keluar. Transformasi agregasi, dedup, dan penyaringan referensi bekerja pada ratusan ribu baris sekaligus, sehingga kesalahan kecil berlipat menjadi selisih besar di laporan. Pendekatan kita adalah asersi berbasis data: tabel input dan tabel expected output diperlakukan sebagai data uji, lalu klaim dinyatakan di dua level — per baris dan per himpunan. Batasnya juga perlu jelas sejak awal: ini berbeda dari Data Validation yang menjalankan aturan pada data produksi saat runtime; transformation testing menguji kode pada data uji terkendali sebelum kode menyentuh produksi.

Input dan Expected Output sebagai Data Uji #

Tabel uji yang baik sengaja memuat jebakan. Enam baris pesanan berikut berisi dua kasus yang paling sering merusak agregasi di dunia nyata: pesanan 2 muncul dua kali karena revisi, dan pesanan 4 merujuk pelanggan 999 yang tidak terdaftar.

Dengan aturan transformasi — duplikat diselesaikan dengan revisi terakhir, pesanan tanpa referensi pelanggan sah dibuang, lalu agregasi per tanggal — expected output yang benar adalah tabel berikut. Kedua tabel inilah yang nantinya dituangkan ke dalam test.

Transformasi yang Diuji #

Transformasi ditulis dengan pandas dan berupa fungsi murni: DataFrame pesanan masuk, DataFrame harian keluar. Kontrak skema keluaran dideklarasikan sebagai konstanta SKEMA_OUTPUT agar drift guard punya satu sumber kebenaran. Operasi groupby dan agg yang dipakai di dalamnya terdokumentasi lengkap di dokumentasi pandas .

"""Transformasi pesanan menjadi agregat harian: dedup + referential integrity."""
import pandas as pd

# Kontrak skema output: nama kolom dan tipe yang dijaga test drift guard.
SKEMA_OUTPUT = [
    ("tanggal", "object"),
    ("jumlah_pesanan", "int64"),
    ("total_penjualan", "int64"),
]


def transform_harian(pesanan: pd.DataFrame, pelanggan_sah: set) -> pd.DataFrame:
    """Agregasi pesanan per tanggal setelah dedup dan saring referensi."""
    bersih = pesanan.drop_duplicates(subset="pesanan_id", keep="last")
    bersih = bersih[bersih["pelanggan_id"].isin(pelanggan_sah)]
    harian = (
        bersih.groupby("tanggal", as_index=False)
        .agg(jumlah_pesanan=("pesanan_id", "count"), total_penjualan=("total", "sum"))
    )
    return harian.sort_values("tanggal").reset_index(drop=True)

Asersi Level Baris dan Level Himpunan #

Test suite menerjemahkan kedua tabel di atas menjadi klaim pytest. Level baris memeriksa nilai satu baris keluaran lewat parametrize; level himpunan memeriksa properti seluruh tabel; dua test terakhir mengunci perilaku jebakan dan skema.

"""Data transformation test: asersi level baris, level himpunan, drift guard."""
import pandas as pd
import pytest

from transform_harian import SKEMA_OUTPUT, transform_harian

PESANAN = pd.DataFrame(
    [
        [1, "2026-06-01", 101, 460000],
        [2, "2026-06-01", 102, 850000],
        [3, "2026-06-02", 101, 300000],
        [2, "2026-06-01", 102, 900000],  # revisi pesanan 2: duplikat pesanan_id
        [4, "2026-06-02", 999, 125000],  # pelanggan tak terdaftar
        [5, "2026-06-02", 103, 240000],
    ],
    columns=["pesanan_id", "tanggal", "pelanggan_id", "total"],
)

PELANGGAN_SAH = {101, 102, 103}


@pytest.mark.parametrize(
    "tanggal, jumlah, total",
    [("2026-06-01", 2, 1360000), ("2026-06-02", 2, 540000)],
)
def test_baris_harian_level_baris(tanggal, jumlah, total):
    hasil = transform_harian(PESANAN, PELANGGAN_SAH)
    baris = hasil[hasil["tanggal"] == tanggal].iloc[0]
    assert baris["jumlah_pesanan"] == jumlah
    assert baris["total_penjualan"] == total


def test_invarian_level_himpunan():
    hasil = transform_harian(PESANAN, PELANGGAN_SAH)
    assert len(hasil) == 2
    assert hasil["tanggal"].is_unique
    assert hasil["total_penjualan"].sum() == 1900000
    assert hasil["jumlah_pesanan"].sum() == 4


def test_dedup_memilih_revisi_terakhir():
    hasil = transform_harian(PESANAN, PELANGGAN_SAH)
    baris = hasil[hasil["tanggal"] == "2026-06-01"].iloc[0]
    assert baris["total_penjualan"] == 1360000  # 460000 + 900000, bukan 850000


def test_referensi_tak_sah_dibuang():
    hasil = transform_harian(PESANAN, PELANGGAN_SAH)
    baris = hasil[hasil["tanggal"] == "2026-06-02"].iloc[0]
    assert baris["jumlah_pesanan"] == 2  # pesanan pelanggan 999 tidak ikut terhitung
    assert 125000 not in hasil["total_penjualan"].tolist()


def test_skema_output_tidak_bergeser():
    hasil = transform_harian(PESANAN, PELANGGAN_SAH)
    assert list(zip(hasil.columns, hasil.dtypes.astype(str))) == SKEMA_OUTPUT

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

Output yang Dihasilkan:

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

test_transform_harian.py::test_baris_harian_level_baris[2026-06-01-2-1360000] PASSED [ 16%]
test_transform_harian.py::test_baris_harian_level_baris[2026-06-02-2-540000] PASSED [ 33%]
test_transform_harian.py::test_invarian_level_himpunan PASSED            [ 50%]
test_transform_harian.py::test_dedup_memilih_revisi_terakhir PASSED      [ 66%]
test_transform_harian.py::test_referensi_tak_sah_dibuang PASSED          [ 83%]
test_transform_harian.py::test_skema_output_tidak_bergeser PASSED        [100%]

============================== 6 passed in 1.31s ==============================

Mari kita telaah baris-baris kode di atas:

1. Baris 7-19 (PESANAN dan PELANGGAN_SAH): tabel uji enam baris memuat dua jebakan yang disengaja — pesanan 2 dengan dua total berbeda, dan pesanan 4 milik pelanggan 999. Setiap jebakan punya test yang menjawabnya.
2. Baris 22-30 (level baris): parametrize memetakan dua baris expected output menjadi dua test; hasil[hasil["tanggal"] == tanggal].iloc[0] mengambil satu baris keluaran untuk tanggal yang sedang diuji.
3. Baris 33-38 (level himpunan): empat properti seluruh tabel diperiksa sekaligus — dua tanggal unik, kolom tanggal tanpa duplikat, penjumlahan total_penjualan 1900000, dan penjumlahan jumlah_pesanan 4.
4. Baris 41-44 (dedup): total 1 Juni tepat 1360000 membuktikan keep="last" memilih revisi 900000, bukan nilai awal 850000.
5. Baris 47-51 (referensi): tanggal 2 Juni hanya menghitung dua pesanan, dan nilai 125000 dari pelanggan tak terdaftar tidak pernah muncul di kolom mana pun.
6. Baris 54-56 (drift guard): pasangan kolom dan tipe data di-zip lalu dibandingkan dengan SKEMA_OUTPUT — satu sumber kebenaran yang juga dibaca pembaca modul.

Tiga Level Asersi dan Guard Schema Drift #

Schema drift adalah perubahan bentuk data yang datang diam-diam dari hulu: kolom berganti nama, tipe berubah dari integer menjadi string, kolom baru muncul tanpa pemberitahuan. Drift guard di atas menangkapnya di sisi keluaran transformasi — test gagal cepat sebelum tabel rusak menyebar ke mart. Untuk drift di sisi ingestion dan strategi evolusi skema jangka panjang, pembahasan tuntasnya ada di chapter Schema Evolution dan Schema Validation.

Gagasan Property-Based Testing #

Semua test di atas bersifat example-based: kita menulis contoh tetap dan pytest memeriksanya. Property-based testing membalik logika tersebut — kita menyatakan properti yang harus benar untuk semua input, lalu generator membuat ratusan input acak dan mencari satu pun yang melanggar properti. Contoh properti untuk transform_harian: jumlah_pesanan keluaran tidak pernah melebihi jumlah baris masukan; penjumlahan total_penjualan tidak pernah melebihi penjumlahan total baris yang sah; dan kolom tanggal keluaran selalu unik.

Di ekosistem Python, library Hypothesis adalah standar untuk pola ini dan terintegrasi mulus dengan pytest — generator-nya bahkan mampu mengecilkan contoh yang gagal menjadi kasus paling sederhana yang masih melanggar properti. Contoh tetap dan properti bukan pilihan salah satu: contoh tetap menjaga nilai bisnis penting yang sudah diketahui, sedangkan properti menjaga kita dari kombinasi input yang belum terpikirkan.

Dengan asersi level baris, himpunan, dan skema, transformasi kita terkunci dari tiga arah. Pertanyaan berikutnya adalah komposisi: berapa banyak test di tiap level, data uji seperti apa yang dipakai, dan kapan seluruh suite boleh menghentikan merge. Chapter Pipeline Testing di section ini menyusun metodologi lengkapnya.