import sqlite3 import tempfile import unittest from pathlib import Path from unittest.mock import patch from pypos.modules.penjualan.models.settlement_model import SettlementModel def _seed_settlement_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, oleh_id INTEGER DEFAULT 0, oleh_nama TEXT, cabang_id INTEGER DEFAULT 0, cabang_nama TEXT, transaksi_nilai REAL, transaksi_bulat REAL, transaksi_dibayar REAL, settlement_id INTEGER, bank_nama TEXT, pembayaran_non_tunai REAL, 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_ord_hrg REAL, produk_ord_jml INTEGER, trash INTEGER DEFAULT 0 ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS transaksi_settlement ( id INTEGER PRIMARY KEY AUTOINCREMENT, counter TEXT, oleh_id INTEGER, oleh_dtime TEXT, approval_id INTEGER, approval_counter TEXT, approval_nama TEXT, status INTEGER, data_transaksi_id TEXT, cabang_id INTEGER, oleh_nama TEXT, cabang_nama TEXT ) """ ) conn.commit() conn.close() class SettlementModelTestCase(unittest.TestCase): def test_settlement_total_harus_net_of_return_and_voucher(self): with tempfile.TemporaryDirectory() as td: db_path = str(Path(td) / "settlement_net.db") _seed_settlement_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ INSERT INTO transaksi (id, nomer, dtime, customers_nama, oleh_nama, transaksi_nilai, transaksi_bulat, transaksi_dibayar, settlement_id, bank_nama, pembayaran_non_tunai, trash) VALUES (1, 'INV-1', '2026-02-11 10:00:00', 'Cust A', 'Kasir A', 100000, 100000, 100000, 1, 'Tunai', 0, 0) """ ) cur.execute( """ INSERT INTO transaksi_data (transaksi_id, produk_id, produk_ord_hrg, produk_ord_jml, trash) VALUES ('1', 'P-1', 10000, 10, 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, jumlah INTEGER NOT NULL DEFAULT 0, harga REAL NOT NULL DEFAULT 0 ) """ ) cur.execute( "INSERT INTO return_transaksi_penjualan (id, transaksi_id, refund_method, refund_amount) VALUES (1, '1', 'cash', 999999)" ) cur.execute( "INSERT INTO detail_return_transaksi_penjualan (return_id, produk_id, jumlah, harga) VALUES (1, 'P-1', 2, 10000)" ) cur.execute( """ CREATE TABLE IF NOT EXISTS voucher_return ( id INTEGER PRIMARY KEY AUTOINCREMENT, return_id INTEGER NULL, transaksi_id TEXT NULL, customer_id TEXT NULL, kode TEXT UNIQUE NOT NULL, nilai_awal REAL NOT NULL, saldo REAL NOT NULL, status TEXT NOT NULL DEFAULT 'aktif', dtime_terbit TEXT NOT NULL, dtime_expired TEXT 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', 15000, 0, 'habis', '2026-02-11 09:00:00') """ ) cur.execute( """ INSERT INTO voucher_usage (voucher_id, transaksi_id, nilai_pakai, dtime) VALUES (1, '1', 15000, '2026-02-11 10:01:00') """ ) conn.commit() conn.close() model = SettlementModel(db_path) model.set_settlement([1], admin="admin-a", kasir="kasir-a", total_disetor=65000) conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute("SELECT total_harus, total_disetor, total_refund_cash, status FROM settlement_history ORDER BY id DESC LIMIT 1") history = cur.fetchone() self.assertIsNotNone(history) self.assertEqual(float(history["total_harus"]), 65000.0) self.assertEqual(float(history["total_disetor"]), 65000.0) self.assertEqual(float(history["total_refund_cash"]), 20000.0) self.assertEqual(str(history["status"]), "Sesuai") cur.execute("SELECT settlement_id FROM transaksi WHERE id = 1") settlement_id = cur.fetchone()["settlement_id"] conn.close() self.assertEqual(int(settlement_id), 0) # edited by glg def test_cek_transaksi_settlement_connect_error_tidak_memicu_unboundlocal(self): model = SettlementModel(db_path=":memory:") with patch.object(model, "connect", side_effect=sqlite3.Error("db_down")): result = model.cek_transaksi_settlement(include_today=False) self.assertEqual(result, []) def test_settlement_selisih_berbasis_kewajiban_tunai(self): # edited by glg with tempfile.TemporaryDirectory() as td: db_path = str(Path(td) / "settlement_cash_diff.db") _seed_settlement_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.executemany( """ INSERT INTO transaksi (id, nomer, dtime, customers_nama, oleh_nama, transaksi_nilai, transaksi_bulat, transaksi_dibayar, settlement_id, bank_nama, pembayaran_non_tunai, trash) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0) """, [ (1, "INV-TUNAI", "2026-03-30 10:00:00", "Cust A", "Kasir A", 100000, 100000, 100000, 1, "Tunai", 0), (2, "INV-DEBIT", "2026-03-30 10:05:00", "Cust B", "Kasir A", 50000, 50000, 50000, 1, "Debit", 50000), ], ) conn.commit() conn.close() model = SettlementModel(db_path) ok = model.set_settlement([1, 2], admin="admin-a", kasir="kasir-a", total_disetor=100000) self.assertTrue(bool(ok)) conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute( "SELECT total_harus, total_non_tunai, total_disetor, selisih, status FROM settlement_history ORDER BY id DESC LIMIT 1" ) history = cur.fetchone() conn.close() self.assertIsNotNone(history) self.assertEqual(float(history["total_harus"]), 150000.0) self.assertEqual(float(history["total_non_tunai"]), 50000.0) self.assertEqual(float(history["total_disetor"]), 100000.0) self.assertEqual(float(history["selisih"]), 0.0) self.assertEqual(str(history["status"]), "Sesuai") # edited by glg def test_hitung_total_per_metode_by_ids_hanya_scope_batch(self): with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: db_path = str(Path(td) / "settlement_scope_batch.db") _seed_settlement_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.executemany( """ INSERT INTO transaksi (id, nomer, dtime, customers_nama, oleh_nama, transaksi_nilai, transaksi_bulat, transaksi_dibayar, settlement_id, bank_nama, pembayaran_non_tunai, trash) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0) """, [ (1, "INV-1", "2026-03-06 10:00:00", "Cust A", "Kasir A", 100000, 100000, 100000, 0, "Tunai", 0), (2, "INV-2", "2026-03-06 11:00:00", "Cust B", "Kasir A", 50000, 50000, 50000, 0, "BCA", 50000), (3, "INV-3", "2026-03-06 12:00:00", "Cust C", "Kasir A", 25000, 25000, 25000, 0, "Tunai", 0), ], ) conn.commit() conn.close() model = SettlementModel(db_path) total_map = model.hitung_total_per_metode_by_ids([2]) self.assertAlmostEqual(float(total_map.get("Tunai", 0)), 0.0, places=2) self.assertAlmostEqual(float(total_map.get("Non Tunai", 0)), 50000.0, places=2) # edited by glg def test_get_settlement_lock_map_detects_history_meski_settlement_id_belum_0(self): with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: db_path = str(Path(td) / "settlement_lock_map.db") _seed_settlement_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ INSERT INTO transaksi (id, nomer, dtime, customers_nama, oleh_nama, transaksi_nilai, transaksi_bulat, transaksi_dibayar, settlement_id, bank_nama, pembayaran_non_tunai, trash) VALUES (1, 'INV-1', '2026-03-26 10:00:00', 'Cust A', 'Kasir A', 100000, 100000, 100000, 1, 'Tunai', 0, 0) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS settlement_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, tanggal TEXT, admin TEXT, kasir TEXT, total_harus REAL DEFAULT 0, total_disetor REAL DEFAULT 0, total_non_tunai REAL DEFAULT 0, total_refund_cash REAL DEFAULT 0, selisih REAL DEFAULT 0, status TEXT, data_transaksi_id TEXT ) """ ) cur.execute( """ INSERT INTO settlement_history (tanggal, admin, kasir, status, data_transaksi_id) VALUES ('2026-03-26 11:00:00', 'admin', 'kasir', 'Sesuai', '[1]') """ ) conn.commit() conn.close() model = SettlementModel(db_path) lock_map = model.get_settlement_lock_map([1, 2]) self.assertTrue(bool(lock_map.get(1))) self.assertFalse(bool(lock_map.get(2))) # edited by glg def test_get_settlement_lock_map_backfills_map_table_from_history_json(self): with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: db_path = str(Path(td) / "settlement_lock_map_backfill.db") _seed_settlement_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ INSERT INTO transaksi (id, nomer, dtime, customers_nama, oleh_nama, transaksi_nilai, transaksi_bulat, transaksi_dibayar, settlement_id, bank_nama, pembayaran_non_tunai, trash) VALUES (11, 'INV-11', '2026-03-27 10:00:00', 'Cust B', 'Kasir B', 120000, 120000, 120000, 1, 'Tunai', 0, 0) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS settlement_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, tanggal TEXT, admin TEXT, kasir TEXT, total_harus REAL DEFAULT 0, total_disetor REAL DEFAULT 0, total_non_tunai REAL DEFAULT 0, total_refund_cash REAL DEFAULT 0, selisih REAL DEFAULT 0, status TEXT, data_transaksi_id TEXT ) """ ) cur.execute( """ INSERT INTO settlement_history (tanggal, admin, kasir, status, data_transaksi_id) VALUES ('2026-03-27 11:00:00', 'admin', 'kasir', 'Sesuai', '[11, 999]') """ ) conn.commit() conn.close() model = SettlementModel(db_path) lock_map = model.get_settlement_lock_map([11]) self.assertTrue(bool(lock_map.get(11))) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ SELECT COUNT(1) FROM settlement_history_transaksi_map WHERE transaksi_id = 11 """ ) mapped_count = int(cur.fetchone()[0] or 0) conn.close() self.assertGreaterEqual(mapped_count, 1) def test_get_history_settlement_by_range_respects_config_limit(self): with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: db_path = str(Path(td) / "settlement_history_limit.db") _seed_settlement_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ CREATE TABLE IF NOT EXISTS settlement_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, tanggal TEXT, admin TEXT, kasir TEXT, total_harus REAL DEFAULT 0, total_disetor REAL DEFAULT 0, total_non_tunai REAL DEFAULT 0, total_refund_cash REAL DEFAULT 0, selisih REAL DEFAULT 0, status TEXT, data_transaksi_id TEXT ) """ ) cur.executemany( """ INSERT INTO settlement_history (tanggal, admin, kasir, total_harus, total_disetor, total_non_tunai, status, data_transaksi_id) VALUES (?, 'admin', 'kasir', 100000, 100000, 0, 'Sesuai', '[]') """, [ ("2026-03-01 10:00:00",), ("2026-03-02 10:00:00",), ("2026-03-03 10:00:00",), ], ) conn.commit() conn.close() model = SettlementModel(db_path) with patch( "pypos.modules.penjualan.models.settlement_model.read_config", return_value={"settlement_history_max_rows": 2}, ): rows = model.get_history_settlement_by_range("2026-03-01", "2026-03-31") self.assertEqual(len(rows), 2) self.assertEqual(str(rows[0]["tanggal"]), "2026-03-03") self.assertEqual(str(rows[1]["tanggal"]), "2026-03-02") def test_execute_settlement_atomic_rollback_jika_set_settlement_gagal(self): with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as td: db_path = str(Path(td) / "settlement_atomic_rollback.db") _seed_settlement_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ INSERT INTO transaksi (id, nomer, dtime, customers_nama, oleh_nama, transaksi_nilai, transaksi_bulat, transaksi_dibayar, settlement_id, bank_nama, pembayaran_non_tunai, trash) VALUES (1, 'INV-1', '2026-04-01 10:00:00', 'Cust A', 'Kasir A', 100000, 100000, 100000, 1, 'Tunai', 0, 0) """ ) conn.commit() conn.close() model = SettlementModel(db_path) with patch.object(model, "set_settlement", return_value=False): result = model.execute_settlement_atomic( kasir_id=1, kasir_nama="Kasir A", cabang_id=10, cabang_nama="Cabang A", transaksi_list=[{"tanggal": "2026-04-01", "kasir": "Kasir A"}], approval_id=99, approval_nama="admin", transaksi_ids_override=[1], admin="admin", total_disetor=100000, trace_id="settlement-test-trace", ) self.assertFalse(bool((result or {}).get("ok"))) self.assertEqual(str((result or {}).get("error_code") or ""), "SETTLEMENT_STATUS_UPDATE_FAILED") self.assertEqual(str((result or {}).get("reason") or ""), "status_update_failed") self.assertEqual(str((result or {}).get("trace_id") or ""), "settlement-test-trace") conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("SELECT COUNT(1) FROM transaksi_settlement") settlement_rows = int(cur.fetchone()[0] or 0) conn.close() self.assertEqual(settlement_rows, 0) if __name__ == "__main__": unittest.main()