Contract Testing

Chapter Data Contracts menuliskan janji produsen data sebagai berkas YAML. Janji itu juga berlaku pada aras layanan: kasir TokoKita (provider) menyajikan JSON bagi pipeline analytics, dan perubahan bentuk respons tanpa koordinasi menjatuhkan pipeline di hilir. Contract testing menguji janji itu dari sisi konsumen: konsumen menulis ekspektasi sebagai berkas, lalu dibuktikan otomatis terhadap respons provider — perubahan diam-diam berubah status dari insiden pukul tiga pagi menjadi kegagalan test.

Ekspektasi yang Ditulis Konsumen #

Dalam pola consumer-driven, arah kepemilikan dibalik dari kebiasaan tim: kontrak ditulis dan dimiliki konsumen, bukan penyedia layanan. Tim analytics mendefinisikan bentuk respons yang pipeline-nya butuhkan, menyimpan berkasnya di repositori konsumen, dan menjalankan pemeriksaannya di CI konsumen. Provider tinggal membuktikan; mengubah bentuk respons berarti meyakinkan kontrak lebih dahulu.

Aspek	Data Contracts	Contract test (chapter ini)
Objek yang diuji	isi tabel: schema, kualitas, freshness	bentuk respons layanan
Pemilik berkas	produsen dan konsumen bersepakat	konsumen menulis sendiri

Di ekosistem luas, Pact adalah alat contract testing yang paling dikenal. Di sini kita menulis pemeriksa kecil sendiri dengan pytest dan YAML supaya mekanismenya terlihat utuh.

Provider dan Kontrak sebagai Berkas #

Objek uji kita adalah satu endpoint internal TokoKita. Provider sengaja dibuat fungsi biasa — contract test tidak peduli transportnya; yang diuji adalah bentuk responsnya.

python
"""Provider API internal TokoKita: menyajikan pesanan sebagai JSON."""

def ambil_pesanan(pesanan_id):
    """Endpoint GET /pesanan/{id} versi 1."""
    if pesanan_id != 101:
        return None
    return {
        "order_id": "ORD-101", "pelanggan": "Budi", "total": 550000, "status": "dibayar",
        "item": [{"produk": "Kemeja Batik", "qty": 1, "harga": 240000},
                 {"produk": "Rok Tenun", "qty": 1, "harga": 310000}],
    }

Di sisi lain, tim analytics menulis kontrak sebagai berkas YAML. Setiap field respons mendapat klausul: tipe, wajib, pola, minimum, daftar nilai sah, dan jumlah item.

yaml
nama: pesanan-api
versi: 2
pemilik: tim-analytics
contoh_permintaan:
  pesanan_id: 101
respons:
  order_id: {tipe: string, wajib: true, pola: "^ORD-[0-9]+$"}
  pelanggan: {tipe: string, wajib: true}
  total: {tipe: integer, wajib: true, min: 0}
  status: {tipe: string, wajib: true, nilai_diizinkan: [baru, dibayar, kirim, selesai]}
  item: {tipe: array, wajib: true, min_item: 1}

Contract Test dengan pytest #

Test berikut memuat kontrak, memanggil provider dengan contoh permintaan dari kontrak, lalu membandingkan respons dengan seluruh klausul. Pemeriksanya ditulis tangan di file yang sama — inti contract testing ada pada disiplin ekspektasi, bukan pustakanya.

python
"""Contract test konsumen: klausul YAML dibuktikan terhadap respons provider."""
import re

import yaml

from provider import ambil_pesanan

TIPE_PYTHON = {"string": str, "integer": int, "array": list}


def periksa_respons(respons: dict, kontrak: dict) -> list[str]:
    """Bandingkan satu respons dengan klausul 'respons' pada kontrak."""
    galat = []
    for nama, klausul in kontrak["respons"].items():
        if nama not in respons:
            if klausul.get("wajib"):
                galat.append(f"[gagal] '{nama}' wajib ada, tidak ditemukan")
            continue
        nilai = respons[nama]
        if not isinstance(nilai, TIPE_PYTHON[klausul["tipe"]]):
            galat.append(f"[gagal] '{nama}': tipe {klausul['tipe']} diharapkan, "
                          f"diterima {type(nilai).__name__}")
            continue
        if "pola" in klausul and not re.match(klausul["pola"], nilai):
            galat.append(f"[gagal] '{nama}': tidak cocok pola {klausul['pola']}")
        if "min" in klausul and nilai < klausul["min"]:
            galat.append(f"[gagal] '{nama}': di bawah minimum {klausul['min']}")
        if "nilai_diizinkan" in klausul and nilai not in klausul["nilai_diizinkan"]:
            galat.append(f"[gagal] '{nama}': nilai '{nilai}' di luar daftar diizinkan")
        if "min_item" in klausul and len(nilai) < klausul["min_item"]:
            galat.append(f"[gagal] '{nama}': minimal {klausul['min_item']} item")
    return galat


def test_respons_memenuhi_kontrak():
    with open("kontrak_pesanan.yaml", encoding="utf-8") as f:
        kontrak = yaml.safe_load(f)
    respons = ambil_pesanan(kontrak["contoh_permintaan"]["pesanan_id"])
    assert respons is not None, "provider mengembalikan None untuk contoh permintaan"
    galat = periksa_respons(respons, kontrak)
    assert galat == [], "\n".join(galat)

