Bagian 8: Menguji Pipeline

Pipeline yang tidak diuji akan dipercaya sampai hari ia gagal. Pada bagian ini kita menata pengujian dalam tiga lapis sesuai ukuran risikonya: fungsi murni diuji langsung, pemuatan diuji dengan basisdata sementara, dan keseluruhan alur diuji dari berkas CSV sampai isi tabel. Seluruh pengujian memakai pytest yang telah dibahas di Pengujian Kode Python.

Susunan Berkas Pengujian #

Lima berkas di tests/ mengikuti susunan modul di src/:

Berkas	Lapisan	Contoh kasus
`test_validate.py`	unit	email tanpa `@` ditolak, `2024-06-31` ditolak
`test_clean.py`	unit	`210.000` menjadi `210000`, kategori ikut kanonik
`test_dedup.py`	unit	duplikat persis dibuang, konflik dihitung
`test_load.py`	integrasi	upsert dua kali, constraint menolak jumlah nol
`test_pipeline_e2e.py`	end-to-end	delapan baris CSV menjadi empat baris tabel

Berkas conftest.py memastikan direktori root proyek selalu ada di sys.path, sehingga pengujian dapat mengimpor src dari mana pun pytest dijalankan:

python
AKAR_PROYEK = Path(__file__).resolve().parents[1]
if str(AKAR_PROYEK) not in sys.path:
    sys.path.insert(0, str(AKAR_PROYEK))

Pola Pengujian Unit #

Pengujian unit memberi masukan yang tepat dan memeriksa keluaran yang tepat. Contoh dari test_dedup.py:

python
def test_konflik_pertahankan_pertama(self):
    pertama = baris("ORD-2024-0046", jumlah=3)
    revisi = baris("ORD-2024-0046", jumlah=6, status="Diproses")
    hasil, statistik = dedup.dedup_baris([pertama, revisi])
    assert hasil == [pertama]
    assert statistik.konflik == 1
    assert statistik.duplikat_persis == 0

Contoh dari test_load.py memanfaatkan fixture pytest. Fixture db menyiapkan basisdata sementara di tmp_path, membuat skema, lalu menutupnya setelah tiap pengujian:

python
@pytest.fixture()
def db(tmp_path):
    koneksi = sqlite3.connect(tmp_path / "uji.db")
    load.buat_skema(koneksi)
    yield koneksi
    koneksi.close()


def test_upsert_memperbarui_isi(self, db):
    load.muat_batch(db, [baris("ORD-2024-0001", harga=150000)])
    load.muat_batch(db, [baris("ORD-2024-0001", harga=175000)])
    harga = db.execute("SELECT harga FROM pesanan WHERE order_id='ORD-2024-0001'").fetchone()[0]
    assert harga == 175000
    assert load.hitung_baris(db) == 1

Pengujian constraint juga ditulis eksplisit, karena constraint yang tidak pernah teruji belum terbukti ada:

python
def test_check_jumlah_ditolak(self, db):
    with pytest.raises(sqlite3.IntegrityError):
        load.muat_batch(db, [baris("ORD-2024-0001", jumlah=0)])

End-to-End dengan Berkas Kecil #

Pengujian menyeluruh memakai satu berkas delapan baris yang sengaja dibuat: empat baris sehat, satu baris kotor yang valid, satu baris galat email, satu baris galat jumlah, dan dua salinan persis dari baris pertama. Angka yang diharapkan lalu pasti:

8 baris dibaca, 2 ditolak, 6 lolos validasi,
2 duplikat persis dibuang, 4 baris unik,
tabel berisi 4 baris, dan baris kotor tersimpan dalam bentuk bersihnya.

Satu pengujian bahkan memeriksa isi baris sampai kolomnya, untuk membuktikan pembersihan benar-benar terjadi di sepanjang jalan:

python
def test_end_to_end_baris_kotor_tersimpan_bersih(tmp_path, berkas_sumber):
    _, db, _ = jalankan(tmp_path, [berkas_sumber])
    koneksi = sqlite3.connect(db)
    isi = koneksi.execute(
        "SELECT pelanggan, email, kategori, jumlah, harga, status "
        "FROM pesanan WHERE order_id='ORD-2024-0002'"
    ).fetchone()
    assert isi == ("Siti Rahayu", "[email protected]", "Fashion", 3, 275000, "Selesai")

Menjalankan Seluruh Pengujian #

Perintah berikut dijalankan dari root proyek:

bash

python3 -m pytest tests/ -q

Output yang Dihasilkan:

text
........................................................................ [ 80%]
.................                                                        [100%]
89 passed in 0.33s

Angka 89 terbagi rapi per lapisan:

Berkas	Jumlah pengujian
`test_validate.py`	33
`test_clean.py`	17
`test_load.py`	18
`test_dedup.py`	10
`test_pipeline_e2e.py`	11

Untuk mengeksplorasi satu lapisan saja, sebut berkasnya langsung:

bash

python3 -m pytest tests/test_dedup.py -q

Angka 89 adalah jumlah fungsi pengujian, dan durasinya bergerak antar eksekusi. Seluruh pengujian membaca berkas sementara dari tmp_path, bukan berkas asli data/, sehingga pengujian tidak pernah mengubah data proyek. Penyelarasan sys.path di conftest.py membuat perintah ini sah dari direktori mana pun.

Pendalaman pola pengujian pipeline ada di Unit Testing, Pipeline Testing, dan Test Fixtures dan Sample Data. Rujukan resmi: dokumentasi pytest .

Seluruh lapisan telah lulus pengujian dan setiap keputusan meninggalkan jejak. Satu langkah tersisa: menjalankan pipeline utuh pada data asli dan membaca hasilnya. Lanjutkan ke Bagian 9: Menjalankan Pipeline.