# edited by glg import sqlite3 import tempfile import unittest from unittest.mock import patch from pypos.modules.sinkronisasi.services.transaction_export_service import TransactionExportService class LegacyTransaksiRequiredFieldTests(unittest.TestCase): def _build_service(self): service = TransactionExportService() service._fetch_transaksi_data_by_transaksi_ids = lambda transaksi_ids, limit=0: [] service._fetch_produk_snapshot = lambda produk_ids: {} service._fetch_diskon_free_snapshot = lambda produk_ids: {} service._build_legacy_items_payload = lambda detail_rows, produk_map: {} service._build_legacy_free_items_payload = ( lambda detail_rows, produk_map, diskon_map, dtime_value: {} ) service._build_legacy_payment_payload = ( lambda transaksi_row, id_penjualan, resolved_cabang_id: [] ) service._get_export_device_context = lambda machine_id_hint="": { "machine_id": "MACHINE-1" } return service def test_legacy_transaksi_payload_includes_required_export_fields(self): service = self._build_service() transaksi_row = { "id": 10, "dtime": "2026-03-06 10:00:00", "fulldate": "2026-03-06", "nomer": "INV-10", "jenis_label": "invoice", "oleh_id": 183, "oleh_nama": "kasir glg", "cabang_id": 101, "gudang_id": -1010, "machine_id": "MACHINE-1", "transaksi_bulat": 10000, "transaksi_nilai": 9000, "transaksi_dibayar": 10000, "transaksi_dibayar_return": 1000, "diskon_nilai": 1000, "tambahan_nilai": 0, "customers_id": 1, "point_transaksi": 0, "diskon_log": "", "print_oleh_nama": "", "kontainer_size": "", "ekspedisi_status": "", } with patch( "pypos.modules.sinkronisasi.services.transaction_export_service.read_config", return_value={ "export_default_print_oleh_nama": "ops-default", "export_default_kontainer_size": "size-default", "export_default_ekspedisi_status": "ship-default", }, ): payload = service._build_legacy_transaksi_rows( transaksi_rows=[transaksi_row], batch_end="2026-03-06 10:00:00", ) self.assertEqual(len(payload), 1) row = payload[0] self.assertEqual(row.get("print_oleh_nama"), "kasir glg") self.assertEqual(row.get("kontainer_size"), "size-default") self.assertEqual(row.get("ekspedisi_status"), "ship-default") self.assertIn("diskon_rp", row) self.assertIn("diskon_persen", row) self.assertIn("diskon_tambahan_persen", row) self.assertIn("diskon_tambahan_nilai", row) self.assertIn("diskon_produk", row) self.assertIn("diskon_member", row) def test_legacy_transaksi_payload_uses_explicit_values_when_present(self): service = self._build_service() transaksi_row = { "id": 11, "dtime": "2026-03-06 11:00:00", "fulldate": "2026-03-06", "nomer": "INV-11", "jenis_label": "invoice", "oleh_id": 183, "cabang_id": 101, "gudang_id": -1010, "machine_id": "MACHINE-1", "transaksi_bulat": 20000, "transaksi_nilai": 18000, "transaksi_dibayar": 20000, "transaksi_dibayar_return": 2000, "diskon_nilai": 2000, "tambahan_nilai": 0, "customers_id": 1, "point_transaksi": 0, "diskon_log": "", "print_oleh_nama": "operator-a", "kontainer_size": "L", "ekspedisi_status": "delivered", } with patch( "pypos.modules.sinkronisasi.services.transaction_export_service.read_config", return_value={ "export_default_print_oleh_nama": "ops-default", "export_default_kontainer_size": "size-default", "export_default_ekspedisi_status": "ship-default", }, ): payload = service._build_legacy_transaksi_rows( transaksi_rows=[transaksi_row], batch_end="2026-03-06 11:00:00", ) self.assertEqual(len(payload), 1) row = payload[0] self.assertEqual(row.get("print_oleh_nama"), "operator-a") self.assertEqual(row.get("kontainer_size"), "L") self.assertEqual(row.get("ekspedisi_status"), "delivered") def test_legacy_transaksi_payload_resolves_gudang_from_per_cabang_when_zero(self): service = self._build_service() with tempfile.TemporaryDirectory() as td: db_path = f"{td}/legacy_export.db" conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ CREATE TABLE per_cabang ( id INTEGER PRIMARY KEY, nama TEXT, gudang_id INTEGER, gudang_nama TEXT, status INTEGER DEFAULT 1, trash INTEGER DEFAULT 0 ) """ ) cur.execute( """ INSERT INTO per_cabang (id, nama, gudang_id, gudang_nama, status, trash) VALUES (101, 'Cabang A', -1010, 'Gudang Cabang A', 1, 0) """ ) conn.commit() conn.close() service.db_path = db_path transaksi_row = { "id": 12, "dtime": "2026-03-06 12:00:00", "fulldate": "2026-03-06", "nomer": "INV-12", "jenis_label": "invoice", "oleh_id": 183, "oleh_nama": "kasir glg", "cabang_id": 101, "gudang_id": 0, "machine_id": "MACHINE-1", "transaksi_bulat": 20000, "transaksi_nilai": 18000, "transaksi_dibayar": 20000, "transaksi_dibayar_return": 2000, "diskon_nilai": 2000, "tambahan_nilai": 0, "customers_id": 1, "point_transaksi": 0, "diskon_log": "", "print_oleh_nama": "", "kontainer_size": "", "ekspedisi_status": "", } with patch( "pypos.modules.sinkronisasi.services.transaction_export_service.read_config", return_value={}, ): payload = service._build_legacy_transaksi_rows( transaksi_rows=[transaksi_row], batch_end="2026-03-06 12:00:00", ) self.assertEqual(len(payload), 1) row = payload[0] self.assertEqual(int(row.get("gudangID") or 0), -1010) def test_legacy_payment_payload_includes_explicit_discount_components(self): service = TransactionExportService() payment_payload = service._build_legacy_payment_payload( transaksi_row={ "nomer": "INV-99", "transaksi_bulat": 100000, "transaksi_nilai": 90000, "transaksi_dibayar": 90000, "transaksi_dibayar_return": 0, "diskon_persen": 10, "diskon_nilai": 12000, "tambahan_nilai": 10000, "diskon_log": "diskon_customer=2000;cashback=0;point=0", "bank_id": 101, "bank_nama": "Tunai", "bank_rekening_id": 108, "bank_rekening_nama": "Tunai", "oleh_id": 188, "gudang_id": -1010, "dtime": "2026-03-26 12:00:00", "returned": 0, "returns": 0, "_diskon_produk_total": 15000, }, id_penjualan=99, cabang_id=101, ) self.assertEqual(len(payment_payload), 1) row = payment_payload[0] self.assertEqual(int(row.get("diskon_rp") or 0), 10000) self.assertEqual(float(row.get("diskon_persen") or 0), 10.0) self.assertEqual(int(row.get("diskon_tambahan_nilai") or 0), 10000) self.assertEqual(float(row.get("diskon_tambahan_persen") or 0), 10.0) self.assertEqual(int(row.get("diskon_produk") or 0), 15000) self.assertEqual(int(row.get("diskon_member") or 0), 2000) self.assertEqual(int(row.get("total_diskon_produk") or 0), 15000) self.assertEqual(int(row.get("total_diskon_member") or 0), 2000) self.assertEqual(int(row.get("total_diskon_tambahan") or 0), 10000) self.assertEqual(int(row.get("nilai_kurang_bayar") or 0), 0) def test_discount_components_prevent_duplicate_product_vs_additional(self): # edited by glg # Reproduksi kasus: diskon_total master overstated, padahal additional diskon = 0. service = TransactionExportService() transaksi_row = { "nomer": "INV-DSC-1", "transaksi_bulat": 308600, "transaksi_nilai": 302870, "transaksi_dibayar": 302870, "transaksi_dibayar_return": 0, "diskon_nilai": 11460, "tambahan_nilai": 0, "add_disc": 0, "diskon_log": "diskon_customer=0;cashback=0;point=0", "_diskon_produk_total": 5730, "_diskon_tambahan_total": 0, "bank_nama": "Tunai", "bank_rekening_id": 108, "oleh_id": 183, "gudang_id": -1010, "dtime": "2026-03-28 13:02:00", } components = service._resolve_transaksi_discount_components( transaksi_row, diskon_produk_hint=5730, bruto_hint=308600, tagihan_hint=302870, ) self.assertEqual(int(components.get("diskon_produk") or 0), 5730) self.assertEqual(int(components.get("diskon_tambahan") or 0), 0) payment_payload = service._build_legacy_payment_payload( transaksi_row=transaksi_row, id_penjualan=1, cabang_id=101, ) self.assertEqual(len(payment_payload), 1) row = payment_payload[0] self.assertEqual(int(row.get("diskon_produk") or 0), 5730) self.assertEqual(int(row.get("diskon_tambahan_nilai") or 0), 0) def test_legacy_items_payload_prioritizes_nominal_line_discount(self): # edited by glg # Nominal line discount harus diprioritaskan dibanding persen agar sinkron struk/export. service = TransactionExportService() items = service._build_legacy_items_payload( detail_rows=[ { "transaksi_id": 1, "produk_id": 11, "produk_nama": "Produk A", "produk_jenis": "item", "produk_ord_hrg": 10000, "produk_ord_jml": 3, "produk_ord_diskon_persen": 10, "produk_ord_diskon_khusus": 5000, "satuan": "pcs", } ], produk_map={}, ) self.assertIn("11", items) row = items["11"] self.assertEqual(int(row.get("subtotal") or 0), 30000) self.assertEqual(int(row.get("discNilai") or 0), 5000) def test_legacy_transaksi_payload_appends_return_rows(self): # edited by glg service = self._build_service() transaksi_row = { "id": 21, "dtime": "2026-03-06 10:00:00", "fulldate": "2026-03-06", "nomer": "INV-21", "jenis_label": "invoice", "oleh_id": 183, "oleh_nama": "kasir glg", "cabang_id": 101, "gudang_id": -1010, "machine_id": "MACHINE-1", "transaksi_bulat": 10000, "transaksi_nilai": 9000, "transaksi_dibayar": 10000, "transaksi_dibayar_return": 1000, "diskon_nilai": 1000, "tambahan_nilai": 0, "customers_id": 1, "point_transaksi": 0, "diskon_log": "", "print_oleh_nama": "", "kontainer_size": "", "ekspedisi_status": "", } return_row = { "id_penjualan": 9001, "jenis_label": "return", "nomer": "RET-9001-INV-21", "dtime": "2026-03-06 10:05:00", "referenceID": 21, "return_transaksi_id": 21, "return_transaksi_nomer": "INV-21", "items": {}, "payment": [], } payload = service._build_legacy_transaksi_rows( transaksi_rows=[transaksi_row], batch_end="2026-03-06 10:10:00", extra_return_rows=[return_row], ) self.assertEqual(len(payload), 2) self.assertEqual(str(payload[0].get("nomer")), "INV-21") self.assertEqual(str(payload[1].get("nomer")), "RET-9001-INV-21") self.assertEqual(int(payload[1].get("return_transaksi_id") or 0), 21) def test_build_legacy_return_rows_payload_contains_reference_fields(self): # edited by glg service = TransactionExportService() service._get_export_device_context = lambda machine_id_hint="": { "machine_id": "MACHINE-1", "cabang_id": 101, "cabang_nama": "Cabang A", } service._fetch_detail_return_rows_for_return_ids = lambda return_ids: [ { "id": 1, "return_id": 9001, "produk_id": 201, "produk_nama": "Produk A", "jumlah": 2, "harga": 5000, "subtotal": 10000, } ] service._fetch_produk_snapshot = lambda produk_ids: { "201": { "id": 201, "nama": "Produk A", "satuan": "pcs", "satuan_id": 1, "tipe_pajak": 1, "ppn": 0, "ppn_produk": 1, "barcode": "P201", "coa_code": "4-201", "hpp": 3000, } } rows = service._build_legacy_return_rows_payload( return_rows=[ { "id": 9001, "transaksi_id": 21, "tanggal_return": "2026-03-06 10:05:00", "refund_amount": 10000, "total_return": 10000, "refund_method": "cash", "jenis_return": "partial", "keterangan": "uji return", "kode_voucher": "RET-CODE-1", "transaksi_nomer": "INV-21", "transaksi_cabang_id": 101, "transaksi_cabang_nama": "Cabang A", "transaksi_gudang_id": -1010, "transaksi_oleh_id": 183, "transaksi_oleh_nama": "kasir glg", "transaksi_machine_id": "MACHINE-1", "transaksi_cpu_info": "cpu", "transaksi_com_info": "pc", } ], batch_end="2026-03-06 10:10:00", ) self.assertEqual(len(rows), 1) row = rows[0] self.assertEqual(str(row.get("jenis_label")), "return") self.assertEqual(str(row.get("jenis_kode")), "982") self.assertEqual(int(row.get("penjualan_return") or 0), 10000) self.assertEqual(int(row.get("reference_id") or 0), 21) self.assertEqual(str(row.get("reference_nomer")), "INV-21") self.assertEqual(int(row.get("referenceID") or 0), 21) self.assertEqual(str(row.get("referenceNomer")), "INV-21") self.assertEqual(int(row.get("return_transaksi_id") or 0), 21) self.assertEqual(str(row.get("return_jenis")), "partial") produk_return_list = row.get("produk_return_list") or [] self.assertEqual(len(produk_return_list), 1) self.assertEqual(str(produk_return_list[0].get("produk_id")), "201") self.assertEqual(str(produk_return_list[0].get("produk_nama")), "Produk A") def test_export_transaksi_integration_mixed_sale_return_and_cancellation(self): # edited by glg # Integrasi campuran: jual tunai + non-tunai + return parsial + pembatalan full nota. service = TransactionExportService() with tempfile.TemporaryDirectory() as td: db_path = f"{td}/mixed_export.db" conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ CREATE TABLE transaksi ( id INTEGER PRIMARY KEY, nomer TEXT, jenis_label TEXT, dtime TEXT, fulldate TEXT, oleh_id INTEGER, oleh_nama TEXT, cabang_id INTEGER, cabang_nama TEXT, gudang_id INTEGER, machine_id TEXT, cpu_info TEXT, com_info TEXT, transaksi_bulat REAL, transaksi_nilai REAL, transaksi_dibayar REAL, transaksi_dibayar_return REAL, diskon_nilai REAL, tambahan_nilai REAL, customers_id INTEGER, point_transaksi INTEGER, diskon_log TEXT, returned INTEGER, returns INTEGER, bank_id INTEGER, bank_nama TEXT, bank_rekening_id INTEGER, bank_rekening_nama TEXT, pembayaran_sys TEXT, pembayaran_tunai REAL, pembayaran_non_tunai REAL, trash INTEGER, status INTEGER, cancel_dtime TEXT, reference_id INTEGER, reference_nomer TEXT, reference_jenis INTEGER ) """ ) cur.execute( """ CREATE TABLE transaksi_data ( id INTEGER PRIMARY KEY, transaksi_id INTEGER, produk_id INTEGER, produk_nama TEXT, valid_qty INTEGER, produk_ord_jml INTEGER, produk_ord_hrg REAL ) """ ) cur.execute( """ CREATE TABLE return_transaksi_penjualan ( id INTEGER PRIMARY KEY, transaksi_id TEXT, tanggal_return TEXT, total_return REAL, kode_voucher TEXT, nilai_voucher REAL, customer_id TEXT, keterangan TEXT, jenis_return TEXT, refund_method TEXT, refund_amount REAL ) """ ) cur.execute( """ CREATE TABLE detail_return_transaksi_penjualan ( id INTEGER PRIMARY KEY, return_id INTEGER, produk_id TEXT, produk_nama TEXT, jumlah INTEGER, jenis_return TEXT, harga REAL, subtotal REAL ) """ ) cur.execute( """ CREATE TABLE pembatalan_transaksi_history ( id INTEGER PRIMARY KEY, transaksi_id TEXT, nomer TEXT, transaksi_dtime TEXT, customers_nama TEXT, kasir_nama TEXT, transaksi_nilai REAL, admin_verifikasi TEXT, dibatalkan_oleh_id TEXT, dibatalkan_oleh_nama TEXT, cancel_dtime TEXT ) """ ) cur.execute( """ INSERT INTO transaksi ( id, nomer, jenis_label, dtime, fulldate, oleh_id, oleh_nama, cabang_id, cabang_nama, gudang_id, machine_id, cpu_info, com_info, transaksi_bulat, transaksi_nilai, transaksi_dibayar, transaksi_dibayar_return, diskon_nilai, tambahan_nilai, customers_id, point_transaksi, diskon_log, returned, returns, bank_id, bank_nama, bank_rekening_id, bank_rekening_nama, pembayaran_sys, pembayaran_tunai, pembayaran_non_tunai, trash, status, cancel_dtime, reference_id, reference_nomer, reference_jenis ) VALUES ( 1001, 'INV-1001', 'invoice', '2026-04-04 09:00:00', '2026-04-04', 183, 'kasir', 101, 'BPJ', -1010, 'M1', 'cpu', 'pc', 50000, 50000, 50000, 0, 0, 0, 1, 0, '', 0, 0, 108, 'Tunai', 108, 'Tunai', 'tunai', 50000, 0, 0, 1, '', 0, '', 0 ) """ ) cur.execute( """ INSERT INTO transaksi ( id, nomer, jenis_label, dtime, fulldate, oleh_id, oleh_nama, cabang_id, cabang_nama, gudang_id, machine_id, cpu_info, com_info, transaksi_bulat, transaksi_nilai, transaksi_dibayar, transaksi_dibayar_return, diskon_nilai, tambahan_nilai, customers_id, point_transaksi, diskon_log, returned, returns, bank_id, bank_nama, bank_rekening_id, bank_rekening_nama, pembayaran_sys, pembayaran_tunai, pembayaran_non_tunai, trash, status, cancel_dtime, reference_id, reference_nomer, reference_jenis ) VALUES ( 1002, 'INV-1002', 'invoice', '2026-04-04 09:10:00', '2026-04-04', 183, 'kasir', 101, 'BPJ', -1010, 'M1', 'cpu', 'pc', 70000, 70000, 70000, 0, 0, 0, 1, 0, '', 0, 0, 109, 'Debit', 109, 'Debit', 'non_tunai', 0, 70000, 1, 1, '2026-04-04 10:20:00', 0, '', 0 ) """ ) cur.execute( """ INSERT INTO transaksi_data ( id, transaksi_id, produk_id, produk_nama, valid_qty, produk_ord_jml, produk_ord_hrg ) VALUES ( 1, 1001, 10, 'Produk A', 2, 2, 25000 ) """ ) cur.execute( """ INSERT INTO transaksi_data ( id, transaksi_id, produk_id, produk_nama, valid_qty, produk_ord_jml, produk_ord_hrg ) VALUES ( 2, 1002, 20, 'Produk B', 2, 2, 35000 ) """ ) cur.execute( """ INSERT INTO return_transaksi_penjualan ( id, transaksi_id, tanggal_return, total_return, kode_voucher, nilai_voucher, customer_id, keterangan, jenis_return, refund_method, refund_amount ) VALUES ( 5001, '1001', '2026-04-04 10:00:00', 15000, 'RET-5001', 0, '1', 'return parsial', 'partial', 'cash', 15000 ) """ ) cur.execute( """ INSERT INTO detail_return_transaksi_penjualan ( id, return_id, produk_id, produk_nama, jumlah, jenis_return, harga, subtotal ) VALUES ( 1, 5001, '10', 'Produk A', 1, 'partial', 15000, 15000 ) """ ) cur.execute( """ INSERT INTO pembatalan_transaksi_history ( id, transaksi_id, nomer, transaksi_dtime, customers_nama, kasir_nama, transaksi_nilai, admin_verifikasi, dibatalkan_oleh_id, dibatalkan_oleh_nama, cancel_dtime ) VALUES ( 6001, '1002', 'INV-1002', '2026-04-04 09:10:00', 'Cust B', 'kasir', 70000, 'admin', '182', 'admin', '2026-04-04 10:20:00' ) """ ) conn.commit() conn.close() service.db_path = db_path service._get_export_device_context = lambda machine_id_hint="": { "machine_id": "M1", "cabang_id": 101, "cabang_nama": "BPJ", } conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute("SELECT * FROM transaksi ORDER BY id ASC") transaksi_rows = [dict(row) for row in cur.fetchall()] conn.close() filtered_main_rows = service._filter_transaksi_invoice_rows( transaksi_rows, source="main_batch", ) # edited by glg # Invoice pembatalan tetap harus ikut payload transaksi # (dengan marker cancel), selain event return/cancellation. self.assertEqual( {int(row.get("id") or 0) for row in filtered_main_rows}, {1001, 1002}, ) return_updates = service._fetch_return_rows_for_transaksi_export( start_batch_end="2026-04-04 00:00:00", end_batch_end="2026-04-04 23:59:59", limit=100, ) self.assertEqual(len(return_updates), 2) self.assertEqual( {str((row or {}).get("event_source") or "") for row in return_updates}, {"return", "cancellation"}, ) extra_rows = service._build_legacy_return_rows_payload( return_rows=return_updates, batch_end="2026-04-04 23:59:59", ) payload = service._build_legacy_transaksi_rows( transaksi_rows=filtered_main_rows, batch_end="2026-04-04 23:59:59", extra_return_rows=extra_rows, ) # 2 transaksi jual (termasuk invoice batal) + 2 event return/pembatalan self.assertEqual(len(payload), 4) sale_rows = [row for row in payload if str(row.get("jenis_label")) == "invoice"] self.assertEqual(len(sale_rows), 2) self.assertEqual( {int(row.get("id_penjualan") or 0) for row in sale_rows}, {1001, 1002}, ) cancel_sale_row = next( row for row in sale_rows if int(row.get("id_penjualan") or 0) == 1002 ) self.assertEqual(str(cancel_sale_row.get("cancel_dtime") or ""), "2026-04-04 10:20:00") self.assertEqual(int(cancel_sale_row.get("is_cancelled") or 0), 1) self.assertEqual(int(cancel_sale_row.get("status_cancel") or 0), 1) self.assertEqual(int(cancel_sale_row.get("trash") or 0), 1) return_rows = [row for row in payload if str(row.get("jenis_label")) == "return"] self.assertEqual(len(return_rows), 2) for row in return_rows: self.assertEqual(str(row.get("jenis_kode")), "982") self.assertIn("referenceID", row) self.assertIn("referenceNomer", row) self.assertIn("reference_id", row) self.assertIn("reference_nomer", row) self.assertEqual(int(row.get("return_penjualan") or 0), 1) self.assertGreater(int(row.get("penjualan_return") or 0), 0) produk_list = row.get("produk_return_list") or [] self.assertGreaterEqual(len(produk_list), 1) for produk in produk_list: self.assertTrue(str(produk.get("produk_id") or "").strip()) self.assertTrue(str(produk.get("produk_nama") or "").strip()) partial_row = next( row for row in return_rows if int(row.get("reference_id") or 0) == 1001 ) cancel_row = next( row for row in return_rows if int(row.get("reference_id") or 0) == 1002 ) self.assertEqual(int(partial_row.get("penjualan_return") or 0), 15000) self.assertEqual(int(cancel_row.get("penjualan_return") or 0), 70000) self.assertEqual(str(cancel_row.get("event_source")), "cancellation") self.assertEqual(str((partial_row.get("produk_return_list") or [{}])[0].get("produk_id")), "10") self.assertEqual(str((partial_row.get("produk_return_list") or [{}])[0].get("produk_nama")), "Produk A") self.assertEqual(str((cancel_row.get("produk_return_list") or [{}])[0].get("produk_id")), "20") self.assertEqual(str((cancel_row.get("produk_return_list") or [{}])[0].get("produk_nama")), "Produk B") def test_scope_cancel_updates_hanya_untuk_invoice_main_batch(self): # edited by glg # Jika invoice sudah pernah diexport (tidak ada di main batch), # cancel update tidak boleh memicu kirim ulang invoice. service = TransactionExportService() scoped_empty_main = service._scope_cancel_updates_to_current_main_batch( main_rows=[], cancel_rows=[ {"id": 1002, "jenis_label": "invoice", "cancel_dtime": "2026-04-04 15:00:00"}, ], ) self.assertEqual(scoped_empty_main, []) scoped_mixed = service._scope_cancel_updates_to_current_main_batch( main_rows=[ {"id": 1002, "jenis_label": "invoice", "cancel_dtime": "2026-04-04 15:00:00"}, ], cancel_rows=[ {"id": 1002, "jenis_label": "invoice", "cancel_dtime": "2026-04-04 15:00:00"}, {"id": 999, "jenis_label": "invoice", "cancel_dtime": "2026-04-04 15:01:00"}, ], ) self.assertEqual(len(scoped_mixed), 1) self.assertEqual(int(scoped_mixed[0].get("id") or 0), 1002) def test_cancellation_only_tetap_jadi_982_saat_kolom_opsional_transaksi_tidak_ada(self): # edited by glg service = TransactionExportService() with tempfile.TemporaryDirectory() as td: db_path = f"{td}/cancel_schema_legacy.db" conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ CREATE TABLE transaksi ( id INTEGER PRIMARY KEY, nomer TEXT, cabang_id INTEGER, cabang_nama TEXT, gudang_id INTEGER, oleh_id INTEGER, oleh_nama TEXT ) """ ) cur.execute( """ CREATE TABLE pembatalan_transaksi_history ( id INTEGER PRIMARY KEY, transaksi_id TEXT, nomer TEXT, transaksi_dtime TEXT, kasir_nama TEXT, transaksi_nilai REAL, cancel_dtime TEXT ) """ ) cur.execute( """ INSERT INTO transaksi ( id, nomer, cabang_id, cabang_nama, gudang_id, oleh_id, oleh_nama ) VALUES ( 2001, 'INV-2001', 101, 'Cabang A', -1010, 183, 'kasir' ) """ ) cur.execute( """ INSERT INTO pembatalan_transaksi_history ( id, transaksi_id, nomer, transaksi_dtime, kasir_nama, transaksi_nilai, cancel_dtime ) VALUES ( 7001, '2001', 'INV-2001', '2026-04-05 09:00:00', 'kasir', 45000, '2026-04-05 09:30:00' ) """ ) conn.commit() conn.close() service.db_path = db_path service._get_export_device_context = lambda machine_id_hint="": { "machine_id": "M-FALLBACK", "cabang_id": 101, "cabang_nama": "Cabang A", } service._fetch_transaksi_data_by_transaksi_ids = lambda transaksi_ids, limit=0: [] return_updates = service._fetch_return_rows_for_transaksi_export( start_batch_end="2026-04-05 00:00:00", end_batch_end="2026-04-05 23:59:59", limit=100, ) self.assertEqual(len(return_updates), 1) self.assertEqual(str(return_updates[0].get("event_source")), "cancellation") self.assertEqual(str(return_updates[0].get("transaksi_machine_id") or ""), "") self.assertEqual(str(return_updates[0].get("transaksi_cpu_info") or ""), "") self.assertEqual(str(return_updates[0].get("transaksi_com_info") or ""), "") payload_rows = service._build_legacy_return_rows_payload( return_rows=return_updates, batch_end="2026-04-05 23:59:59", ) self.assertEqual(len(payload_rows), 1) payload_row = payload_rows[0] self.assertEqual(str(payload_row.get("event_source")), "cancellation") self.assertEqual(str(payload_row.get("jenis_kode")), "982") self.assertEqual(str(payload_row.get("machineID")), "M-FALLBACK") self.assertEqual(int(payload_row.get("penjualan_return") or 0), 45000) def test_legacy_transaksi_rows_mendukung_return_only_tanpa_invoice_reexport(self): # edited by glg # Batch pembatalan untuk invoice lama: payload transaksi cukup kirim # baris return/cancellation tanpa kirim ulang invoice. service = self._build_service() return_only_rows = service._build_legacy_transaksi_rows( transaksi_rows=[], batch_end="2026-04-04 23:59:59", extra_return_rows=[ { "id_penjualan": 6001, "jenis_label": "return", "jenis_kode": "982", "event_source": "cancellation", "dtime": "2026-04-04 15:00:00", "reference_id": 1002, "reference_nomer": "INV-1002", } ], ) self.assertEqual(len(return_only_rows), 1) row = return_only_rows[0] self.assertEqual(str(row.get("jenis_label")), "return") self.assertEqual(str(row.get("event_source")), "cancellation") self.assertEqual(int(row.get("reference_id") or 0), 1002) def test_legacy_items_payload_does_not_merge_free_qty_with_sale_qty_same_product(self): # edited by glg service = TransactionExportService() items = service._build_legacy_items_payload( detail_rows=[ { "transaksi_id": 1, "produk_id": 11702, "produk_nama": "Produk X", "produk_jenis": "item", "produk_ord_hrg": 1950, "produk_ord_jml": 12, "valid_qty": 12, "produk_ord_diskon_persen": 5.128205, "satuan": "pcs", }, { "transaksi_id": 1, "produk_id": 11702, "produk_nama": "[FREE 6 x Produk X]", "produk_jenis": "free_produk", "produk_ord_hrg": 0, "produk_ord_jml": 6, "valid_qty": 6, "satuan": "pcs", }, ], produk_map={ "11702": { "id": 11702, "nama": "Produk X", "satuan": "pcs", "satuan_id": 16, "tipe_pajak": 1, "ppn": 0, "ppn_produk": 1, "barcode": "8997204306774", "coa_code": "4-11702", "hpp": 1200, } }, ) self.assertIn("11702", items) row = items["11702"] # qty jual harus tetap 12, tidak boleh jadi 18 self.assertEqual(int(row.get("jumlah") or 0), 12) self.assertEqual(int(row.get("subtotal") or 0), 23400) def test_legacy_items_payload_keeps_free_b_outside_items_for_buy_a_free_b(self): # edited by glg service = TransactionExportService() details = [ { "id": 10, "transaksi_id": 1, "produk_id": 11702, "produk_nama": "Produk A", "produk_jenis": "item", "produk_ord_hrg": 10000, "produk_ord_jml": 2, "valid_qty": 2, "satuan": "pcs", }, { "id": 11, "transaksi_id": 1, "produk_id": 22001, "produk_nama": "Produk B Bonus", "produk_jenis": "free_produk", "produk_ord_hrg": 0, "produk_ord_jml": 1, "valid_qty": 1, "satuan": "pcs", "ext_intext": "{\"free_relasi\":{\"source_produk_id\":11702,\"source_produk_nama\":\"Produk A\",\"source_qty\":2,\"free_produk_id\":22001,\"free_produk_nama\":\"Produk B Bonus\",\"free_qty\":1}}", }, ] produk_map = { "11702": {"id": 11702, "nama": "Produk A", "satuan": "pcs", "satuan_id": 1, "tipe_pajak": 1, "ppn": 0, "ppn_produk": 1, "barcode": "PA", "coa_code": "4-A", "hpp": 5000}, "22001": {"id": 22001, "nama": "Produk B Bonus", "satuan": "pcs", "satuan_id": 1, "tipe_pajak": 1, "ppn": 0, "ppn_produk": 1, "barcode": "PB", "coa_code": "4-B", "hpp": 1000}, } diskon_map = { "11702": { "id": 9, "produk_id": 11702, "free_produk_id": 22001, "free_produk_nama": "Produk B Bonus", "kelipatan": 2, "minim": 2, "quota_global": 0, "quota_used": 0, } } items_payload = service._build_legacy_items_payload(details, produk_map) free_items_payload = service._build_legacy_free_items_payload( details, produk_map, diskon_map, "2026-04-01 10:00:00", ) # items hanya produk A (jual), produk B free tidak boleh muncul self.assertIn("11702", items_payload) self.assertNotIn("22001", items_payload) self.assertIn("11702", free_items_payload) if __name__ == "__main__": unittest.main()