Bagian 4: Testing dan Backup

"""Backup dan restore database Pencatat Keuangan Pribadi."""

import sqlite3
from contextlib import closing
from pathlib import Path

import db


def buat_backup(path_sumber, path_tujuan):
    """Salin database secara konsisten dengan backup API, lalu verifikasi hasil."""
    with closing(db.get_connection(path_sumber)) as sumber:
        with closing(sqlite3.connect(path_tujuan)) as tujuan:
            sumber.backup(tujuan)
    if not verifikasi_integritas(path_tujuan):
        raise sqlite3.DatabaseError(f"Integritas {path_tujuan} tidak lolos uji.")
    return path_tujuan


def restore_backup(path_backup, path_tujuan):
    """Pulihkan database dari berkas backup, lalu verifikasi hasilnya."""
    with closing(db.get_connection(path_backup)) as sumber:
        with closing(sqlite3.connect(path_tujuan)) as tujuan:
            sumber.backup(tujuan)
    if not verifikasi_integritas(path_tujuan):
        raise sqlite3.DatabaseError(f"Integritas {path_tujuan} tidak lolos uji.")
    return path_tujuan


def verifikasi_integritas(path):
    """Jalankan PRAGMA integrity_check pada satu berkas database."""
    with closing(db.get_connection(path)) as con:
        hasil = con.execute("PRAGMA integrity_check").fetchone()
        return hasil[0] == "ok"


def main():
    while True:
        print("=== Backup dan Restore ===")
        print("1. Buat backup")
        print("2. Restore dari backup")
        print("3. Keluar")
        pilihan = input("Pilihan: ").strip()

        if pilihan == "1":
            if not Path(db.NAMA_DB).exists():
                print(f"Berkas {db.NAMA_DB} belum ada — jalankan main.py dulu.")
                continue
            tujuan = input("Nama berkas backup: ").strip()
            if not tujuan:
                print("Nama berkas tidak boleh kosong.")
                continue
            try:
                buat_backup(db.NAMA_DB, tujuan)
            except sqlite3.DatabaseError as e:
                print(f"Backup gagal: {e}")
                continue
            print(f"Backup {db.NAMA_DB} tersimpan ke {tujuan}.")
            print("Verifikasi integritas: ok")
        elif pilihan == "2":
            sumber = input("Nama berkas backup: ").strip()
            if not sumber:
                print("Nama berkas tidak boleh kosong.")
                continue
            if not Path(sumber).exists():
                print(f"Berkas {sumber} tidak ditemukan.")
                continue
            try:
                restore_backup(sumber, db.NAMA_DB)
            except sqlite3.DatabaseError as e:
                print(f"Restore gagal: {e}")
                continue
            print(f"Database {db.NAMA_DB} dipulihkan dari {sumber}.")
            print("Verifikasi integritas: ok")
        elif pilihan == "3":
            print("Sampai jumpa!")
            break
        else:
            print("Pilihan tidak dikenal.")


if __name__ == "__main__":
    main()

"""Modul akses data Pencatat Keuangan Pribadi.

Satu-satunya tempat yang membuka koneksi database dan menjalankan SQL;
modul lain cukup memanggil fungsi-fungsi di bawah.
"""

import sqlite3
from contextlib import closing
from pathlib import Path

NAMA_DB = "keuangan.db"
BERKAS_SKEMA = Path(__file__).with_name("schema.sql")


def get_connection(path=NAMA_DB):
    """Buka koneksi baru dengan pengaturan seragam untuk seluruh aplikasi."""
    con = sqlite3.connect(path)
    con.row_factory = sqlite3.Row
    con.execute("PRAGMA foreign_keys = ON")
    con.execute("PRAGMA busy_timeout = 5000")
    return con


def init_db(path=NAMA_DB):
    """Jalankan schema.sql; aman dipanggil berulang karena memakai IF NOT EXISTS."""
    skema = BERKAS_SKEMA.read_text(encoding="utf-8")
    with closing(get_connection(path)) as con:
        con.executescript(skema)


def format_rupiah(jumlah):
    """Format angka menjadi teks rupiah, mis. 5500000 menjadi Rp5.500.000."""
    return f"Rp{jumlah:,.0f}".replace(",", ".")


def insert_transaksi(tanggal, jenis, kategori, jumlah, catatan="", path=NAMA_DB):
    """Simpan satu transaksi, kembalikan id baris baru."""
    with closing(get_connection(path)) as con:
        with con:
            cur = con.execute(
                "INSERT INTO transaksi (tanggal, jenis, kategori, jumlah, catatan)"
                " VALUES (?, ?, ?, ?, ?)",
                (tanggal, jenis, kategori, jumlah, catatan),
            )
            return cur.lastrowid


