Integration Testing

Chapter Unit Testing membuktikan logika parsing dan mapping berjalan benar dalam isolasi sempurna. Padahal kegagalan pipeline paling sering lahir di sambungan antar komponen: SQL join yang kehilangan baris karena referensi tidak cocok, nama kolom mart yang bergeser, respons API eksternal yang berubah bentuk. Integration test menguji sambungan tersebut dengan komponen sungguhan — mesin SQLite, schema tabel, serialization JSON — tetapi dalam ukuran kecil yang selesai dalam hitungan detik. Kita tetap memakai pytest dari chapter Pengujian Kode Python; yang baru hanyalah kerelaan menyentuh I/O nyata secara terkendali.

Arti Integration Test pada Pipeline Data #

Integration test menjalankan dua komponen atau lebih secara bersamaan, termasuk jalur I/O di antara keduanya. Untuk pipeline data, kerangka berpikirnya adalah kontrak: kesepakatan tentang bentuk data yang melintasi setiap batas — kolom apa saja yang hadir, tipe apa yang dipakai, dan berapa baris yang berhak lolos. Kontrak yang paling sering patah di TokoKita ada pada tiga batas: stage tables ke transformasi SQL, transformasi SQL ke tabel mart, serta pipeline ke API eksternal. Dua batas pertama kita uji dengan SQLite nyata di direktori sementara; batas ketiga ditiru lewat unittest.mock dari pustaka standar agar test tidak bergantung pada jaringan.

Sambungan Pertama: Stage Tables ke Mart #

Objek uji pertama adalah muat_mart.py, satu stage transform-plus-load yang membaca dua stage tables lalu mengisi tabel mart penjualan per kota.

python
"""Stage tables -> transform SQL -> mart: objek uji integration test."""
import sqlite3

KUERI_TRANSFORM = """
INSERT OR REPLACE INTO mart_penjualan_kota
SELECT substr(p.waktu, 1, 10) AS tanggal,
       c.kota,
       COUNT(*) AS jumlah_pesanan,
       SUM(p.total) AS total_penjualan
FROM stg_pesanan p
JOIN stg_pelanggan c ON c.pelanggan_id = p.pelanggan_id
GROUP BY tanggal, c.kota
"""


def bangun_mart(berkas_db):
    """Jalankan stage transform+load: stg_pesanan JOIN stg_pelanggan -> mart."""
    koneksi = sqlite3.connect(berkas_db)
    try:
        koneksi.execute(
            "CREATE TABLE IF NOT EXISTS mart_penjualan_kota ("
            "tanggal TEXT NOT NULL, kota TEXT NOT NULL, "
            "jumlah_pesanan INTEGER NOT NULL, total_penjualan INTEGER NOT NULL, "
            "PRIMARY KEY (tanggal, kota))"
        )
        koneksi.execute("DELETE FROM mart_penjualan_kota")
        koneksi.execute(KUERI_TRANSFORM)
        koneksi.commit()
    finally:
        koneksi.close()

Perhatikan dua sisi kontrak yang berpotensi patah. Kueri KUERI_TRANSFORM menggantungkan diri pada nama kolom waktu, pelanggan_id, dan total di stage tables — satu saja berganti nama, seluruh stage gagal. Sementara itu JOIN berperilaku tegas: pesanan yang pelanggannya tidak terdaftar di stg_pelanggan hilang dari mart tanpa pesan galat, sehingga perilaku tersebut wajib ditegaskan sebagai kontrak di test.

Sambungan Kedua: API Eksternal yang Ditiru #

Objek uji kedua adalah klien API kurs internal. Fungsinya sengaja ramping: satu panggilan requests.get, pemeriksaan status, lalu pembacaan field JSON.

python
"""Klien API kurs internal TokoKita: sumber eksternal yang harus dimock."""
import requests


