Bagian 9: Menguji Pipeline

Pipeline ingestion berjalan tanpa pengawasan, sering kali tengah malam, dan kegagalannya baru terasa ketika laporan di hilir kehilangan data. Pengujian yang baik memindahkan penemuan masalah ke sebelum kode berjalan, bukan setelahnya. Proyek ini memakai tiga lapis pengujian: unit test untuk fungsi murni, integrasi dengan fake server di localhost, dan end-to-end yang menjalankan pipeline utuh. Seluruhnya sudah kita isi bertahap pada setiap bagian; sekarang kita rangkum mesinnya dan jalankan semuanya.

Ancangan Fake Server #

Kunci dari seluruh suite adalah fixture di tests/conftest.py yang menyalakan server sungguhan di port acak:

@pytest.fixture()
def fake_api():
    """Server fake di port acak; otomatis dimatikan setelah tiap tes."""
    with FakeTokoKitaAPI() as api:
        yield api


@pytest.fixture()
def klien(fake_api):
    """ApiClient dengan backoff kecil agar tes tidak menunggu lama."""
    policy = RetryPolicy(max_retries=3, backoff_base=0.05,
                         backoff_cap=0.2, jitter=0.0)
    return ApiClient(fake_api.url, timeout=5.0, policy=policy)

Mari kita telaah baris-baris kode di atas:

1. Baris 22-23 (with ... yield api): Server menyala sebelum tes berjalan dan padam setelah tes selesai; pola yield pada fixture pytest menjamin pembersihan bahkan ketika tes gagal.
2. Port acak (FakeTokoKitaAPI()): Setiap tes mendapat port kosong dari sistem operasi, sehingga tes tidak pernah berebut port dan bisa berjalan paralel.
3. Baris 29-30 (backoff_base=0.05): Kebijakan retry asli menunggu setengah detik; tes menyuntik kebijakan cepat agar suite selesai dalam hitungan detik tanpa mengubah kode produksi.

Karena server adalah proses HTTP sungguhan, tes membaca perilaku apa adanya lewat status_log, last_headers, dan request_count yang dicatat server. Dua tes menunggu satu detik penuh sesuai Retry-After, sisanya memakai monkeypatch untuk merekam durasi tunggu.

Lapisan Pengujian #

Delapan berkas tes memetakan perilaku pipeline dari unit terkecil hingga run utuh:

Tes end-to-end membangun pipeline utuh dengan direktori sementara tmp_path, lalu memeriksa hasilnya sampai ke isi berkas. Berikut tes run full yang membandingkan isi zona raw dengan dataset server:

def test_run_full_menulis_seluruh_pesanan(tmp_path):
    with FakeTokoKitaAPI() as api:
        pipeline = buat_pipeline(api, tmp_path)
        hasil = pipeline.run(mode="full")

        assert hasil.baris_diambil == 137
        assert hasil.baris_valid == 137
        assert hasil.baris_dilewati == 0
        assert read_jsonl(hasil.output) == api.pesanan
        kursor = max(baris["updated_at"] for baris in api.pesanan)
        assert hasil.kursor == kursor
        assert load_state(tmp_path / "state.json") == {
            "updated_at": kursor, "last_run_id": hasil.run_id,
            "last_mode": "full",
        }

Pemeriksaan read_jsonl(hasil.output) == api.pesanan adalah jaminan terkuat di seluruh suite: seluruh perjalanan data, dari handler HTTP hingga berkas JSONL, tidak mengubah satu byte informasi pun. Pembagian lapisan semacam ini mengikuti chapter Unit Testing, Integration Testing, dan End-to-End Testing pada bagian pengujian course.

Menjalankan Seluruh Suite #

Dari akar bundle, satu perintah menjalankan semuanya:

python3 -m pytest tests/ -q

Output yang Dihasilkan:

....................................................                     [100%]
52 passed in 8.34s

Untuk melihat nama tes satu per satu, jalankan berkas end-to-end dengan opsi verbose:

python3 -m pytest tests/test_pipeline_e2e.py -v

Output yang Dihasilkan (potongan):

tests/test_pipeline_e2e.py::test_run_full_menulis_seluruh_pesanan PASSED [ 16%]
tests/test_pipeline_e2e.py::test_run_full_gagal_cepat_pada_baris_rusak PASSED [ 33%]
tests/test_pipeline_e2e.py::test_run_skip_melewati_baris_rusak PASSED    [ 50%]
tests/test_pipeline_e2e.py::test_incremental_tanpa_state_bertindak_seperti_full PASSED [ 66%]
tests/test_pipeline_e2e.py::test_incremental_hanya_mengambil_data_baru PASSED [ 83%]
tests/test_pipeline_e2e.py::test_incremental_tanpa_perubahan_tidak_menulis_berkas PASSED [100%]

Durasi run bervariasi sedikit antar eksekusi karena dua tes menunggu Retry-After sesaat, dan itu satu-satunya bagian suite yang bergantung pada waktu nyata. Enam tes end-to-end merangkum seluruh proyek: full load menulis 137 baris, baris rusak memicu gagal cepat atau dilewati sesuai kebijakan, run incremental tanpa state menjadi full load, data baru terambil selektif, dan run tanpa perubahan tidak menulis berkas apa pun.

Kriteria Selesai dan Cara Memutar Ulang #

Proyek dinyatakan selesai bila tiga kondisi terpenuhi dan dapat diperiksa ulang kapan saja:

• python3 -m pytest tests/ -q melaporkan seluruh tes lulus tanpa galat.
• python3 -m src.pipeline --mode full --demo menghasilkan berkas JSONL berisi 137 baris di data/raw/.
• python3 -m src.pipeline --mode incremental --demo mengambil nol baris setelah kursor terisi, lalu mengambil hanya baris baru ketika sumber berubah.

Seluruh artefak yang Anda lihat di halaman-halaman proyek ini, dari ringkasan run hingga isi config/state.json, dihasilkan oleh perintah-perintah tersebut di lingkungan penulisan. Dokumentasi pytest yang dipakai sebagai acara fixture, monkeypatch, dan tmp_path tersedia di docs.pytest.org , dan strategi pengujian pipeline secara umum dibahas pada chapter Pipeline Testing.

Dengan pipeline ingestion yang tervalidasi penuh, fondasi data mentah kita siap dipakai lebih jauh. Data di zona raw masih perlu dibersihkan, digabungkan, dan dimuat ke tabel analitik secara terjadwal, tepat seperti yang akan kita kerjakan pada proyek berikutnya, Proyek: Pipeline ETL Terjadwal.