def transaksi_terakhir(batas=10, path=NAMA_DB):
    """Ambil sejumlah transaksi terbaru, ditata dari yang paling baru."""
    with closing(get_connection(path)) as con:
        return con.execute(
            "SELECT id, tanggal, jenis, kategori, jumlah, catatan"
            " FROM transaksi ORDER BY id DESC LIMIT ?",
            (batas,),
        ).fetchall()

"""Laporan bulanan untuk Pencatat Keuangan Pribadi."""

from contextlib import closing

import db


def ringkasan_bulanan(bulan, path=db.NAMA_DB):
    """Totalkan pemasukan, pengeluaran, dan saldo satu bulan (format YYYY-MM)."""
    with closing(db.get_connection(path)) as con:
        return con.execute(
            "SELECT"
            " COALESCE(SUM(CASE WHEN jenis = 'masuk' THEN jumlah END), 0) AS total_masuk,"
            " COALESCE(SUM(CASE WHEN jenis = 'keluar' THEN jumlah END), 0) AS total_keluar,"
            " COALESCE(SUM(CASE WHEN jenis = 'masuk' THEN jumlah END), 0)"
            " - COALESCE(SUM(CASE WHEN jenis = 'keluar' THEN jumlah END), 0) AS saldo"
            " FROM transaksi WHERE strftime('%Y-%m', tanggal) = ?",
            (bulan,),
        ).fetchone()


def kategori_terbesar(bulan, batas=3, path=db.NAMA_DB):
    """Ambil kategori pengeluaran terbesar pada satu bulan."""
    with closing(db.get_connection(path)) as con:
        return con.execute(
            "SELECT kategori, COUNT(*) AS jumlah_transaksi, SUM(jumlah) AS total"
            " FROM transaksi"
            " WHERE jenis = 'keluar' AND strftime('%Y-%m', tanggal) = ?"
            " GROUP BY kategori"
            " ORDER BY total DESC, kategori"
            " LIMIT ?",
            (bulan, batas),
        ).fetchall()


def rincian_per_kategori(bulan, path=db.NAMA_DB):
    """Hitung jumlah transaksi serta total masuk dan keluar tiap kategori."""
    with closing(db.get_connection(path)) as con:
        return con.execute(
            "SELECT kategori, COUNT(*) AS jumlah_transaksi,"
            " SUM(CASE WHEN jenis = 'masuk' THEN jumlah ELSE 0 END) AS total_masuk,"
            " SUM(CASE WHEN jenis = 'keluar' THEN jumlah ELSE 0 END) AS total_keluar"
            " FROM transaksi"
            " WHERE strftime('%Y-%m', tanggal) = ?"
            " GROUP BY kategori"
            " ORDER BY total_keluar DESC, kategori",
            (bulan,),
        ).fetchall()


def tampilkan_laporan(bulan, path=db.NAMA_DB):
    """Cetak laporan lengkap satu bulan ke layar."""
    ringkasan = ringkasan_bulanan(bulan, path)
    if ringkasan["total_masuk"] == 0 and ringkasan["total_keluar"] == 0:
        print(f"Tidak ada transaksi pada bulan {bulan}.")
        return

    print(f"=== Laporan Bulan {bulan} ===")
    print(f"Total pemasukan  : {db.format_rupiah(ringkasan['total_masuk'])}")
    print(f"Total pengeluaran: {db.format_rupiah(ringkasan['total_keluar'])}")
    print(f"Saldo            : {db.format_rupiah(ringkasan['saldo'])}")

    print()
    print("Pengeluaran terbesar:")
    for nomor, baris in enumerate(kategori_terbesar(bulan, path=path), start=1):
        print(
            f"{nomor}. {baris['kategori']:<13}"
            f" {db.format_rupiah(baris['total'])} ({baris['jumlah_transaksi']} transaksi)"
        )

    print()
    print("Rincian per kategori:")
    for baris in rincian_per_kategori(bulan, path=path):
        if baris["total_masuk"] > 0:
            print(
                f"{baris['kategori']:<14} {baris['jumlah_transaksi']:>2} transaksi"
                f"  masuk   {db.format_rupiah(baris['total_masuk'])}"
            )
        if baris["total_keluar"] > 0:
            print(
                f"{baris['kategori']:<14} {baris['jumlah_transaksi']:>2} transaksi"
                f"  keluar  {db.format_rupiah(baris['total_keluar'])}"
            )


