Pipeline Testing

Laporan penjualan harian toko online kita dibaca ratusan orang setiap pagi, dan pipeline yang menghasilkan laporan tersebut tidak pernah mengirim satu pun pesan galat. Masalahnya, fungsi transform yang menghitung kolom total bisa salah secara diam-diam: diskon terpotong dua kali, satuan harga tertukar dengan qty, dan tabel hasil tetap terisi penuh tanpa satu baris gagal. Kegalatan seperti ini tidak tertangkap monitoring infrastruktur karena pipeline tampak sukses. Pengujian bertingkat menutup celah tersebut dari tiga arah sekaligus.

Tiga Level Pengujian Pipeline #

Pengujian pipeline paling efektif bila disusun berlapis, karena setiap lapisan menangkap kelas galat yang berbeda. Unit test menguji satu fungsi transform secara murni tanpa I/O — cepat, boleh diperbanyak, dan menunjukkan lokasi galat persis saat logika salah. Integration test menjalankan satu stage utuh dengan I/O kecil yang meniru produksi, misalnya memuat beberapa baris ke SQLite di direktori sementara. End-to-end test menjalankan alur penuh dari berkas sumber sampai tabel akhir, lalu memverifikasi hasil akhirnya.

Level	Cakupan	Contoh pada pipeline pesanan
Unit transform	satu fungsi murni, tanpa I/O	total setelah diskon dihitung benar
Integration stage	satu stage utuh dengan I/O kecil	stage muat mengisi tabel sementara dengan benar
End-to-end run	alur penuh sampai tabel akhir	pipeline penuh lalu smoke query di tabel hasil

Data Assertion dan Smoke Test #

Kode yang lulus uji belum menjamin data yang dihasilkan sehat. Data assertion membalik arah pengujian: alih-alih menguji perilaku fungsi, kita mengklaim properti data yang harus selalu benar — kolom total tidak pernah negatif, jumlah baris per hari masuk akal, dan penjumlahan total untuk satu hari cocok dengan hitung manual. Asersi semacam ini bisa menjadi test pytest atau langkah validasi di dalam pipeline. Smoke test adalah bentuk paling ringan: beberapa query cepat setelah load. Bila smoke test gagal, pipeline wajib menahan tabel hasil agar job downstream tidak membaca data yang patut dicurigai.

Implementasi: pytest untuk Transform dan Smoke Query #

Kita pakai satu berkas pytest untuk ketiga level sekaligus. Fixture bawaan tmp_path menyediakan direktori sementara yang unik per test, sehingga test integration dan end-to-end tidak saling mengganggu dan tidak menyisakan berkas di mesin pengembang.

python
import json
import sqlite3

import pytest

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


def normalisasi_pesanan(baris):
    """Ubah satu baris JSON mentah dari kasir menjadi baris siap muat."""
    data = json.loads(baris)
    for kolom in ("pesanan_id", "waktu", "pelanggan", "produk", "qty", "harga"):
        if kolom not in data:
            raise ValueError(f"kolom wajib hilang: {kolom}")
    if not isinstance(data["qty"], int) or data["qty"] <= 0:
        raise ValueError(f"qty tidak valid: {data['qty']!r}")
    total = data["qty"] * data["harga"] - (data.get("diskon") or 0)
    return (data["pesanan_id"], data["waktu"], data["pelanggan"],
            data["produk"], data["qty"], data["harga"], total)


def muat_pesanan(daftar_baris, berkas_db):
    """Jalankan mini-pipeline: transform tiap baris lalu muat ke SQLite."""
    koneksi = sqlite3.connect(berkas_db)
    koneksi.execute(
        """CREATE TABLE IF NOT EXISTS pesanan (
               pesanan_id INTEGER PRIMARY KEY, waktu TEXT NOT NULL,
               pelanggan TEXT NOT NULL, produk TEXT NOT NULL,
               qty INTEGER NOT NULL, harga INTEGER NOT NULL,
               total INTEGER NOT NULL)"""
    )
    for baris in daftar_baris:
        koneksi.execute(
            "INSERT OR REPLACE INTO pesanan VALUES (?, ?, ?, ?, ?, ?, ?)",
            normalisasi_pesanan(baris),
        )
    koneksi.commit()
    return koneksi


def test_unit_total_setelah_diskon():
    baris = json.dumps({
        "pesanan_id": 7, "waktu": "2026-02-03T09:00:00", "pelanggan": "Rina",
        "produk": "Kain Songket", "qty": 2, "harga": 240000, "diskon": 20000,
    })
    hasil = normalisasi_pesanan(baris)
    assert hasil[6] == 460000


def test_unit_qty_nol_ditolak():
    baris = json.dumps({
        "pesanan_id": 8, "waktu": "2026-02-03T10:00:00", "pelanggan": "Siti",
        "produk": "Rok Tenun", "qty": 0, "harga": 150000, "diskon": 0,
    })
    with pytest.raises(ValueError, match="qty tidak valid"):
        normalisasi_pesanan(baris)