Mari kita telaah baris-baris kode di atas:

Baris 8 (TIPE_PYTHON): pemetaan nama tipe kontrak ke tipe Python — satu sumber kebenaran untuk seluruh pemeriksaan tipe.
Baris 11-32 (periksa_respons): klausul diuji berurutan — kehadiran field, tipe, lalu aturan nilai — dan semua pelanggaran dikumpulkan agar satu laporan memuat semuanya.
Baris 35-41 (badan test): kontrak dimuat dengan yaml.safe_load seperti pada chapter Data Contracts; contoh permintaan diambil dari kontrak sehingga provider dan konsumen menguji kasus yang sama.

bash

python3 -m pytest test_kontrak.py -q

Output yang Dihasilkan:

text

.                                                                        [100%]
1 passed in 0.20s

Sekarang tim kasir merilis provider v2 tanpa memberi tahu: kunci total diganti total_tagihan dan status kini PAID — seperti insiden chapter Investigasi Kegagalan Data. Kontrak tidak berubah, jadi perintah yang sama kini berteriak; durasi 0.20s berubah-ubah antar eksekusi, sedangkan jumlah passed/failed stabil — persis seperti pada chapter Pengujian Kode Python.

Output yang Dihasilkan:

text
F                                                                        [100%]
=================================== FAILURES ===================================
________________________ test_respons_memenuhi_kontrak _________________________

    def test_respons_memenuhi_kontrak():
        with open("kontrak_pesanan.yaml", encoding="utf-8") as f:
            kontrak = yaml.safe_load(f)
        respons = ambil_pesanan(kontrak["contoh_permintaan"]["pesanan_id"])
        assert respons is not None, "provider mengembalikan None untuk contoh permintaan"
        galat = periksa_respons(respons, kontrak)
>       assert galat == [], "\n".join(galat)
E       AssertionError: [gagal] 'total' wajib ada, tidak ditemukan
E         [gagal] 'status': nilai 'PAID' di luar daftar diizinkan

Pemeriksaan Kompatibilitas Schema #

Contract test menguji satu respons contoh. Untuk schema — janji bagi seluruh keluarga data — pemeriksaannya adalah kompatibilitas: schema baru boleh naik asalkan pembaca lama tidak rusak. Aturan ini ditegakkan Schema Registry pada topik Kafka dan dibedah di chapter Schema Evolution.

python
"""Pemeriksa kompatibilitas schema: perubahan aditif boleh, merusak ditolak."""
LAMA = {
    "order_id": {"tipe": "string"},
    "pelanggan": {"tipe": "string"},
    "total": {"tipe": "integer"},
    "status": {"tipe": "string", "nilai_diizinkan": ["baru", "dibayar", "kirim", "selesai"]},
}
ADITIF = dict(LAMA, diskon={"tipe": "integer", "default": 0})
MERUSAK = {
    "order_id": {"tipe": "string"},
    "total": {"tipe": "string"},
    "status": {"tipe": "string", "nilai_diizinkan": ["PAID", "UNPAID"]},
    "diskon": {"tipe": "integer"},
}

def cek_kompatibilitas(lama: dict, baru: dict) -> list[str]:
    """Kembalikan perubahan yang merusak pembaca yang masih memakai schema lama."""
    galat = []
    for nama, aturan in lama.items():
        if nama not in baru:
            galat.append(f"[rusak] field '{nama}' dihapus")
        elif baru[nama]["tipe"] != aturan["tipe"]:
            galat.append(f"[rusak] '{nama}': tipe {aturan['tipe']} menjadi {baru[nama]['tipe']}")
        elif "nilai_diizinkan" in aturan:
            hilang = set(aturan["nilai_diizinkan"]) - set(baru[nama].get("nilai_diizinkan", []))
            if hilang:
                galat.append(f"[rusak] '{nama}': nilai sah lama hilang: {sorted(hilang)}")
    for nama, aturan in baru.items():
        if nama not in lama and "default" not in aturan:
            galat.append(f"[rusak] '{nama}' baru tanpa default")
    return galat

for label, kandidat in [("ADITIF (tambah 'diskon' ber-default)", ADITIF),
                        ("MERUSAK (ubah tipe, ganti daftar, tanpa default)", MERUSAK)]:
    galat = cek_kompatibilitas(LAMA, kandidat)
    print(f"schema lama -> {label}:")
    print("\n".join(galat) if galat else "  KOMPATIBEL: pembaca schema lama tidak terganggu")

Output yang Dihasilkan:

text
schema lama -> ADITIF (tambah 'diskon' ber-default):
  KOMPATIBEL: pembaca schema lama tidak terganggu
schema lama -> MERUSAK (ubah tipe, ganti daftar, tanpa default):
[rusak] field 'pelanggan' dihapus
[rusak] 'total': tipe integer menjadi string
[rusak] 'status': nilai sah lama hilang: ['baru', 'dibayar', 'kirim', 'selesai']
[rusak] 'diskon' baru tanpa default

Penambahan field ber-default lolos karena pembaca lama cukup mengisi default; empat perubahan lain ditolak karena meninggalkan pembaca lama tanpa jalan keluar. Rujukan resmi: dokumentasi pytest . Berikutnya: data uji yang murah, deterministik, dan dekat kenyataan — pokok bahasan chapter Test Fixtures dan Sample Data.