def main():
    db.init_db()
    bulan = input("Bulan laporan (YYYY-MM): ").strip()
    tampilkan_laporan(bulan)


if __name__ == "__main__":
    main()

"""Pencatat Keuangan Pribadi — aplikasi CLI berbasis SQLite."""

import sqlite3
from datetime import date

import db
import laporan


def tampilkan_menu():
    print("=== Pencatat Keuangan ===")
    print("1. Catat pemasukan")
    print("2. Catat pengeluaran")
    print("3. Lihat transaksi terakhir")
    print("4. Laporan bulanan")
    print("5. Keluar")


def baca_tanggal():
    while True:
        teks = input("Tanggal (YYYY-MM-DD, Enter = hari ini): ").strip()
        if not teks:
            return date.today().isoformat()
        try:
            return date.fromisoformat(teks).isoformat()
        except ValueError:
            print("Format tanggal harus YYYY-MM-DD, mis. 2025-06-15.")


def baca_jumlah():
    while True:
        teks = input("Jumlah (Rp): ").strip()
        try:
            return float(teks)
        except ValueError:
            print("Jumlah harus berupa angka, mis. 25000.")


def baca_teks(label):
    while True:
        teks = input(label).strip()
        if teks:
            return teks
        print("Masukan tidak boleh kosong.")


def catat_transaksi(jenis):
    if jenis == "masuk":
        print("--- Catat Pemasukan ---")
    else:
        print("--- Catat Pengeluaran ---")
    tanggal = baca_tanggal()
    kategori = baca_teks("Kategori: ")
    jumlah = baca_jumlah()
    catatan = input("Catatan (opsional): ").strip()

    try:
        id_baru = db.insert_transaksi(tanggal, jenis, kategori, jumlah, catatan)
    except sqlite3.IntegrityError as e:
        print(f"Transaksi ditolak database: {e}")
        return
    print(f"Transaksi tersimpan dengan id {id_baru}.")


def tampilkan_transaksi():
    daftar = db.transaksi_terakhir()
    if not daftar:
        print("Belum ada transaksi.")
        return
    print(f"{'id':>3}  {'tanggal':<10}  {'jenis':<7}  {'kategori':<14}  {'jumlah':>12}")
    for t in daftar:
        print(
            f"{t['id']:>3}  {t['tanggal']:<10}  {t['jenis']:<7}  "
            f"{t['kategori']:<14}  {db.format_rupiah(t['jumlah']):>12}"
        )


def main():
    db.init_db()
    while True:
        tampilkan_menu()
        pilihan = input("Pilihan: ").strip()

        if pilihan == "1":
            catat_transaksi("masuk")
        elif pilihan == "2":
            catat_transaksi("keluar")
        elif pilihan == "3":
            tampilkan_transaksi()
        elif pilihan == "4":
            bulan = input("Bulan laporan (YYYY-MM): ").strip()
            laporan.tampilkan_laporan(bulan)
        elif pilihan == "5":
            print("Sampai jumpa!")
            break
        else:
            print("Pilihan tidak dikenal.")


if __name__ == "__main__":
    main()

-- Skema Pencatat Keuangan Pribadi.
-- Satu tabel transaksi menampung pemasukan dan pengeluaran sekaligus.

CREATE TABLE IF NOT EXISTS transaksi (
    id       INTEGER PRIMARY KEY,
    tanggal  TEXT NOT NULL CHECK (tanggal IS date(tanggal)),
    jenis    TEXT NOT NULL CHECK (jenis IN ('masuk', 'keluar')),
    kategori TEXT NOT NULL,
    jumlah   REAL NOT NULL CHECK (jumlah > 0),
    catatan  TEXT NOT NULL DEFAULT ''
);

"""Test untuk Pencatat Keuangan Pribadi."""

import sqlite3

import pytest

import backup
import db
import laporan


@pytest.fixture
def path_db(tmp_path):
    """Siapkan database bersih dengan skema sudah dibuat."""
    path = tmp_path / "keuangan.db"
    db.init_db(path)
    return path


