Seluruh fitur aplikasi sudah berjalan, tetapi kita baru mengujinya dengan mengetik di menu — cara yang lambat, tidak bisa diulang persis, dan cenderung berhenti setelah sekali coba. Sebelum menulis test, ada satu halangan desain: setiap function di database.py membuka koneksinya sendiri, sehingga test sulit mengarahkan operasi ke database miliknya. Pada bagian penutup ini kita me-refactor lapisan data agar koneksi menjadi parameter, memindahkan seluruh query dari Bagian 5 ke tempatnya, lalu membuktikan semuanya lewat 18 test pytest.
Masalah dengan Koneksi Tersembunyi #
Cara paling alami untuk menguji lapisan data adalah database :memory: — database yang hidup hanya di RAM dan musnah saat koneksi ditutup. Tetapi perhatikan jebakan berikut: pada SQLite, setiap koneksi baru ke :memory: adalah database yang berbeda. Jika tambah_produk() membuka koneksinya sendiri dan test membuka koneksi lain untuk memeriksa hasil, keduanya berbicara tentang dua database yang tidak saling kenal.
Solusinya adalah membalik arah ketergantungan: alih-alih function membuka koneksi sendiri, pemanggil menyuntikkan koneksi sebagai parameter. Aplikasi menyuntikkan satu koneksi inventaris.db yang sama sepanjang sesi; test menyuntikkan koneksi :memory: miliknya. Bonusnya, query laporan yang sempat menumpuk di main.py pada Bagian 5 ikut pindah ke rumahnya di database.py.
database.py Versi Final #
Berikut isi lengkap database.py setelah refactor. Function koneksi() dan inisialisasi() tidak berubah; seluruh function operasi data kini menerima conn sebagai parameter pertama dan tidak lagi memanggil koneksi() di dalamnya:
"""Lapisan akses data untuk aplikasi inventaris.
Seluruh function operasi data menerima objek koneksi sebagai
parameter pertama. Aplikasi memakai satu koneksi selama program
berjalan, sedangkan test memakai database :memory: miliknya sendiri.
"""
import sqlite3
from pathlib import Path
NAMA_DATABASE = "inventaris.db"
BERKAS_SKEMA = Path(__file__).with_name("schema.sql")
BATAS_STOK_MENIPIS = 5
JUMLAH_TOP_PRODUK = 5
def koneksi(db_path: str = NAMA_DATABASE) -> sqlite3.Connection:
"""Buka koneksi baru ke database inventaris."""
conn = sqlite3.connect(db_path)
conn.execute("PRAGMA foreign_keys = ON")
conn.row_factory = sqlite3.Row
return conn
def inisialisasi(db_path: str = NAMA_DATABASE) -> None:
"""Terapkan schema.sql ke database. Aman dijalankan berulang."""
conn = koneksi(db_path)
try:
conn.executescript(BERKAS_SKEMA.read_text(encoding="utf-8"))
conn.commit()
finally:
conn.close()
def validasi_produk(nama: str, kategori: str, harga: int, stok: int) -> None:
"""Periksa masukan produk; angkat ValueError bila tidak sah."""
if not nama.strip():
raise ValueError("Nama produk tidak boleh kosong.")
if not kategori.strip():
raise ValueError("Kategori tidak boleh kosong.")
if isinstance(harga, bool) or not isinstance(harga, int):
raise ValueError("Harga harus berupa bilangan bulat.")
if harga < 0:
raise ValueError("Harga tidak boleh negatif.")
if isinstance(stok, bool) or not isinstance(stok, int):
raise ValueError("Stok harus berupa bilangan bulat.")
if stok < 0:
raise ValueError("Stok tidak boleh negatif.")
def tambah_produk(conn: sqlite3.Connection, nama: str, kategori: str,
harga: int, stok: int) -> int:
"""Simpan produk baru, kembalikan id-nya."""
sql = "INSERT INTO produk (nama, kategori, harga, stok) VALUES (?, ?, ?, ?)"
kursor = conn.execute(sql, (nama, kategori, harga, stok))
conn.commit()
return int(kursor.lastrowid)
def lihat_produk(conn: sqlite3.Connection) -> list[sqlite3.Row]:
"""Ambil seluruh produk, urut berdasarkan nama."""
return conn.execute("SELECT * FROM produk ORDER BY nama").fetchall()
def ambil_produk(conn: sqlite3.Connection,
produk_id: int) -> sqlite3.Row | None:
"""Ambil satu produk berdasarkan id; None bila tidak ada."""
kursor = conn.execute("SELECT * FROM produk WHERE id = ?", (produk_id,))
return kursor.fetchone()
def perbarui_harga(conn: sqlite3.Connection, produk_id: int,
harga_baru: int) -> bool:
"""Set harga produk menjadi nilai baru. False bila id tidak ada."""
kursor = conn.execute(
"UPDATE produk SET harga = ? WHERE id = ?", (harga_baru, produk_id)
)
conn.commit()
return kursor.rowcount > 0
def sesuaikan_stok(conn: sqlite3.Connection, produk_id: int, perubahan: int,
keterangan: str) -> None:
"""Ubah stok sebesar perubahan (+/-) dan catat ke riwayat secara atomik."""
with conn: # commit bila sukses, rollback bila ada galat
kursor = conn.execute(
"UPDATE produk SET stok = stok + ? WHERE id = ?",
(perubahan, produk_id),
)
if kursor.rowcount == 0:
raise ValueError(f"Produk dengan id {produk_id} tidak ditemukan.")
conn.execute(
"INSERT INTO riwayat (produk_id, perubahan, keterangan)"
" VALUES (?, ?, ?)",
(produk_id, perubahan, keterangan),
)
def lihat_riwayat(conn: sqlite3.Connection,
produk_id: int) -> list[sqlite3.Row]:
"""Ambil riwayat penyesuaian stok sebuah produk, terbaru lebih dulu."""
return conn.execute(
"SELECT * FROM riwayat WHERE produk_id = ? ORDER BY id DESC",
(produk_id,),
).fetchall()
def hapus_produk(conn: sqlite3.Connection, produk_id: int) -> bool:
"""Hapus produk beserta riwayatnya. False bila id tidak ada."""
kursor = conn.execute("DELETE FROM produk WHERE id = ?", (produk_id,))
conn.commit()
return kursor.rowcount > 0
def escape_like(keyword: str) -> str:
"""Amankan karakter wildcard LIKE (persen dan garis bawah)."""
return (
keyword.replace("\\", "\\\\")
.replace("%", "\\%")
.replace("_", "\\_")
)
def cari_produk(conn: sqlite3.Connection, keyword: str) -> list[sqlite3.Row]:
"""Cari produk yang namanya mengandung keyword (LIKE yang di-escape)."""
pola = f"%{escape_like(keyword)}%"
sql = "SELECT * FROM produk WHERE nama LIKE ? ESCAPE '\\' ORDER BY nama"
return conn.execute(sql, (pola,)).fetchall()
def filter_kategori(conn: sqlite3.Connection,
kategori: str) -> list[sqlite3.Row]:
"""Ambil seluruh produk pada satu kategori."""
sql = "SELECT * FROM produk WHERE kategori = ? ORDER BY nama"
return conn.execute(sql, (kategori,)).fetchall()
def laporan_stok_menipis(conn: sqlite3.Connection,
batas: int = BATAS_STOK_MENIPIS) -> list[sqlite3.Row]:
"""Ambil produk dengan stok di bawah batas, stok terkecil lebih dulu."""
sql = "SELECT * FROM produk WHERE stok < ? ORDER BY stok, nama"
return conn.execute(sql, (batas,)).fetchall()
def nilai_inventaris_per_kategori(
conn: sqlite3.Connection,
) -> list[sqlite3.Row]:
"""Hitung jumlah produk dan total nilai stok per kategori."""
sql = """
SELECT kategori,
COUNT(*) AS jumlah_produk,
SUM(harga * stok) AS nilai_total
FROM produk
GROUP BY kategori
ORDER BY nilai_total DESC
"""
return conn.execute(sql).fetchall()
def top_produk(conn: sqlite3.Connection,
jumlah: int = JUMLAH_TOP_PRODUK) -> list[sqlite3.Row]:
"""Ambil produk dengan nilai stok terbesar."""
sql = """
SELECT id, nama, kategori, harga, stok,
harga * stok AS nilai
FROM produk
ORDER BY nilai DESC, nama
LIMIT ?
"""
return conn.execute(sql, (jumlah,)).fetchall()Perhatikan dua hal pada versi final ini. Pertama, sesuaikan_stok() masih memakai with conn: dari Bagian 4 — kini blok tersebut mengelola transaksi pada koneksi yang disuntikkan, dan pemanggil-lah yang menentukan kapan koneksi ditutup. Kedua, function yang menulis data (tambah_produk, perbarui_harga, hapus_produk) tetap memanggil conn.commit() sendiri; pemanggil test maupun aplikasi tinggal memanggil tanpa memikirkan detail transaksi.
main.py dengan Satu Koneksi #
Perubahan di main.py bersifat mekanis namun menyeluruh: main() membuka satu koneksi di awal, setiap handler menerima conn sebagai parameter, dan blok finally menjamin koneksi tertutup. Isi lengkapnya:
"""Aplikasi Inventaris SQLite - antarmuka pengguna."""
import sqlite3
import database
def format_rupiah(nilai: int) -> str:
"""Format angka menjadi teks rupiah, contoh 15000 menjadi Rp15.000."""
return "Rp" + f"{nilai:,}".replace(",", ".")
def cetak_tabel(rows: list) -> None:
"""Cetak daftar produk sebagai tabel rapi."""
print(f"{'ID':<3} {'Nama':<18} {'Kategori':<10} {'Harga':>10} {'Stok':>4}")
print(f"{'---':<3} {'-' * 18} {'-' * 10} {'-' * 10} {'-' * 4}")
for row in rows:
print(
f"{row['id']:<3} {row['nama']:<18} {row['kategori']:<10} "
f"{format_rupiah(row['harga']):>10} {row['stok']:>4}"
)
def baca_angka(prompt: str) -> int:
"""Minta masukan angka bulat, ulangi sampai masukan sah."""
while True:
teks = input(prompt).strip()
try:
return int(teks)
except ValueError:
print("Masukkan angka bulat, misalnya 15000.")
def menu_tambah(conn: sqlite3.Connection) -> None:
nama = input("Nama produk: ").strip()
kategori = input("Kategori: ").strip()
harga = baca_angka("Harga (Rp): ")
stok = baca_angka("Stok awal: ")
try:
database.validasi_produk(nama, kategori, harga, stok)
except ValueError as galat:
print(f"Gagal: {galat}")
return
try:
id_baru = database.tambah_produk(conn, nama, kategori, harga, stok)
except sqlite3.IntegrityError:
print(f"Produk dengan nama '{nama}' sudah terdaftar.")
return
print(f"Produk '{nama}' tersimpan dengan id {id_baru}.")
def menu_lihat(conn: sqlite3.Connection) -> None:
rows = database.lihat_produk(conn)
if not rows:
print("Belum ada produk.")
return
cetak_tabel(rows)
def menu_cari(conn: sqlite3.Connection) -> None:
keyword = input("Kata kunci nama: ").strip()
if not keyword:
print("Kata kunci tidak boleh kosong.")
return
hasil = database.cari_produk(conn, keyword)
if not hasil:
print(f"Tidak ada produk yang mengandung '{keyword}'.")
return
print(f"Hasil pencarian untuk '{keyword}':")
cetak_tabel(hasil)
def menu_filter(conn: sqlite3.Connection) -> None:
kategori = input("Kategori: ").strip()
if not kategori:
print("Kategori tidak boleh kosong.")
return
hasil = database.filter_kategori(conn, kategori)
if not hasil:
print(f"Tidak ada produk pada kategori '{kategori}'.")
return
print(f"Produk pada kategori '{kategori}':")
cetak_tabel(hasil)
def menu_ubah_harga(conn: sqlite3.Connection) -> None:
produk_id = baca_angka("ID produk: ")
harga_baru = baca_angka("Harga baru (Rp): ")
if database.perbarui_harga(conn, produk_id, harga_baru):
print("Harga berhasil diperbarui.")
else:
print(f"Tidak ada produk dengan id {produk_id}.")
def menu_sesuaikan_stok(conn: sqlite3.Connection) -> None:
produk_id = baca_angka("ID produk: ")
perubahan = baca_angka("Perubahan stok (boleh negatif): ")
keterangan = input("Keterangan: ").strip() or "penyesuaian manual"
try:
database.sesuaikan_stok(conn, produk_id, perubahan, keterangan)
except ValueError as galat:
print(f"Gagal: {galat}")
return
except sqlite3.IntegrityError:
print("Gagal: penyesuaian membuat stok negatif. Transaksi dibatalkan.")
return
print("Stok berhasil disesuaikan.")
def menu_hapus(conn: sqlite3.Connection) -> None:
produk_id = baca_angka("ID produk: ")
if database.hapus_produk(conn, produk_id):
print("Produk berhasil dihapus.")
else:
print(f"Tidak ada produk dengan id {produk_id}.")
def menu_riwayat(conn: sqlite3.Connection) -> None:
produk_id = baca_angka("ID produk: ")
produk = database.ambil_produk(conn, produk_id)
if produk is None:
print(f"Tidak ada produk dengan id {produk_id}.")
return
rows = database.lihat_riwayat(conn, produk_id)
if not rows:
print(f"Belum ada riwayat untuk '{produk['nama']}'.")
return
print(f"Riwayat untuk '{produk['nama']}':")
print(f"{'Waktu':<21} {'Perubahan':>9} Keterangan")
for row in rows:
print(f"{row['waktu']:<21} {row['perubahan']:>+9} {row['keterangan']}")
def menu_stok_menipis(conn: sqlite3.Connection) -> None:
hasil = database.laporan_stok_menipis(conn)
if not hasil:
print(f"Tidak ada produk dengan stok di bawah {database.BATAS_STOK_MENIPIS}.")
return
print(f"Produk dengan stok di bawah {database.BATAS_STOK_MENIPIS}:")
print(f"{'ID':<3} {'Nama':<18} {'Kategori':<10} {'Stok':>4}")
print(f"{'---':<3} {'-' * 18} {'-' * 10} {'-' * 4}")
for row in hasil:
print(f"{row['id']:<3} {row['nama']:<18} {row['kategori']:<10} {row['stok']:>4}")
def menu_nilai_inventaris(conn: sqlite3.Connection) -> None:
hasil = database.nilai_inventaris_per_kategori(conn)
if not hasil:
print("Belum ada produk untuk dihitung.")
return
print(f"{'Kategori':<12} {'Jumlah':>7} {'Nilai Total':>13}")
print(f"{'-' * 12} {'-' * 7} {'-' * 13}")
total_nilai = 0
total_produk = 0
for row in hasil:
print(
f"{row['kategori']:<12} {row['jumlah_produk']:>7} "
f"{format_rupiah(row['nilai_total']):>13}"
)
total_nilai += row["nilai_total"]
total_produk += row["jumlah_produk"]
print(f"{'-' * 12} {'-' * 7} {'-' * 13}")
print(f"{'TOTAL':<12} {total_produk:>7} {format_rupiah(total_nilai):>13}")
def menu_top_produk(conn: sqlite3.Connection) -> None:
hasil = database.top_produk(conn)
if not hasil:
print("Belum ada produk untuk dihitung.")
return
print(f"Top {database.JUMLAH_TOP_PRODUK} produk berdasarkan nilai stok:")
print(f"{'#':<3} {'Nama':<18} {'Harga':>10} {'Stok':>4} {'Nilai':>12}")
print(f"{'---':<3} {'-' * 18} {'-' * 10} {'-' * 4} {'-' * 12}")
for peringkat, row in enumerate(hasil, start=1):
print(
f"{peringkat:<3} {row['nama']:<18} "
f"{format_rupiah(row['harga']):>10} {row['stok']:>4} "
f"{format_rupiah(row['nilai']):>12}"
)
def tampilkan_menu() -> None:
print()
print("=== Aplikasi Inventaris ===")
print("1. Tambah produk")
print("2. Lihat semua produk")
print("3. Ubah harga")
print("4. Sesuaikan stok (+/-)")
print("5. Hapus produk")
print("6. Riwayat produk")
print("7. Cari produk")
print("8. Filter per kategori")
print("9. Laporan stok menipis")
print("10. Nilai inventaris per kategori")
print("11. Top produk")
print("0. Keluar")
def tangani_pilihan(conn: sqlite3.Connection, pilihan: str) -> None:
match pilihan:
case "1":
menu_tambah(conn)
case "2":
menu_lihat(conn)
case "3":
menu_ubah_harga(conn)
case "4":
menu_sesuaikan_stok(conn)
case "5":
menu_hapus(conn)
case "6":
menu_riwayat(conn)
case "7":
menu_cari(conn)
case "8":
menu_filter(conn)
case "9":
menu_stok_menipis(conn)
case "10":
menu_nilai_inventaris(conn)
case "11":
menu_top_produk(conn)
case _:
print("Pilihan tidak dikenal.")
def main() -> None:
database.inisialisasi()
conn = database.koneksi()
try:
print("Database siap. Selamat datang!")
while True:
tampilkan_menu()
try:
pilihan = input("Pilih menu: ").strip()
except EOFError:
print()
break
if pilihan == "0":
print("Sampai jumpa!")
break
tangani_pilihan(conn, pilihan)
finally:
conn.close()
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("\nProgram dihentikan.")Tambahan finalisasi yang layak dibedah: try ... finally: conn.close() memastikan koneksi ditutup bagaimanapun program berhenti, dan except KeyboardInterrupt di titik masuk membuat penekanan Ctrl+C berakhir dengan pesan singkat alih-alih traceback.
Menulis Test dengan pytest #
Test hanya berbicara dengan database.py — menu dan input() tidak ikut diuji di sini. Fixture berikut menyiapkan database :memory: lengkap dengan skema dan data awal untuk setiap test:
"""Uji otomatis untuk lapisan data aplikasi inventaris.
Jalankan dari folder proyek:
python3 -m pytest test_database.py -v
"""
import sqlite3
from pathlib import Path
import pytest
import database
BERKAS_SKEMA = Path(database.__file__).with_name("schema.sql")
@pytest.fixture
def conn() -> sqlite3.Connection:
"""Database SQLite :memory: dengan skema dan data awal."""
koneksi_uji = sqlite3.connect(":memory:")
koneksi_uji.execute("PRAGMA foreign_keys = ON")
koneksi_uji.row_factory = sqlite3.Row
koneksi_uji.executescript(BERKAS_SKEMA.read_text(encoding="utf-8"))
yield koneksi_uji
koneksi_uji.close()
def test_inisialisasi_idempoten(tmp_path):
"""inisialisasi dua kali tidak menduplikasi tabel maupun data."""
berkas_db = tmp_path / "uji.db"
database.inisialisasi(str(berkas_db))
database.inisialisasi(str(berkas_db))
conn = sqlite3.connect(berkas_db)
try:
tabel = {
row[0]
for row in conn.execute(
"SELECT name FROM sqlite_master WHERE type = 'table'"
)
}
assert {"produk", "riwayat"} <= tabel
jumlah = conn.execute("SELECT COUNT(*) FROM produk").fetchone()[0]
assert jumlah == 6
finally:
conn.close()
def test_tambah_lalu_ambil(conn):
id_baru = database.tambah_produk(conn, "Susu Kotak", "Minuman", 6000, 12)
produk = database.ambil_produk(conn, id_baru)
assert produk["nama"] == "Susu Kotak"
assert produk["kategori"] == "Minuman"
assert produk["harga"] == 6000
assert produk["stok"] == 12
def test_nama_duplikat_ditolak(conn):
with pytest.raises(sqlite3.IntegrityError):
database.tambah_produk(conn, "Kopi Tubruk", "Minuman", 15000, 5)
@pytest.mark.parametrize(
"nama, kategori, harga, stok",
[
("", "Minuman", 1000, 5),
(" ", "Minuman", 1000, 5),
("Teh", "", 1000, 5),
("Teh", "Minuman", -1, 5),
("Teh", "Minuman", 1000, -5),
],
)
def test_validasi_produk_menolak_masukan_salah(nama, kategori, harga, stok):
with pytest.raises(ValueError):
database.validasi_produk(nama, kategori, harga, stok)
def test_sesuaikan_stok_mencatat_riwayat(conn):
database.sesuaikan_stok(conn, 1, 5, "restock mingguan")
produk = database.ambil_produk(conn, 1)
assert produk["stok"] == 24 + 5
riwayat = database.lihat_riwayat(conn, 1)
assert len(riwayat) == 1
assert riwayat[0]["perubahan"] == 5
assert riwayat[0]["keterangan"] == "restock mingguan"
def test_sesuaikan_stok_negatif_dibatalkan(conn):
with pytest.raises(sqlite3.IntegrityError):
database.sesuaikan_stok(conn, 1, -9999, "salah input")
produk = database.ambil_produk(conn, 1)
assert produk["stok"] == 24
assert len(database.lihat_riwayat(conn, 1)) == 0
def test_sesuaikan_stok_produk_hilang(conn):
with pytest.raises(ValueError):
database.sesuaikan_stok(conn, 999, 1, "uji")
def test_hapus_produk_menghapus_riwayat(conn):
database.sesuaikan_stok(conn, 1, 5, "restock")
assert len(database.lihat_riwayat(conn, 1)) == 1
assert database.hapus_produk(conn, 1) is True
assert database.ambil_produk(conn, 1) is None
assert len(database.lihat_riwayat(conn, 1)) == 0
def test_perbarui_harga(conn):
assert database.perbarui_harga(conn, 1, 17500) is True
assert database.ambil_produk(conn, 1)["harga"] == 17500
assert database.perbarui_harga(conn, 999, 17500) is False
def test_cari_produk_dengan_karakter_like(conn):
database.tambah_produk(conn, "Coklat Batang 80%", "Makanan", 12000, 10)
hasil = database.cari_produk(conn, "%")
assert [row["nama"] for row in hasil] == ["Coklat Batang 80%"]
def test_filter_kategori(conn):
hasil = database.filter_kategori(conn, "Minuman")
assert [row["nama"] for row in hasil] == ["Kopi Tubruk", "Teh Celup"]
def test_laporan_stok_menipis(conn):
hasil = database.laporan_stok_menipis(conn)
assert [(row["nama"], row["stok"]) for row in hasil] == [
("Keripik Singkong", 2),
("Teh Celup", 3),
("Pulpen Gel", 4),
]
def test_nilai_inventaris_per_kategori(conn):
hasil = database.nilai_inventaris_per_kategori(conn)
assert [(row["kategori"], row["nilai_total"]) for row in hasil] == [
("Minuman", 384000),
("Alat Tulis", 214000),
("Makanan", 200000),
]
def test_top_produk(conn):
hasil = database.top_produk(conn, jumlah=3)
assert [(row["nama"], row["nilai"]) for row in hasil] == [
("Kopi Tubruk", 360000),
("Buku Tulis", 200000),
("Roti Tawar", 180000),
]Empat teknik pytest pada berkas tersebut yang layak diperhatikan:
- Fixture
conn: setiap test menerima database:memory:yang segar — satu koneksi yang sama dipakai untuk menyiapkan skema dan memeriksa hasil, sehingga jebakan “dua koneksi:memory:berbeda” tidak terjadi. Pemanggilanyieldmemisahkan persiapan dari pembersihan. pytest.raises(...): menegaskan bahwa exception justru diharapkan muncul, seperti yang dibahas pada materi Testing Exceptions.@pytest.mark.parametrize: lima kasus validasi salah berjalan sebagai lima test terpisah; satu kasus gagal tidak menyembunyikan empat lainnya.tmp_path: fixture bawaan pytest yang menyediakan folder sementara unik per test, dipakaitest_inisialisasi_idempotenuntuk menguji database berkas sungguhan dua kali.
Menjalankan Seluruh Test #
Pastikan test_database.py, database.py, dan schema.sql berada dalam satu folder, lalu jalankan:
python3 -m pytest test_database.py -vOutput yang Dihasilkan:
============================= test session starts ==============================
platform linux -- Python 3.12.15, pytest-9.0.2, pluggy-1.6.0 -- /home/z/.venv/bin/python3
cachedir: .pytest_cache
metadata: {'Python': '3.12.15', 'Platform': 'Linux-5.10.134-013.15.kangaroo.al8.x86_64-x86_64-with-glibc2.41', 'Packages': {'pytest': '9.0.2', 'pluggy': '1.6.0'}, 'Plugins': {'Faker': '40.1.2', 'metadata': '3.1.1', 'asyncio': '1.3.0', 'ddtrace': '4.2.2', 'cov': '7.0.0', 'json-report': '1.5.0', 'anyio': '4.13.0'}}
rootdir: /home/z
configfile: pyproject.toml
plugins: Faker-40.1.2, metadata-3.1.1, asyncio-1.3.0, ddtrace-4.2.2, cov-7.0.0, json-report-1.5.0, anyio-4.13.0
asyncio: mode=Mode.STRICT, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collecting ... collected 18 items
test_database.py::test_inisialisasi_idempoten PASSED [ 5%]
test_database.py::test_tambah_lalu_ambil PASSED [ 11%]
test_database.py::test_nama_duplikat_ditolak PASSED [ 16%]
test_database.py::test_validasi_produk_menolak_masukan_salah[-Minuman-1000-5] PASSED [ 22%]
test_database.py::test_validasi_produk_menolak_masukan_salah[ -Minuman-1000-5] PASSED [ 27%]
test_database.py::test_validasi_produk_menolak_masukan_salah[Teh--1000-5] PASSED [ 33%]
test_database.py::test_validasi_produk_menolak_masukan_salah[Teh-Minuman--1-5] PASSED [ 38%]
test_database.py::test_validasi_produk_menolak_masukan_salah[Teh-Minuman-1000--5] PASSED [ 44%]
test_database.py::test_sesuaikan_stok_mencatat_riwayat PASSED [ 50%]
test_database.py::test_sesuaikan_stok_negatif_dibatalkan PASSED [ 55%]
test_database.py::test_sesuaikan_stok_produk_hilang PASSED [ 61%]
test_database.py::test_hapus_produk_menghapus_riwayat PASSED [ 66%]
test_database.py::test_perbarui_harga PASSED [ 72%]
test_database.py::test_cari_produk_dengan_karakter_like PASSED [ 77%]
test_database.py::test_filter_kategori PASSED [ 83%]
test_database.py::test_laporan_stok_menipis PASSED [ 88%]
test_database.py::test_nilai_inventaris_per_kategori PASSED [ 94%]
test_database.py::test_top_produk PASSED [100%]
============================== 18 passed in 0.15s ==============================Test test_sesuaikan_stok_negatif_dibatalkan adalah pengunci utama proyek ini: kegagalan UPDATE membuat stok negatif, dan test membuktikan stok tidak berubah serta riwayat tetap kosong — atomikitas transaksi yang kita rancang sejak Bagian 4 kini terverifikasi otomatis.
Uji Asap Aplikasi Final #
Sebagai penutup, kita menjalankan aplikasi final sekali dari database bersih: menyesuaikan stok Teh Celup (id 2) sebanyak -1, mencari celup, lalu melihat laporan stok menipis — tiga operasi berbagi satu koneksi yang sama:
printf '4\n2\n-1\npenjualan offline\n7\ncelup\n9\n0\n' | python3 main.pyOutput yang Dihasilkan:
Database siap. Selamat datang!
=== Aplikasi Inventaris ===
1. Tambah produk
2. Lihat semua produk
3. Ubah harga
4. Sesuaikan stok (+/-)
5. Hapus produk
6. Riwayat produk
7. Cari produk
8. Filter per kategori
9. Laporan stok menipis
10. Nilai inventaris per kategori
11. Top produk
0. Keluar
Pilih menu: ID produk: Perubahan stok (boleh negatif): Keterangan: Stok berhasil disesuaikan.
=== Aplikasi Inventaris ===
1. Tambah produk
2. Lihat semua produk
3. Ubah harga
4. Sesuaikan stok (+/-)
5. Hapus produk
6. Riwayat produk
7. Cari produk
8. Filter per kategori
9. Laporan stok menipis
10. Nilai inventaris per kategori
11. Top produk
0. Keluar
Pilih menu: Kata kunci nama: Hasil pencarian untuk 'celup':
ID Nama Kategori Harga Stok
--- ------------------ ---------- ---------- ----
2 Teh Celup Minuman Rp8.000 2
=== Aplikasi Inventaris ===
1. Tambah produk
2. Lihat semua produk
3. Ubah harga
4. Sesuaikan stok (+/-)
5. Hapus produk
6. Riwayat produk
7. Cari produk
8. Filter per kategori
9. Laporan stok menipis
10. Nilai inventaris per kategori
11. Top produk
0. Keluar
Pilih menu: Produk dengan stok di bawah 5:
ID Nama Kategori Stok
--- ------------------ ---------- ----
4 Keripik Singkong Makanan 2
2 Teh Celup Minuman 2
6 Pulpen Gel Alat Tulis 4
=== Aplikasi Inventaris ===
1. Tambah produk
2. Lihat semua produk
3. Ubah harga
4. Sesuaikan stok (+/-)
5. Hapus produk
6. Riwayat produk
7. Cari produk
8. Filter per kategori
9. Laporan stok menipis
10. Nilai inventaris per kategori
11. Top produk
0. Keluar
Pilih menu: Sampai jumpa!Stok Teh Celup berubah dari 3 menjadi 2 dan laporan langsung mencerminkannya — satu koneksi, satu sumber kebenaran. Proyek aplikasi inventaris kini tuntas: skema dengan constraint yang bekerja, CRUD berparameter, transaksi atomik, pencarian yang aman, laporan agregat, dan 18 test yang menjaga semuanya. Ketiga berkas hasil akhir tersedia pada berkas sumber di bawah halaman, dan langkah lanjutan yang menanti adalah eksplorasi Anda sendiri: pagination, ekspor laporan ke CSV, atau migrasi ke PostgreSQL lewat materi Pengenalan PostgreSQL dan MySQL.
Berkas Sumber
Berkas pendukung halaman ini dapat diunduh dan dijalankan langsung sesuai penjelasan di atas.
database.py
Unduh berkas"""Lapisan akses data untuk aplikasi inventaris.
Seluruh function operasi data menerima objek koneksi sebagai
parameter pertama. Aplikasi memakai satu koneksi selama program
berjalan, sedangkan test memakai database :memory: miliknya sendiri.
"""
import sqlite3
from pathlib import Path
NAMA_DATABASE = "inventaris.db"
BERKAS_SKEMA = Path(__file__).with_name("schema.sql")
BATAS_STOK_MENIPIS = 5
JUMLAH_TOP_PRODUK = 5
def koneksi(db_path: str = NAMA_DATABASE) -> sqlite3.Connection:
"""Buka koneksi baru ke database inventaris."""
conn = sqlite3.connect(db_path)
conn.execute("PRAGMA foreign_keys = ON")
conn.row_factory = sqlite3.Row
return conn
def inisialisasi(db_path: str = NAMA_DATABASE) -> None:
"""Terapkan schema.sql ke database. Aman dijalankan berulang."""
conn = koneksi(db_path)
try:
conn.executescript(BERKAS_SKEMA.read_text(encoding="utf-8"))
conn.commit()
finally:
conn.close()
def validasi_produk(nama: str, kategori: str, harga: int, stok: int) -> None:
"""Periksa masukan produk; angkat ValueError bila tidak sah."""
if not nama.strip():
raise ValueError("Nama produk tidak boleh kosong.")
if not kategori.strip():
raise ValueError("Kategori tidak boleh kosong.")
if isinstance(harga, bool) or not isinstance(harga, int):
raise ValueError("Harga harus berupa bilangan bulat.")
if harga < 0:
raise ValueError("Harga tidak boleh negatif.")
if isinstance(stok, bool) or not isinstance(stok, int):
raise ValueError("Stok harus berupa bilangan bulat.")
if stok < 0:
raise ValueError("Stok tidak boleh negatif.")
def tambah_produk(conn: sqlite3.Connection, nama: str, kategori: str,
harga: int, stok: int) -> int:
"""Simpan produk baru, kembalikan id-nya."""
sql = "INSERT INTO produk (nama, kategori, harga, stok) VALUES (?, ?, ?, ?)"
kursor = conn.execute(sql, (nama, kategori, harga, stok))
conn.commit()
return int(kursor.lastrowid)
def lihat_produk(conn: sqlite3.Connection) -> list[sqlite3.Row]:
"""Ambil seluruh produk, urut berdasarkan nama."""
return conn.execute("SELECT * FROM produk ORDER BY nama").fetchall()
def ambil_produk(conn: sqlite3.Connection,
produk_id: int) -> sqlite3.Row | None:
"""Ambil satu produk berdasarkan id; None bila tidak ada."""
kursor = conn.execute("SELECT * FROM produk WHERE id = ?", (produk_id,))
return kursor.fetchone()
def perbarui_harga(conn: sqlite3.Connection, produk_id: int,
harga_baru: int) -> bool:
"""Set harga produk menjadi nilai baru. False bila id tidak ada."""
kursor = conn.execute(
"UPDATE produk SET harga = ? WHERE id = ?", (harga_baru, produk_id)
)
conn.commit()
return kursor.rowcount > 0
def sesuaikan_stok(conn: sqlite3.Connection, produk_id: int, perubahan: int,
keterangan: str) -> None:
"""Ubah stok sebesar perubahan (+/-) dan catat ke riwayat secara atomik."""
with conn: # commit bila sukses, rollback bila ada galat
kursor = conn.execute(
"UPDATE produk SET stok = stok + ? WHERE id = ?",
(perubahan, produk_id),
)
if kursor.rowcount == 0:
raise ValueError(f"Produk dengan id {produk_id} tidak ditemukan.")
conn.execute(
"INSERT INTO riwayat (produk_id, perubahan, keterangan)"
" VALUES (?, ?, ?)",
(produk_id, perubahan, keterangan),
)
def lihat_riwayat(conn: sqlite3.Connection,
produk_id: int) -> list[sqlite3.Row]:
"""Ambil riwayat penyesuaian stok sebuah produk, terbaru lebih dulu."""
return conn.execute(
"SELECT * FROM riwayat WHERE produk_id = ? ORDER BY id DESC",
(produk_id,),
).fetchall()
def hapus_produk(conn: sqlite3.Connection, produk_id: int) -> bool:
"""Hapus produk beserta riwayatnya. False bila id tidak ada."""
kursor = conn.execute("DELETE FROM produk WHERE id = ?", (produk_id,))
conn.commit()
return kursor.rowcount > 0
def escape_like(keyword: str) -> str:
"""Amankan karakter wildcard LIKE (persen dan garis bawah)."""
return (
keyword.replace("\\", "\\\\")
.replace("%", "\\%")
.replace("_", "\\_")
)
def cari_produk(conn: sqlite3.Connection, keyword: str) -> list[sqlite3.Row]:
"""Cari produk yang namanya mengandung keyword (LIKE yang di-escape)."""
pola = f"%{escape_like(keyword)}%"
sql = "SELECT * FROM produk WHERE nama LIKE ? ESCAPE '\\' ORDER BY nama"
return conn.execute(sql, (pola,)).fetchall()
def filter_kategori(conn: sqlite3.Connection,
kategori: str) -> list[sqlite3.Row]:
"""Ambil seluruh produk pada satu kategori."""
sql = "SELECT * FROM produk WHERE kategori = ? ORDER BY nama"
return conn.execute(sql, (kategori,)).fetchall()
def laporan_stok_menipis(conn: sqlite3.Connection,
batas: int = BATAS_STOK_MENIPIS) -> list[sqlite3.Row]:
"""Ambil produk dengan stok di bawah batas, stok terkecil lebih dulu."""
sql = "SELECT * FROM produk WHERE stok < ? ORDER BY stok, nama"
return conn.execute(sql, (batas,)).fetchall()
def nilai_inventaris_per_kategori(
conn: sqlite3.Connection,
) -> list[sqlite3.Row]:
"""Hitung jumlah produk dan total nilai stok per kategori."""
sql = """
SELECT kategori,
COUNT(*) AS jumlah_produk,
SUM(harga * stok) AS nilai_total
FROM produk
GROUP BY kategori
ORDER BY nilai_total DESC
"""
return conn.execute(sql).fetchall()
def top_produk(conn: sqlite3.Connection,
jumlah: int = JUMLAH_TOP_PRODUK) -> list[sqlite3.Row]:
"""Ambil produk dengan nilai stok terbesar."""
sql = """
SELECT id, nama, kategori, harga, stok,
harga * stok AS nilai
FROM produk
ORDER BY nilai DESC, nama
LIMIT ?
"""
return conn.execute(sql, (jumlah,)).fetchall()
main.py
Unduh berkas"""Aplikasi Inventaris SQLite - antarmuka pengguna."""
import sqlite3
import database
def format_rupiah(nilai: int) -> str:
"""Format angka menjadi teks rupiah, contoh 15000 menjadi Rp15.000."""
return "Rp" + f"{nilai:,}".replace(",", ".")
def cetak_tabel(rows: list) -> None:
"""Cetak daftar produk sebagai tabel rapi."""
print(f"{'ID':<3} {'Nama':<18} {'Kategori':<10} {'Harga':>10} {'Stok':>4}")
print(f"{'---':<3} {'-' * 18} {'-' * 10} {'-' * 10} {'-' * 4}")
for row in rows:
print(
f"{row['id']:<3} {row['nama']:<18} {row['kategori']:<10} "
f"{format_rupiah(row['harga']):>10} {row['stok']:>4}"
)
def baca_angka(prompt: str) -> int:
"""Minta masukan angka bulat, ulangi sampai masukan sah."""
while True:
teks = input(prompt).strip()
try:
return int(teks)
except ValueError:
print("Masukkan angka bulat, misalnya 15000.")
def menu_tambah(conn: sqlite3.Connection) -> None:
nama = input("Nama produk: ").strip()
kategori = input("Kategori: ").strip()
harga = baca_angka("Harga (Rp): ")
stok = baca_angka("Stok awal: ")
try:
database.validasi_produk(nama, kategori, harga, stok)
except ValueError as galat:
print(f"Gagal: {galat}")
return
try:
id_baru = database.tambah_produk(conn, nama, kategori, harga, stok)
except sqlite3.IntegrityError:
print(f"Produk dengan nama '{nama}' sudah terdaftar.")
return
print(f"Produk '{nama}' tersimpan dengan id {id_baru}.")
def menu_lihat(conn: sqlite3.Connection) -> None:
rows = database.lihat_produk(conn)
if not rows:
print("Belum ada produk.")
return
cetak_tabel(rows)
def menu_cari(conn: sqlite3.Connection) -> None:
keyword = input("Kata kunci nama: ").strip()
if not keyword:
print("Kata kunci tidak boleh kosong.")
return
hasil = database.cari_produk(conn, keyword)
if not hasil:
print(f"Tidak ada produk yang mengandung '{keyword}'.")
return
print(f"Hasil pencarian untuk '{keyword}':")
cetak_tabel(hasil)
def menu_filter(conn: sqlite3.Connection) -> None:
kategori = input("Kategori: ").strip()
if not kategori:
print("Kategori tidak boleh kosong.")
return
hasil = database.filter_kategori(conn, kategori)
if not hasil:
print(f"Tidak ada produk pada kategori '{kategori}'.")
return
print(f"Produk pada kategori '{kategori}':")
cetak_tabel(hasil)
def menu_ubah_harga(conn: sqlite3.Connection) -> None:
produk_id = baca_angka("ID produk: ")
harga_baru = baca_angka("Harga baru (Rp): ")
if database.perbarui_harga(conn, produk_id, harga_baru):
print("Harga berhasil diperbarui.")
else:
print(f"Tidak ada produk dengan id {produk_id}.")
def menu_sesuaikan_stok(conn: sqlite3.Connection) -> None:
produk_id = baca_angka("ID produk: ")
perubahan = baca_angka("Perubahan stok (boleh negatif): ")
keterangan = input("Keterangan: ").strip() or "penyesuaian manual"
try:
database.sesuaikan_stok(conn, produk_id, perubahan, keterangan)
except ValueError as galat:
print(f"Gagal: {galat}")
return
except sqlite3.IntegrityError:
print("Gagal: penyesuaian membuat stok negatif. Transaksi dibatalkan.")
return
print("Stok berhasil disesuaikan.")
def menu_hapus(conn: sqlite3.Connection) -> None:
produk_id = baca_angka("ID produk: ")
if database.hapus_produk(conn, produk_id):
print("Produk berhasil dihapus.")
else:
print(f"Tidak ada produk dengan id {produk_id}.")
def menu_riwayat(conn: sqlite3.Connection) -> None:
produk_id = baca_angka("ID produk: ")
produk = database.ambil_produk(conn, produk_id)
if produk is None:
print(f"Tidak ada produk dengan id {produk_id}.")
return
rows = database.lihat_riwayat(conn, produk_id)
if not rows:
print(f"Belum ada riwayat untuk '{produk['nama']}'.")
return
print(f"Riwayat untuk '{produk['nama']}':")
print(f"{'Waktu':<21} {'Perubahan':>9} Keterangan")
for row in rows:
print(f"{row['waktu']:<21} {row['perubahan']:>+9} {row['keterangan']}")
def menu_stok_menipis(conn: sqlite3.Connection) -> None:
hasil = database.laporan_stok_menipis(conn)
if not hasil:
print(f"Tidak ada produk dengan stok di bawah {database.BATAS_STOK_MENIPIS}.")
return
print(f"Produk dengan stok di bawah {database.BATAS_STOK_MENIPIS}:")
print(f"{'ID':<3} {'Nama':<18} {'Kategori':<10} {'Stok':>4}")
print(f"{'---':<3} {'-' * 18} {'-' * 10} {'-' * 4}")
for row in hasil:
print(f"{row['id']:<3} {row['nama']:<18} {row['kategori']:<10} {row['stok']:>4}")
def menu_nilai_inventaris(conn: sqlite3.Connection) -> None:
hasil = database.nilai_inventaris_per_kategori(conn)
if not hasil:
print("Belum ada produk untuk dihitung.")
return
print(f"{'Kategori':<12} {'Jumlah':>7} {'Nilai Total':>13}")
print(f"{'-' * 12} {'-' * 7} {'-' * 13}")
total_nilai = 0
total_produk = 0
for row in hasil:
print(
f"{row['kategori']:<12} {row['jumlah_produk']:>7} "
f"{format_rupiah(row['nilai_total']):>13}"
)
total_nilai += row["nilai_total"]
total_produk += row["jumlah_produk"]
print(f"{'-' * 12} {'-' * 7} {'-' * 13}")
print(f"{'TOTAL':<12} {total_produk:>7} {format_rupiah(total_nilai):>13}")
def menu_top_produk(conn: sqlite3.Connection) -> None:
hasil = database.top_produk(conn)
if not hasil:
print("Belum ada produk untuk dihitung.")
return
print(f"Top {database.JUMLAH_TOP_PRODUK} produk berdasarkan nilai stok:")
print(f"{'#':<3} {'Nama':<18} {'Harga':>10} {'Stok':>4} {'Nilai':>12}")
print(f"{'---':<3} {'-' * 18} {'-' * 10} {'-' * 4} {'-' * 12}")
for peringkat, row in enumerate(hasil, start=1):
print(
f"{peringkat:<3} {row['nama']:<18} "
f"{format_rupiah(row['harga']):>10} {row['stok']:>4} "
f"{format_rupiah(row['nilai']):>12}"
)
def tampilkan_menu() -> None:
print()
print("=== Aplikasi Inventaris ===")
print("1. Tambah produk")
print("2. Lihat semua produk")
print("3. Ubah harga")
print("4. Sesuaikan stok (+/-)")
print("5. Hapus produk")
print("6. Riwayat produk")
print("7. Cari produk")
print("8. Filter per kategori")
print("9. Laporan stok menipis")
print("10. Nilai inventaris per kategori")
print("11. Top produk")
print("0. Keluar")
def tangani_pilihan(conn: sqlite3.Connection, pilihan: str) -> None:
match pilihan:
case "1":
menu_tambah(conn)
case "2":
menu_lihat(conn)
case "3":
menu_ubah_harga(conn)
case "4":
menu_sesuaikan_stok(conn)
case "5":
menu_hapus(conn)
case "6":
menu_riwayat(conn)
case "7":
menu_cari(conn)
case "8":
menu_filter(conn)
case "9":
menu_stok_menipis(conn)
case "10":
menu_nilai_inventaris(conn)
case "11":
menu_top_produk(conn)
case _:
print("Pilihan tidak dikenal.")
def main() -> None:
database.inisialisasi()
conn = database.koneksi()
try:
print("Database siap. Selamat datang!")
while True:
tampilkan_menu()
try:
pilihan = input("Pilih menu: ").strip()
except EOFError:
print()
break
if pilihan == "0":
print("Sampai jumpa!")
break
tangani_pilihan(conn, pilihan)
finally:
conn.close()
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("\nProgram dihentikan.")
test_database.py
Unduh berkas"""Uji otomatis untuk lapisan data aplikasi inventaris.
Jalankan dari folder proyek:
python3 -m pytest test_database.py -v
"""
import sqlite3
from pathlib import Path
import pytest
import database
BERKAS_SKEMA = Path(database.__file__).with_name("schema.sql")
@pytest.fixture
def conn() -> sqlite3.Connection:
"""Database SQLite :memory: dengan skema dan data awal."""
koneksi_uji = sqlite3.connect(":memory:")
koneksi_uji.execute("PRAGMA foreign_keys = ON")
koneksi_uji.row_factory = sqlite3.Row
koneksi_uji.executescript(BERKAS_SKEMA.read_text(encoding="utf-8"))
yield koneksi_uji
koneksi_uji.close()
def test_inisialisasi_idempoten(tmp_path):
"""inisialisasi dua kali tidak menduplikasi tabel maupun data."""
berkas_db = tmp_path / "uji.db"
database.inisialisasi(str(berkas_db))
database.inisialisasi(str(berkas_db))
conn = sqlite3.connect(berkas_db)
try:
tabel = {
row[0]
for row in conn.execute(
"SELECT name FROM sqlite_master WHERE type = 'table'"
)
}
assert {"produk", "riwayat"} <= tabel
jumlah = conn.execute("SELECT COUNT(*) FROM produk").fetchone()[0]
assert jumlah == 6
finally:
conn.close()
def test_tambah_lalu_ambil(conn):
id_baru = database.tambah_produk(conn, "Susu Kotak", "Minuman", 6000, 12)
produk = database.ambil_produk(conn, id_baru)
assert produk["nama"] == "Susu Kotak"
assert produk["kategori"] == "Minuman"
assert produk["harga"] == 6000
assert produk["stok"] == 12
def test_nama_duplikat_ditolak(conn):
with pytest.raises(sqlite3.IntegrityError):
database.tambah_produk(conn, "Kopi Tubruk", "Minuman", 15000, 5)
@pytest.mark.parametrize(
"nama, kategori, harga, stok",
[
("", "Minuman", 1000, 5),
(" ", "Minuman", 1000, 5),
("Teh", "", 1000, 5),
("Teh", "Minuman", -1, 5),
("Teh", "Minuman", 1000, -5),
],
)
def test_validasi_produk_menolak_masukan_salah(nama, kategori, harga, stok):
with pytest.raises(ValueError):
database.validasi_produk(nama, kategori, harga, stok)
def test_sesuaikan_stok_mencatat_riwayat(conn):
database.sesuaikan_stok(conn, 1, 5, "restock mingguan")
produk = database.ambil_produk(conn, 1)
assert produk["stok"] == 24 + 5
riwayat = database.lihat_riwayat(conn, 1)
assert len(riwayat) == 1
assert riwayat[0]["perubahan"] == 5
assert riwayat[0]["keterangan"] == "restock mingguan"
def test_sesuaikan_stok_negatif_dibatalkan(conn):
with pytest.raises(sqlite3.IntegrityError):
database.sesuaikan_stok(conn, 1, -9999, "salah input")
produk = database.ambil_produk(conn, 1)
assert produk["stok"] == 24
assert len(database.lihat_riwayat(conn, 1)) == 0
def test_sesuaikan_stok_produk_hilang(conn):
with pytest.raises(ValueError):
database.sesuaikan_stok(conn, 999, 1, "uji")
def test_hapus_produk_menghapus_riwayat(conn):
database.sesuaikan_stok(conn, 1, 5, "restock")
assert len(database.lihat_riwayat(conn, 1)) == 1
assert database.hapus_produk(conn, 1) is True
assert database.ambil_produk(conn, 1) is None
assert len(database.lihat_riwayat(conn, 1)) == 0
def test_perbarui_harga(conn):
assert database.perbarui_harga(conn, 1, 17500) is True
assert database.ambil_produk(conn, 1)["harga"] == 17500
assert database.perbarui_harga(conn, 999, 17500) is False
def test_cari_produk_dengan_karakter_like(conn):
database.tambah_produk(conn, "Coklat Batang 80%", "Makanan", 12000, 10)
hasil = database.cari_produk(conn, "%")
assert [row["nama"] for row in hasil] == ["Coklat Batang 80%"]
def test_filter_kategori(conn):
hasil = database.filter_kategori(conn, "Minuman")
assert [row["nama"] for row in hasil] == ["Kopi Tubruk", "Teh Celup"]
def test_laporan_stok_menipis(conn):
hasil = database.laporan_stok_menipis(conn)
assert [(row["nama"], row["stok"]) for row in hasil] == [
("Keripik Singkong", 2),
("Teh Celup", 3),
("Pulpen Gel", 4),
]
def test_nilai_inventaris_per_kategori(conn):
hasil = database.nilai_inventaris_per_kategori(conn)
assert [(row["kategori"], row["nilai_total"]) for row in hasil] == [
("Minuman", 384000),
("Alat Tulis", 214000),
("Makanan", 200000),
]
def test_top_produk(conn):
hasil = database.top_produk(conn, jumlah=3)
assert [(row["nama"], row["nilai"]) for row in hasil] == [
("Kopi Tubruk", 360000),
("Buku Tulis", 200000),
("Roti Tawar", 180000),
]