def test_integrasi_stage_muat(tmp_path):
    koneksi = muat_pesanan(BATCH, tmp_path / "toko_uji.db")
    (jumlah,) = koneksi.execute("SELECT COUNT(*) FROM pesanan").fetchone()
    assert jumlah == len(BATCH)
    (total,) = koneksi.execute(
        "SELECT total FROM pesanan WHERE pesanan_id = 1").fetchone()
    assert total == 460000


def test_asersi_data_total_harian(tmp_path):
    koneksi = muat_pesanan(BATCH, tmp_path / "toko_uji.db")
    (baris_buruk,) = koneksi.execute(
        "SELECT COUNT(*) FROM pesanan WHERE total IS NULL OR total <= 0"
    ).fetchone()
    assert baris_buruk == 0
    (total_harian,) = koneksi.execute(
        "SELECT SUM(total) FROM pesanan WHERE waktu LIKE '2026-02-01%'"
    ).fetchone()
    assert total_harian == 2070000


def test_smoke_setelah_load(tmp_path):
    koneksi = muat_pesanan(BATCH, tmp_path / "toko_uji.db")
    (jumlah_produk,) = koneksi.execute(
        "SELECT COUNT(DISTINCT produk) FROM pesanan").fetchone()
    assert jumlah_produk == 5
    (pelanggan,) = koneksi.execute(
        "SELECT pelanggan FROM pesanan WHERE pesanan_id = 3").fetchone()
    assert pelanggan == "Dewi"

Simpan berkas di atas sebagai test_pipeline_toko.py, lalu jalankan pytest. Opsi -v menampilkan nama setiap test; opsi --no-header hanya menyembunyikan baris informasi lingkungan di bagian atas keluaran.

bash

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

Output yang Dihasilkan:

text
============================= test session starts ==============================
collecting ... collected 5 items

test_pipeline_toko.py::test_unit_total_setelah_diskon PASSED             [ 20%]
test_pipeline_toko.py::test_unit_qty_nol_ditolak PASSED                  [ 40%]
test_pipeline_toko.py::test_integrasi_stage_muat PASSED                  [ 60%]
test_pipeline_toko.py::test_asersi_data_total_harian PASSED              [ 80%]
test_pipeline_toko.py::test_smoke_setelah_load PASSED                    [100%]

============================== 5 passed in 0.16s ===============================

Mari kita telaah baris-baris kode di atas:

Baris 6-13 (BATCH): enam baris JSONL dari sistem kasir menjadi data uji bersama — dua hari pesanan dengan diskon bervariasi, termasuk diskon bernilai 0.
Baris 16-26 (normalisasi_pesanan): unit yang paling sering diuji. Fungsi ini memvalidasi kolom wajib dan qty, mengganti diskon null dengan 0, lalu menghitung total.
Baris 29-45 (muat_pesanan): mini-pipeline untuk level integration dan end-to-end — membuat tabel, memanggil transform per baris, memuat dengan INSERT OR REPLACE, lalu commit.
Baris 48-64 (dua test unit): klaim perilaku murni fungsi transform — nilai total setelah diskon, dan penolakan qty 0 lewat pytest.raises.
Baris 66-73 (test_integrasi_stage_muat): satu stage utuh diuji dengan SQLite nyata di tmp_path — jumlah baris dan isi kolom total diperiksa setelah stage muat selesai.
Baris 75-85 (test_asersi_data_total_harian): data assertion — tidak ada total negatif atau null, dan penjumlahan total 1 Februari harus tepat 2.070.000 sesuai hitung manual.
Baris 87-95 (test_smoke_setelah_load): smoke test setelah load — query cepat memastikan tabel hasil benar-benar terisi dan isinya masuk akal sebelum job downstream diizinkan membaca.

Lima test selesai di bawah satu detik. Angka durasi pada baris terakhir berubah-ubah antar eksekusi mengikuti beban mesin, sedangkan jumlah 5 passed adalah bagian yang stabil — sehingga suite ini wajar dijalankan setiap kali kode berubah. Bila kelak ada perubahan yang membuat diskon terpotong dua kali, test_unit_total_setelah_diskon gagal lebih dulu dengan pesan yang menunjukkan nilai aktual di samping nilai yang diharapkan; asersi total harian kemudian menangkap galat agregasi yang lolos dari level unit.

Pengujian memangkas galat sebelum pipeline menyentuh data produksi. Namun di lapangan kegagalan tetap datang dari luar kode: berkas sumber rusak, disk penuh, jaringan terputus. Chapter Pipeline Recovery membahas cara memulihkan pipeline yang gagal di tengah jalan tanpa mengulang seluruh proses dari awal.