def isi_contoh(path):
    """Isi database dengan transaksi tetap dari dua bulan."""
    data = [
        ("2025-06-01", "masuk", "Gaji", 5500000, "Gaji bulanan"),
        ("2025-06-03", "keluar", "Makan", 45000, "Makan siang"),
        ("2025-06-05", "keluar", "Transportasi", 25000, "Ojek ke kantor"),
        ("2025-06-20", "keluar", "Makan", 38000, "Makan malam"),
        ("2025-07-01", "masuk", "Gaji", 5500000, "Gaji bulanan"),
        ("2025-07-02", "keluar", "Hiburan", 50000, "Tiket bioskop"),
    ]
    for baris in data:
        db.insert_transaksi(*baris, path=path)


def test_init_db_bisa_dipanggil_berulang(tmp_path):
    path = tmp_path / "keuangan.db"
    db.init_db(path)
    db.init_db(path)

    con = db.get_connection(path)
    jumlah_tabel = con.execute(
        "SELECT COUNT(*) FROM sqlite_master WHERE name = 'transaksi'"
    ).fetchone()[0]
    con.close()
    assert jumlah_tabel == 1


def test_insert_dan_baca_bolak_balik(path_db):
    id_baru = db.insert_transaksi(
        "2025-06-01", "masuk", "Gaji", 5500000, "Gaji bulanan", path=path_db
    )

    hasil = db.transaksi_terakhir(path=path_db)
    assert len(hasil) == 1
    baris = hasil[0]
    assert baris["id"] == id_baru
    assert baris["tanggal"] == "2025-06-01"
    assert baris["jenis"] == "masuk"
    assert baris["kategori"] == "Gaji"
    assert baris["jumlah"] == 5500000
    assert baris["catatan"] == "Gaji bulanan"


def test_transaksi_terakhir_urutan_dan_batas(path_db):
    isi_contoh(path_db)

    hasil = db.transaksi_terakhir(batas=4, path=path_db)

    assert [t["id"] for t in hasil] == [6, 5, 4, 3]


def test_jumlah_harus_positif(path_db):
    for jumlah_buruk in (0, -5000):
        with pytest.raises(sqlite3.IntegrityError):
            db.insert_transaksi("2025-06-01", "masuk", "Gaji", jumlah_buruk, path=path_db)


def test_jenis_harus_masuk_atau_keluar(path_db):
    with pytest.raises(sqlite3.IntegrityError):
        db.insert_transaksi("2025-06-01", "transfer", "Lain-lain", 10000, path=path_db)


def test_tanggal_harus_format_iso(path_db):
    with pytest.raises(sqlite3.IntegrityError):
        db.insert_transaksi("15-06-2025", "masuk", "Gaji", 5500000, path=path_db)


def test_ringkasan_bulanan(path_db):
    isi_contoh(path_db)

    ringkasan = laporan.ringkasan_bulanan("2025-06", path=path_db)

    assert ringkasan["total_masuk"] == 5500000
    assert ringkasan["total_keluar"] == 108000
    assert ringkasan["saldo"] == 5392000


def test_ringkasan_bulan_tanpa_transaksi(path_db):
    ringkasan = laporan.ringkasan_bulanan("2025-12", path=path_db)

    assert ringkasan["total_masuk"] == 0
    assert ringkasan["total_keluar"] == 0
    assert ringkasan["saldo"] == 0


def test_kategori_terbesar(path_db):
    isi_contoh(path_db)

    hasil = laporan.kategori_terbesar("2025-06", path=path_db)
    assert [(r["kategori"], r["jumlah_transaksi"], r["total"]) for r in hasil] == [
        ("Makan", 2, 83000),
        ("Transportasi", 1, 25000),
    ]

    satu = laporan.kategori_terbesar("2025-06", batas=1, path=path_db)
    assert [r["kategori"] for r in satu] == ["Makan"]


def test_backup_dan_restore(tmp_path):
    path_db = tmp_path / "keuangan.db"
    db.init_db(path_db)
    isi_contoh(path_db)
    path_backup = tmp_path / "keuangan-salinan.db"

    backup.buat_backup(path_db, path_backup)
    assert backup.verifikasi_integritas(path_backup)

    db.insert_transaksi("2025-07-10", "keluar", "Makan", 30000, path=path_db)
    assert len(db.transaksi_terakhir(path=path_db)) == 7

    backup.restore_backup(path_backup, path_db)
    assert len(db.transaksi_terakhir(path=path_db)) == 6
    assert backup.verifikasi_integritas(path_db)


def test_restore_berkas_rusak(tmp_path):
    path_rusak = tmp_path / "rusak.db"
    path_rusak.write_text("ini bukan berkas database", encoding="utf-8")

    with pytest.raises(sqlite3.DatabaseError):
        backup.restore_backup(path_rusak, tmp_path / "tujuan.db")