import sqlite3 import tempfile import unittest from pathlib import Path from unittest.mock import patch from pypos.modules.penjualan.models.load_transaksi_model import LoadTransaksiModel def _seed_history_schema(db_path: str): conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ CREATE TABLE IF NOT EXISTS transaksi ( id INTEGER PRIMARY KEY, nomer TEXT, dtime TEXT, customers_nama TEXT, transaksi_nilai REAL, oleh_nama TEXT, oleh_id INTEGER DEFAULT 1, settlement_id INTEGER DEFAULT 1, jenis_label TEXT, trash INTEGER DEFAULT 0 ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS transaksi_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, transaksi_id TEXT, produk_id TEXT, produk_nama TEXT, produk_ord_jml INTEGER, produk_ord_hrg REAL, produk_ord_diskon REAL DEFAULT 0, satuan TEXT, trash INTEGER DEFAULT 0 ) """ ) conn.commit() conn.close() class LoadTransaksiModelTestCase(unittest.TestCase): def test_filtered_transaksi_list_uses_net_total(self): with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: db_path = str(Path(td) / "load_history_net.db") _seed_history_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ INSERT INTO transaksi (id, nomer, dtime, customers_nama, transaksi_nilai, oleh_nama, settlement_id, jenis_label, trash) VALUES (1, 'INV-1', '2026-02-11 10:00:00', 'Cust A', 100000, 'Kasir A', 1, 'invoice', 0) """ ) cur.execute( """ INSERT INTO transaksi_data (transaksi_id, produk_id, produk_nama, produk_ord_jml, produk_ord_hrg, satuan, trash) VALUES ('1', 'P-1', 'Produk A', 10, 10000, 'pcs', 0) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS return_transaksi_penjualan ( id INTEGER PRIMARY KEY AUTOINCREMENT, transaksi_id TEXT NOT NULL, refund_method TEXT NULL, refund_amount REAL NULL ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS detail_return_transaksi_penjualan ( id INTEGER PRIMARY KEY AUTOINCREMENT, return_id INTEGER NOT NULL, produk_id TEXT NOT NULL, produk_nama TEXT, jumlah INTEGER NOT NULL DEFAULT 0, harga REAL NOT NULL DEFAULT 0, subtotal REAL NOT NULL DEFAULT 0 ) """ ) cur.execute("INSERT INTO return_transaksi_penjualan (id, transaksi_id, refund_method, refund_amount) VALUES (1, '1', 'cash', 99999)") cur.execute( "INSERT INTO detail_return_transaksi_penjualan (return_id, produk_id, produk_nama, jumlah, harga, subtotal) VALUES (1, 'P-1', 'Produk A', 2, 10000, 20000)" ) cur.execute( """ CREATE TABLE IF NOT EXISTS voucher_return ( id INTEGER PRIMARY KEY AUTOINCREMENT, kode TEXT UNIQUE NOT NULL, nilai_awal REAL NOT NULL DEFAULT 0, saldo REAL NOT NULL DEFAULT 0, status TEXT NOT NULL DEFAULT 'aktif', dtime_terbit TEXT NOT NULL ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS voucher_usage ( id INTEGER PRIMARY KEY AUTOINCREMENT, voucher_id INTEGER NOT NULL, transaksi_id TEXT NOT NULL, nilai_pakai REAL NOT NULL, dtime TEXT NOT NULL ) """ ) cur.execute( "INSERT INTO voucher_return (id, kode, nilai_awal, saldo, status, dtime_terbit) VALUES (1, 'VCR-1', 5000, 0, 'habis', '2026-02-11 09:00:00')" ) cur.execute( "INSERT INTO voucher_usage (voucher_id, transaksi_id, nilai_pakai, dtime) VALUES (1, '1', 5000, '2026-02-11 10:05:00')" ) conn.commit() conn.close() model = LoadTransaksiModel(db_path) rows = model.get_filtered_transaksi_list(include_trashed=True) self.assertTrue(rows) row = rows[0] # 100000 - 20000 - 5000 = 75000 self.assertEqual(float(row[4]), 75000.0) self.assertEqual(int(row[7]), 1) def test_voucher_usage_info_supports_legacy_kode_voucher_column(self): with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: db_path = str(Path(td) / "load_history_legacy_voucher.db") _seed_history_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ CREATE TABLE IF NOT EXISTS voucher_return ( id INTEGER PRIMARY KEY AUTOINCREMENT, kode_voucher TEXT UNIQUE, saldo REAL DEFAULT 0, status TEXT DEFAULT 'aktif' ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS voucher_usage ( id INTEGER PRIMARY KEY AUTOINCREMENT, voucher_id INTEGER NOT NULL, transaksi_id TEXT NOT NULL, nilai_pakai REAL NOT NULL, dtime TEXT NOT NULL ) """ ) cur.execute("INSERT INTO voucher_return (id, kode_voucher, saldo, status) VALUES (1, 'LEGACY-VC-1', 0, 'habis')") cur.execute( "INSERT INTO voucher_usage (voucher_id, transaksi_id, nilai_pakai, dtime) VALUES (1, '1', 12000, '2026-02-11 11:00:00')" ) conn.commit() conn.close() model = LoadTransaksiModel(db_path) rows = model.get_voucher_usage_info("1") self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["kode_voucher"], "LEGACY-VC-1") self.assertEqual(float(rows[0]["nilai_pakai"]), 12000.0) def test_filtered_transaksi_list_respects_config_limit(self): with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: db_path = str(Path(td) / "load_history_limit.db") _seed_history_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.executemany( """ INSERT INTO transaksi (id, nomer, dtime, customers_nama, transaksi_nilai, oleh_nama, settlement_id, jenis_label, trash) VALUES (?, ?, ?, ?, ?, ?, 1, 'invoice', 0) """, [ (1, "INV-1", "2026-03-01 10:00:00", "Cust A", 10000, "Kasir A"), (2, "INV-2", "2026-03-01 11:00:00", "Cust B", 20000, "Kasir A"), (3, "INV-3", "2026-03-01 12:00:00", "Cust C", 30000, "Kasir A"), ], ) cur.executemany( """ INSERT INTO transaksi_data (transaksi_id, produk_id, produk_nama, produk_ord_jml, produk_ord_hrg, satuan, trash) VALUES (?, ?, ?, 1, 10000, 'pcs', 0) """, [ ("1", "P-1", "Produk 1"), ("2", "P-2", "Produk 2"), ("3", "P-3", "Produk 3"), ], ) conn.commit() conn.close() model = LoadTransaksiModel(db_path) with patch( "pypos.modules.penjualan.models.load_transaksi_model.read_config", return_value={"history_transaksi_max_rows": 2}, ): rows = model.get_filtered_transaksi_list(include_trashed=True) self.assertEqual(len(rows), 2) self.assertEqual(int(rows[0][0]), 3) self.assertEqual(int(rows[1][0]), 2) def test_get_tersimpan_transaksi_list_respects_config_limit(self): with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: db_path = str(Path(td) / "load_saved_limit.db") _seed_history_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.executemany( """ INSERT INTO transaksi (id, nomer, dtime, customers_nama, transaksi_nilai, oleh_nama, settlement_id, jenis_label, trash) VALUES (?, ?, ?, ?, ?, ?, 1, 'simpan_transaksi', 0) """, [ (1, "SIM-1", "2026-03-01 10:00:00", "Cust A", 10000, "Kasir A"), (2, "SIM-2", "2026-03-01 11:00:00", "Cust B", 20000, "Kasir A"), (3, "SIM-3", "2026-03-01 12:00:00", "Cust C", 30000, "Kasir A"), ], ) conn.commit() conn.close() model = LoadTransaksiModel(db_path) with patch( "pypos.modules.penjualan.models.load_transaksi_model.read_config", return_value={"load_transaksi_max_rows": 2}, ): rows = model.get_tersimpan_transaksi_list() self.assertEqual(len(rows), 2) self.assertEqual(int(rows[0][0]), 3) self.assertEqual(int(rows[1][0]), 2) # edited by glg def test_get_detail_transaksi_mengabaikan_item_trash(self): with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: db_path = str(Path(td) / "load_detail_exclude_trash.db") _seed_history_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ INSERT INTO transaksi (id, nomer, dtime, customers_nama, transaksi_nilai, oleh_nama, settlement_id, jenis_label, trash) VALUES (10, 'SIM-10', '2026-03-01 10:00:00', 'Cust A', 50000, 'Kasir A', 1, 'simpan_transaksi', 0) """ ) cur.executemany( """ INSERT INTO transaksi_data (transaksi_id, produk_id, produk_nama, produk_ord_jml, produk_ord_hrg, produk_ord_diskon, satuan, trash) VALUES (?, ?, ?, ?, ?, 0, 'pcs', ?) """, [ ("10", "P-A", "Produk A", 2, 10000, 0), ("10", "P-B", "Produk B", 1, 30000, 1), ], ) conn.commit() conn.close() model = LoadTransaksiModel(db_path) detail_rows = model.get_detail_transaksi("10") self.assertEqual(len(detail_rows), 1) self.assertEqual(str(detail_rows[0][0]), "P-A") # edited by glg def test_delete_transaksi_by_id_soft_delete_jika_kolom_trash_tersedia(self): with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: db_path = str(Path(td) / "load_soft_delete.db") _seed_history_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ INSERT INTO transaksi (id, nomer, dtime, customers_nama, transaksi_nilai, oleh_nama, settlement_id, jenis_label, trash) VALUES (11, 'SIM-11', '2026-03-01 10:00:00', 'Cust A', 10000, 'Kasir A', 1, 'simpan_transaksi', 0) """ ) cur.execute( """ INSERT INTO transaksi_data (transaksi_id, produk_id, produk_nama, produk_ord_jml, produk_ord_hrg, satuan, trash) VALUES ('11', 'P-1', 'Produk 1', 1, 10000, 'pcs', 0) """ ) conn.commit() conn.close() model = LoadTransaksiModel(db_path) model.delete_transaksi_by_id("11") conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("SELECT COUNT(1) FROM transaksi WHERE id = 11") self.assertEqual(int(cur.fetchone()[0]), 1) cur.execute("SELECT COALESCE(trash, 0) FROM transaksi WHERE id = 11") self.assertEqual(int(cur.fetchone()[0]), 1) cur.execute("SELECT COALESCE(trash, 0) FROM transaksi_data WHERE transaksi_id = '11'") self.assertEqual(int(cur.fetchone()[0]), 1) conn.close() # edited by glg def test_delete_transaksi_by_id_fallback_hard_delete_untuk_skema_legacy(self): with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: db_path = str(Path(td) / "load_hard_delete_legacy.db") conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ CREATE TABLE transaksi ( id INTEGER PRIMARY KEY, nomer TEXT, jenis_label TEXT ) """ ) cur.execute( """ CREATE TABLE transaksi_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, transaksi_id TEXT, produk_id TEXT ) """ ) cur.execute("INSERT INTO transaksi (id, nomer, jenis_label) VALUES (21, 'SIM-21', 'simpan_transaksi')") cur.execute("INSERT INTO transaksi_data (transaksi_id, produk_id) VALUES ('21', 'P-1')") conn.commit() conn.close() model = LoadTransaksiModel(db_path) model._connect = lambda read_only=False: sqlite3.connect(db_path) model.delete_transaksi_by_id("21") conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("SELECT COUNT(1) FROM transaksi WHERE id = 21") self.assertEqual(int(cur.fetchone()[0]), 0) cur.execute("SELECT COUNT(1) FROM transaksi_data WHERE transaksi_id = '21'") self.assertEqual(int(cur.fetchone()[0]), 0) conn.close() # edited by glg def test_delete_transaksi_by_id_menolak_label_bukan_simpan_transaksi(self): with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: db_path = str(Path(td) / "load_delete_guard.db") _seed_history_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ INSERT INTO transaksi (id, nomer, dtime, customers_nama, transaksi_nilai, oleh_nama, settlement_id, jenis_label, trash) VALUES (31, 'INV-31', '2026-03-02 10:00:00', 'Cust A', 12000, 'Kasir A', 1, 'invoice', 0) """ ) cur.execute( """ INSERT INTO transaksi_data (transaksi_id, produk_id, produk_nama, produk_ord_jml, produk_ord_hrg, satuan, trash) VALUES ('31', 'P-31', 'Produk 31', 1, 12000, 'pcs', 0) """ ) conn.commit() conn.close() model = LoadTransaksiModel(db_path) with self.assertRaises(ValueError): model.delete_transaksi_by_id("31") conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("SELECT COALESCE(trash, 0) FROM transaksi WHERE id = 31") self.assertEqual(int(cur.fetchone()[0]), 0) cur.execute( """ SELECT status FROM transaksi_delete_audit WHERE transaksi_id = '31' ORDER BY id DESC LIMIT 1 """ ) self.assertEqual(str(cur.fetchone()[0]), "rejected_non_saved") conn.close() # edited by glg def test_delete_transaksi_by_id_mencatat_audit_sukses(self): with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: db_path = str(Path(td) / "load_delete_audit_success.db") _seed_history_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ INSERT INTO transaksi (id, nomer, dtime, customers_nama, transaksi_nilai, oleh_nama, settlement_id, jenis_label, trash) VALUES (41, 'SIM-41', '2026-03-02 11:00:00', 'Cust B', 34000, 'Kasir A', 1, 'simpan_transaksi', 0) """ ) cur.execute( """ INSERT INTO transaksi_data (transaksi_id, produk_id, produk_nama, produk_ord_jml, produk_ord_hrg, satuan, trash) VALUES ('41', 'P-41', 'Produk 41', 1, 34000, 'pcs', 0) """ ) conn.commit() conn.close() model = LoadTransaksiModel(db_path) result = model.delete_transaksi_by_id("41") self.assertTrue(bool((result or {}).get("ok"))) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ SELECT mode, status FROM transaksi_delete_audit WHERE transaksi_id = '41' ORDER BY id DESC LIMIT 1 """ ) row = cur.fetchone() self.assertIsNotNone(row) self.assertEqual(str(row[1]), "success") self.assertIn(str(row[0]), {"soft", "hard_fallback"}) conn.close() if __name__ == "__main__": unittest.main()