def ambil_kurs_usd(url: str) -> float:
    """Ambil kurs USD->IDR dari endpoint JSON; angkat galat HTTP bila gagal."""
    respons = requests.get(url, timeout=5)
    respons.raise_for_status()
    data = respons.json()
    return float(data["usd_idr"])

Meniru API dengan mock bukan berarti menguji layanan kurs itu sendiri — itu wilayah tim pemilik API. Yang kita uji adalah sisi kita di batas tersebut: apakah URL dan timeout dikirim benar, apakah field JSON terurai dengan tipe yang tepat, dan apakah galat HTTP diteruskan alih-alih ditelan.

Test Suite Integration #

Satu berkas test menampung kedua sambungan. Fixture db_stage memakai tmp_path bawaan pytest agar setiap test memperoleh direktori kerja unik yang dibersihkan otomatis; strategi fixture yang lebih kaya dibahas di chapter Test Fixtures dan Sample Data.

python
"""Integration test: kontrak stage->mart di SQLite dan API kurs yang dimock."""
import sqlite3
from unittest import mock

import pytest
import requests

from muat_mart import bangun_mart
from sumber_kurs import ambil_kurs_usd

SKEMA_STAGE = """
CREATE TABLE stg_pesanan (pesanan_id INTEGER PRIMARY KEY, waktu TEXT NOT NULL,
                          pelanggan_id INTEGER NOT NULL, total INTEGER NOT NULL);
CREATE TABLE stg_pelanggan (pelanggan_id INTEGER PRIMARY KEY, nama TEXT NOT NULL,
                            kota TEXT NOT NULL);
"""

PESANAN = [
    (1, "2026-03-14T09:10:00", 101, 460000), (2, "2026-03-14T11:30:00", 102, 850000),
    (3, "2026-03-14T15:45:00", 101, 310000), (4, "2026-03-15T08:05:00", 103, 240000),
    (5, "2026-03-15T13:15:00", 102, 230000),
]

PELANGGAN = [(101, "Budi", "Jakarta"), (102, "Siti", "Bandung")]


@pytest.fixture
def db_stage(tmp_path):
    """Siapkan database berisi stage tables terisi; path dikembalikan ke test."""
    berkas_db = tmp_path / "tokokita.db"
    koneksi = sqlite3.connect(berkas_db)
    koneksi.executescript(SKEMA_STAGE)
    koneksi.executemany("INSERT INTO stg_pesanan VALUES (?, ?, ?, ?)", PESANAN)
    koneksi.executemany("INSERT INTO stg_pelanggan VALUES (?, ?, ?)", PELANGGAN)
    koneksi.commit()
    koneksi.close()
    return berkas_db


def test_mart_memenuhi_kontrak_baris(db_stage):
    bangun_mart(db_stage)
    koneksi = sqlite3.connect(db_stage)
    baris = koneksi.execute(
        "SELECT tanggal, kota, jumlah_pesanan, total_penjualan "
        "FROM mart_penjualan_kota ORDER BY tanggal, kota"
    ).fetchall()
    koneksi.close()
    assert baris == [
        ("2026-03-14", "Bandung", 1, 850000),
        ("2026-03-14", "Jakarta", 2, 770000),
        ("2026-03-15", "Bandung", 1, 230000),
    ]


def test_pesanan_pelanggan_tak_terdaftar_tidak_masuk_mart(db_stage):
    bangun_mart(db_stage)
    koneksi = sqlite3.connect(db_stage)
    (jumlah, total_pesanan) = koneksi.execute(
        "SELECT COUNT(*), SUM(jumlah_pesanan) FROM mart_penjualan_kota"
    ).fetchone()
    koneksi.close()
    assert (jumlah, total_pesanan) == (3, 4)  # pesanan pelanggan 103 hilang di JOIN


