import sqlite3 import tempfile import unittest from pathlib import Path from pypos.modules.penjualan.errors import ReturnProcessError from pypos.modules.penjualan.models.return_model import ReturnItem, ReturnModel def _seed_transaksi_schema(db_path: str): conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ CREATE TABLE IF NOT EXISTS transaksi ( id TEXT PRIMARY KEY, customers_id TEXT, transaksi_nilai REAL, settlement_id INTEGER DEFAULT 1 ) """ ) 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, trash INTEGER DEFAULT 0 ) """ ) cur.execute( "INSERT INTO transaksi (id, customers_id, transaksi_nilai, settlement_id) VALUES (?, ?, ?, 1)", ("TRX-1", "CUST-1", 500000), ) cur.execute( "INSERT INTO transaksi_data (transaksi_id, produk_id, produk_nama, produk_ord_jml, produk_ord_hrg, trash) VALUES (?, ?, ?, ?, ?, 0)", ("TRX-1", "P-1", "Produk A", 10, 10000), ) cur.execute( "INSERT INTO transaksi_data (transaksi_id, produk_id, produk_nama, produk_ord_jml, produk_ord_hrg, trash) VALUES (?, ?, ?, ?, ?, 0)", ("TRX-1", "P-2", "Produk B", 5, 20000), ) conn.commit() conn.close() class ReturnModelTestCase(unittest.TestCase): def test_return_qty_exceeds_returnable_rejected(self): with tempfile.TemporaryDirectory() as td: db_path = str(Path(td) / "return_qty_guard.db") _seed_transaksi_schema(db_path) model = ReturnModel(db_path) try: with self.assertRaisesRegex(ValueError, "melebihi sisa qty"): model.insert_return( "TRX-1", [ReturnItem("P-1", "Produk A", 11, 10000, "partial")], "partial", "cash", ) finally: model.close() def test_return_full_note_rejected(self): with tempfile.TemporaryDirectory() as td: db_path = str(Path(td) / "return_full_note_guard.db") _seed_transaksi_schema(db_path) model = ReturnModel(db_path) try: with self.assertRaisesRegex(ValueError, "Return penuh satu nota tidak diizinkan"): model.insert_return( "TRX-1", [ ReturnItem("P-1", "Produk A", 10, 10000, "partial"), ReturnItem("P-2", "Produk B", 5, 20000, "partial"), ], "partial", "cash", ) finally: model.close() def test_return_partial_voucher_creates_voucher(self): with tempfile.TemporaryDirectory() as td: db_path = str(Path(td) / "return_voucher_success.db") _seed_transaksi_schema(db_path) model = ReturnModel(db_path) try: kode = model.insert_return( "TRX-1", [ReturnItem("P-1", "Produk A", 2, 10000, "partial")], "partial", "voucher", ) self.assertTrue(str(kode).startswith("VCR")) conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute( "SELECT refund_method, refund_amount FROM return_transaksi_penjualan WHERE kode_voucher = ?", (kode,), ) row = cur.fetchone() self.assertIsNotNone(row) self.assertEqual(str(row["refund_method"]).lower(), "voucher") self.assertEqual(float(row["refund_amount"] or 0), 20000.0) cur.execute("SELECT kode, saldo FROM voucher_return WHERE kode = ?", (kode,)) voucher = cur.fetchone() conn.close() self.assertIsNotNone(voucher) self.assertEqual(str(voucher["kode"]), kode) self.assertEqual(float(voucher["saldo"]), 20000.0) finally: model.close() # edited by glg def test_return_voucher_atomic_rollback_saat_create_voucher_gagal(self): with tempfile.TemporaryDirectory() as td: db_path = str(Path(td) / "return_voucher_atomic_rollback.db") _seed_transaksi_schema(db_path) model = ReturnModel(db_path) try: def _raise_voucher_error(*_args, **_kwargs): raise RuntimeError("voucher_insert_failed") model.voucher_model.create_voucher = _raise_voucher_error with self.assertRaisesRegex(ReturnProcessError, "RETURN_UNEXPECTED_ERROR"): model.insert_return( "TRX-1", [ReturnItem("P-1", "Produk A", 1, 10000, "partial")], "partial", "voucher", ) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("SELECT COUNT(1) FROM return_transaksi_penjualan") master_count = int(cur.fetchone()[0] or 0) cur.execute("SELECT COUNT(1) FROM detail_return_transaksi_penjualan") detail_count = int(cur.fetchone()[0] or 0) conn.close() self.assertEqual(master_count, 0) self.assertEqual(detail_count, 0) finally: model.close() # edited by glg def test_return_settled_transaction_rejected(self): with tempfile.TemporaryDirectory() as td: db_path = str(Path(td) / "return_settled_rejected.db") _seed_transaksi_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("UPDATE transaksi SET settlement_id = 0 WHERE id = 'TRX-1'") conn.commit() conn.close() model = ReturnModel(db_path) try: with self.assertRaisesRegex( ReturnProcessError, "RETURN_SETTLEMENT_LOCKED" ): model.insert_return( "TRX-1", [ReturnItem("P-1", "Produk A", 1, 10000, "partial")], "partial", "cash", ) finally: model.close() # edited by glg def test_return_rejected_when_transaksi_tercatat_di_history_settlement(self): with tempfile.TemporaryDirectory() as td: db_path = str(Path(td) / "return_hist_settlement_rejected.db") _seed_transaksi_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( "INSERT INTO transaksi (id, customers_id, transaksi_nilai, settlement_id) VALUES ('1', 'CUST-2', 120000, 1)" ) cur.execute( "INSERT INTO transaksi_data (transaksi_id, produk_id, produk_nama, produk_ord_jml, produk_ord_hrg, trash) VALUES ('1', 'P-9', 'Produk X', 3, 40000, 0)" ) cur.execute( """ CREATE TABLE IF NOT EXISTS settlement_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, data_transaksi_id TEXT ) """ ) cur.execute("INSERT INTO settlement_history (data_transaksi_id) VALUES ('[1, 123]')") conn.commit() conn.close() model = ReturnModel(db_path) try: with self.assertRaisesRegex( ReturnProcessError, "RETURN_SETTLEMENT_LOCKED" ): model.insert_return( "1", [ReturnItem("P-9", "Produk X", 1, 40000, "partial")], "partial", "cash", ) finally: model.close() # edited by glg def test_return_rejected_when_transaksi_tercatat_di_history_map(self): with tempfile.TemporaryDirectory() as td: db_path = str(Path(td) / "return_hist_map_rejected.db") _seed_transaksi_schema(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( "INSERT INTO transaksi (id, customers_id, transaksi_nilai, settlement_id) VALUES ('9', 'CUST-9', 120000, 1)" ) cur.execute( "INSERT INTO transaksi_data (transaksi_id, produk_id, produk_nama, produk_ord_jml, produk_ord_hrg, trash) VALUES ('9', 'P-9', 'Produk Z', 3, 40000, 0)" ) cur.execute( """ CREATE TABLE IF NOT EXISTS settlement_history_transaksi_map ( settlement_history_id INTEGER NOT NULL, transaksi_id INTEGER NOT NULL, PRIMARY KEY (settlement_history_id, transaksi_id) ) """ ) cur.execute( "INSERT INTO settlement_history_transaksi_map (settlement_history_id, transaksi_id) VALUES (1, 9)" ) conn.commit() conn.close() model = ReturnModel(db_path) try: with self.assertRaisesRegex( ReturnProcessError, "RETURN_SETTLEMENT_LOCKED" ): model.insert_return( "9", [ReturnItem("P-9", "Produk Z", 1, 40000, "partial")], "partial", "cash", ) finally: model.close() if __name__ == "__main__": unittest.main()