def test_ambil_kurs_usd_sukses():
    respons_palsu = mock.Mock()
    respons_palsu.json.return_value = {"usd_idr": "15850.25"}
    with mock.patch("sumber_kurs.requests.get", return_value=respons_palsu) as get_palsu:
        kurs = ambil_kurs_usd("https://api.example.com/kurs")
    assert kurs == 15850.25
    get_palsu.assert_called_once_with("https://api.example.com/kurs", timeout=5)


def test_ambil_kurs_usd_gagal_http_diteruskan():
    respons_palsu = mock.Mock()
    respons_palsu.json.return_value = {}
    respons_palsu.raise_for_status.side_effect = requests.HTTPError("500 Server Error")
    with mock.patch("sumber_kurs.requests.get", return_value=respons_palsu):
        with pytest.raises(requests.HTTPError):
            ambil_kurs_usd("https://api.example.com/kurs")

Jalankan python3 -m pytest -v --no-header test_integrasi.py dari direktori yang sama dengan kedua modul di atas.

Output yang Dihasilkan:

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

test_integrasi.py::test_mart_memenuhi_kontrak_baris PASSED               [ 25%]
test_integrasi.py::test_pesanan_pelanggan_tak_terdaftar_tidak_masuk_mart PASSED [ 50%]
test_integrasi.py::test_ambil_kurs_usd_sukses PASSED                     [ 75%]
test_integrasi.py::test_ambil_kurs_usd_gagal_http_diteruskan PASSED     [100%]

============================== 4 passed in 0.47s ==============================

Mari kita telaah baris-baris kode di atas:

Baris 27-37 (fixture db_stage): tmp_path menyiapkan direktori unik per test; fixture membuat database SQLite sungguhan berisi dua stage tables terisi — nyata seperti produksi, kecil ukurannya, dan dibersihkan otomatis setelah test selesai.
Baris 40-52 (test_mart_memenuhi_kontrak_baris): seluruh isi mart dibaca terurut lalu dibandingkan dengan daftar tiga baris yang dihitung manual. Perbandingan persis memeriksa nama kolom, urutan baris, dan setiap nilai agregat sekaligus.
Baris 55-62 (test_pesanan_pelanggan_tak_terdaftar): pesanan 4 milik pelanggan 103 yang tidak terdaftar, sehingga JOIN membuang pesanan tersebut — lima pesanan menjadi empat pesanan dalam tiga baris agregat. Tanpa test ini, hilangnya baris berjalan tanpa ada yang menyadari.
Baris 65-71 (test_ambil_kurs_usd_sukses): mock.Mock() meniru objek respons dan json.return_value menentukan isi JSON-nya; assert_called_once_with memverifikasi kontrak pemanggilan — URL benar dan timeout=5 ikut terkirim. Nilai "15850.25" sengaja berupa string untuk membuktikan konversi float() bekerja.
Baris 74-80 (test_ambil_kurs_usd_gagal_http_diteruskan): raise_for_status.side_effect memerintahkan mock mengangkat HTTPError; test lulus justru karena exception diteruskan ke pemanggil — pipeline tidak boleh menelan galat HTTP diam-diam.

Sambungan Mana yang Layak Diuji #

Sambungan	Kontrak yang diperiksa	Contoh assertion pada suite ini
Stage tables → transform SQL	nama kolom yang dirujuk kueri tetap ada	`JOIN` berjalan tanpa galat `no such column`
Transform SQL → tabel mart	baris agregat sesuai hitungan manual	tiga baris mart persis seperti daftar empat kolom
Referensi antar stage	baris tanpa pasangan referensi tidak lolos	`SUM(jumlah_pesanan)` bernilai 4, bukan 5
Pipeline → API eksternal	parameter panggilan benar, galat diteruskan	`assert_called_once_with(..., timeout=5)`

Integration test menutup celah sambungan satu per satu dengan biaya rendah. Namun pembaca laporan tidak melihat tabel per sambungan — yang mereka lihat adalah hasil akhir alur utuh dari berkas mentah sampai mart. Chapter End-to-End Testing menguji alur penuh tersebut secara menyeluruh.