import os import sqlite3 import tempfile import unittest from pypos.modules.sinkronisasi.models.export_upload_model import ExportUploadModel from pypos.modules.sinkronisasi.services.export_upload_service import ExportUploadService class _UploadApiStub: def __init__( self, ok=True, retryable=True, error="upload_failed", compile_ok=True, compile_payload=None, compile_retryable=True, compile_error="compile_status_pending", ): self.ok = bool(ok) self.retryable = bool(retryable) self.error = str(error) self.compile_ok = bool(compile_ok) self.compile_payload = compile_payload if isinstance(compile_payload, dict) else {} self.compile_retryable = bool(compile_retryable) self.compile_error = str(compile_error) self.calls = 0 self.last_metadata = None self.compile_calls = 0 def upload_file(self, file_path, metadata, timeout): _ = (file_path, metadata, timeout) self.calls += 1 self.last_metadata = dict(metadata or {}) if self.ok: return { "ok": True, "error": "", "retryable": False, "status_code": 200, "response_text": '{"status":1,"idempotency_key":"IDEMPOTENCY-1"}', } return { "ok": False, "error": self.error, "retryable": self.retryable, "status_code": 500 if self.retryable else 400, "response_text": "", } def fetch_compile_status(self, idempotency_key, timeout): _ = (idempotency_key, timeout) self.compile_calls += 1 if self.compile_ok: return { "ok": True, "error": "", "retryable": False, "status_code": 200, "response_text": '{"status":1}', "response_json": dict(self.compile_payload), } return { "ok": False, "error": self.compile_error, "retryable": self.compile_retryable, "status_code": 500 if self.compile_retryable else 400, "response_text": "", "response_json": {}, } class ExportUploadServiceTests(unittest.TestCase): def _prepare_db(self, db_path): conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ CREATE TABLE export_flux ( id INTEGER PRIMARY KEY AUTOINCREMENT, batch_start TEXT NOT NULL, batch_end TEXT NOT NULL, table_name TEXT NULL, server_hash TEXT NULL, file_seq INTEGER NULL, file_hash TEXT NULL, file_size INTEGER NULL, row_count INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL DEFAULT 'PENDING', file_path TEXT NULL, error_log TEXT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """ ) conn.commit() conn.close() def _insert_flux_success(self, db_path, file_path, table_name="transaksi"): conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ INSERT INTO export_flux ( batch_start, batch_end, table_name, server_hash, file_seq, file_hash, file_size, row_count, status, file_path, error_log, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'), datetime('now')) """, ( "0-10", "2026-02-18 10:00:00", str(table_name or "transaksi"), "serverhash", 1, "abc123", 123, 10, "SUCCESS", file_path, "", ), ) conn.commit() conn.close() def test_direct_only_suppresses_settlement_backlog_rows(self): # edited by glg with tempfile.TemporaryDirectory() as tmpdir: db_path = os.path.join(tmpdir, "upload_direct_only.db") settlement_file = os.path.join(tmpdir, "settlement.gz") transaksi_file = os.path.join(tmpdir, "transaksi.gz") with open(settlement_file, "wb") as fh: fh.write(b"settlement-data") with open(transaksi_file, "wb") as fh: fh.write(b"transaksi-data") self._prepare_db(db_path) self._insert_flux_success(db_path, settlement_file, table_name="settlement_history") self._insert_flux_success(db_path, transaksi_file, table_name="transaksi") model = ExportUploadModel(db_path=db_path) api = _UploadApiStub(ok=True) service = ExportUploadService( model=model, api_service=api, enabled_getter=lambda: True, batch_limit_getter=lambda: 10, timeout_getter=lambda: 10, retention_days_getter=lambda: 0, compile_check_enabled_getter=lambda: False, settlement_direct_only_mode_getter=lambda: True, ) result = service.process_pending_uploads() self.assertEqual(int(result.get("suppressed_direct_only") or 0), 1) self.assertEqual(result["uploaded"], 1) self.assertEqual(api.calls, 1) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("SELECT table_name, upload_status, COALESCE(upload_response, '') FROM export_flux ORDER BY id ASC") rows = cur.fetchall() conn.close() self.assertEqual(str(rows[0][0]), "settlement_history") self.assertEqual(str(rows[0][1]), "UPLOADED") self.assertIn("skipped_by_settlement_direct_only_mode", str(rows[0][2])) self.assertEqual(str(rows[1][0]), "transaksi") self.assertEqual(str(rows[1][1]), "UPLOADED") def test_table_toggle_suppresses_disabled_backlog_rows(self): # edited by glg with tempfile.TemporaryDirectory() as tmpdir: db_path = os.path.join(tmpdir, "upload_toggle_tables.db") detail_file = os.path.join(tmpdir, "transaksi_data.gz") header_file = os.path.join(tmpdir, "transaksi.gz") with open(detail_file, "wb") as fh: fh.write(b"detail-data") with open(header_file, "wb") as fh: fh.write(b"header-data") self._prepare_db(db_path) self._insert_flux_success(db_path, detail_file, table_name="transaksi_data") self._insert_flux_success(db_path, header_file, table_name="transaksi") model = ExportUploadModel(db_path=db_path) api = _UploadApiStub(ok=True) service = ExportUploadService( model=model, api_service=api, enabled_getter=lambda: True, batch_limit_getter=lambda: 10, timeout_getter=lambda: 10, retention_days_getter=lambda: 0, compile_check_enabled_getter=lambda: False, settlement_direct_only_mode_getter=lambda: False, allowed_export_tables_getter=lambda: ["transaksi"], ) result = service.process_pending_uploads() self.assertEqual(int(result.get("suppressed_table_toggle") or 0), 1) self.assertEqual(result["uploaded"], 1) self.assertEqual(api.calls, 1) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("SELECT table_name, upload_status, COALESCE(upload_response, '') FROM export_flux ORDER BY id ASC") rows = cur.fetchall() conn.close() self.assertEqual(str(rows[0][0]), "transaksi_data") self.assertEqual(str(rows[0][1]), "UPLOADED") self.assertIn("skipped_by_export_table_toggle", str(rows[0][2])) self.assertEqual(str(rows[1][0]), "transaksi") self.assertEqual(str(rows[1][1]), "UPLOADED") def test_upload_success_marks_uploaded(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = os.path.join(tmpdir, "upload.db") file_path = os.path.join(tmpdir, "a.gz") with open(file_path, "wb") as fh: fh.write(b"gzip-data") self._prepare_db(db_path) self._insert_flux_success(db_path, file_path) model = ExportUploadModel(db_path=db_path) api = _UploadApiStub(ok=True) service = ExportUploadService( model=model, api_service=api, enabled_getter=lambda: True, batch_limit_getter=lambda: 10, timeout_getter=lambda: 10, retention_days_getter=lambda: 0, compile_check_enabled_getter=lambda: False, allowed_export_tables_getter=lambda: ["transaksi", "transaksi_data"], ) result = service.process_pending_uploads() self.assertEqual(result["uploaded"], 1) self.assertEqual(api.calls, 1) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("SELECT upload_status, upload_attempt_count FROM export_flux WHERE id = 1") row = cur.fetchone() conn.close() self.assertEqual(row[0], "UPLOADED") self.assertIn(row[1], (0, None)) def test_upload_retry_marks_retry_state(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = os.path.join(tmpdir, "upload_retry.db") file_path = os.path.join(tmpdir, "b.gz") with open(file_path, "wb") as fh: fh.write(b"gzip-data") self._prepare_db(db_path) self._insert_flux_success(db_path, file_path) model = ExportUploadModel(db_path=db_path) api = _UploadApiStub(ok=False, retryable=True, error="timeout") service = ExportUploadService( model=model, api_service=api, enabled_getter=lambda: True, batch_limit_getter=lambda: 10, timeout_getter=lambda: 10, max_attempt_getter=lambda: 5, backoff_base_getter=lambda: 1, backoff_factor_getter=lambda: 2, backoff_max_getter=lambda: 30, retention_days_getter=lambda: 0, compile_check_enabled_getter=lambda: False, ) result = service.process_pending_uploads() self.assertEqual(result["retried"], 1) self.assertEqual(api.calls, 1) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( "SELECT upload_status, upload_attempt_count, upload_next_retry_at FROM export_flux WHERE id = 1" ) row = cur.fetchone() conn.close() self.assertEqual(row[0], "RETRY") self.assertEqual(int(row[1]), 1) self.assertTrue(str(row[2] or "").strip()) def test_upload_non_retryable_marks_failed(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = os.path.join(tmpdir, "upload_failed.db") file_path = os.path.join(tmpdir, "c.gz") with open(file_path, "wb") as fh: fh.write(b"gzip-data") self._prepare_db(db_path) self._insert_flux_success(db_path, file_path) model = ExportUploadModel(db_path=db_path) api = _UploadApiStub(ok=False, retryable=False, error="missing_toko_id") service = ExportUploadService( model=model, api_service=api, enabled_getter=lambda: True, batch_limit_getter=lambda: 10, timeout_getter=lambda: 10, retention_days_getter=lambda: 0, compile_check_enabled_getter=lambda: False, allowed_export_tables_getter=lambda: ["transaksi", "transaksi_data"], ) result = service.process_pending_uploads() self.assertEqual(result["failed"], 1) self.assertEqual(api.calls, 1) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( "SELECT upload_status, upload_attempt_count, upload_next_retry_at FROM export_flux WHERE id = 1" ) row = cur.fetchone() conn.close() self.assertEqual(row[0], "FAILED") self.assertEqual(int(row[1]), 1) self.assertIn(row[2], (None, "")) def test_requeue_failed_transient_rows_on_startup(self): # edited by glg with tempfile.TemporaryDirectory() as tmpdir: db_path = os.path.join(tmpdir, "upload_requeue_transient.db") file_path = os.path.join(tmpdir, "requeue.gz") with open(file_path, "wb") as fh: fh.write(b"gzip-data") self._prepare_db(db_path) self._insert_flux_success(db_path, file_path) model = ExportUploadModel(db_path=db_path) model.mark_failed(1, "timeout while upload") service = ExportUploadService( model=model, api_service=_UploadApiStub(ok=True), enabled_getter=lambda: False, retention_days_getter=lambda: 0, requeue_failed_transient_enabled_getter=lambda: True, requeue_failed_transient_limit_getter=lambda: 10, requeue_failed_transient_max_age_hours_getter=lambda: 72, retryable_reason_getter=lambda: ["timeout", "tempor"], ) result = service.process_pending_uploads(requeue_failed_transient=True) self.assertEqual(int(result.get("requeued_failed_transient") or 0), 1) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( "SELECT upload_status, upload_attempt_count FROM export_flux WHERE id = 1" ) row = cur.fetchone() conn.close() self.assertEqual(str(row[0]), "RETRY") self.assertEqual(int(row[1] or 0), 0) def test_requeue_failed_transient_skips_non_transient_error(self): # edited by glg with tempfile.TemporaryDirectory() as tmpdir: db_path = os.path.join(tmpdir, "upload_requeue_non_transient.db") file_path = os.path.join(tmpdir, "requeue_non_transient.gz") with open(file_path, "wb") as fh: fh.write(b"gzip-data") self._prepare_db(db_path) self._insert_flux_success(db_path, file_path) model = ExportUploadModel(db_path=db_path) model.mark_failed(1, "missing_toko_id") service = ExportUploadService( model=model, api_service=_UploadApiStub(ok=True), enabled_getter=lambda: False, retention_days_getter=lambda: 0, requeue_failed_transient_enabled_getter=lambda: True, requeue_failed_transient_limit_getter=lambda: 10, requeue_failed_transient_max_age_hours_getter=lambda: 72, retryable_reason_getter=lambda: ["timeout", "tempor"], ) result = service.process_pending_uploads(requeue_failed_transient=True) self.assertEqual(int(result.get("requeued_failed_transient") or 0), 0) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("SELECT upload_status FROM export_flux WHERE id = 1") row = cur.fetchone() conn.close() self.assertEqual(str(row[0]), "FAILED") def test_requeue_failed_transient_respects_max_age(self): # edited by glg with tempfile.TemporaryDirectory() as tmpdir: db_path = os.path.join(tmpdir, "upload_requeue_old_failed.db") file_path = os.path.join(tmpdir, "requeue_old.gz") with open(file_path, "wb") as fh: fh.write(b"gzip-data") self._prepare_db(db_path) self._insert_flux_success(db_path, file_path) model = ExportUploadModel(db_path=db_path) model.mark_failed(1, "timeout while upload") conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( "UPDATE export_flux SET upload_updated_at = datetime('now', '-96 hours') WHERE id = 1" ) conn.commit() conn.close() service = ExportUploadService( model=model, api_service=_UploadApiStub(ok=True), enabled_getter=lambda: False, retention_days_getter=lambda: 0, requeue_failed_transient_enabled_getter=lambda: True, requeue_failed_transient_limit_getter=lambda: 10, requeue_failed_transient_max_age_hours_getter=lambda: 72, retryable_reason_getter=lambda: ["timeout", "tempor"], ) result = service.process_pending_uploads(requeue_failed_transient=True) self.assertEqual(int(result.get("requeued_failed_transient") or 0), 0) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("SELECT upload_status FROM export_flux WHERE id = 1") row = cur.fetchone() conn.close() self.assertEqual(str(row[0]), "FAILED") def test_upload_backfills_required_metadata(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = os.path.join(tmpdir, "upload_metadata.db") file_path = os.path.join(tmpdir, "d.gz") with open(file_path, "wb") as fh: fh.write(b"gzip-data-for-hash") self._prepare_db(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ INSERT INTO export_flux ( batch_start, batch_end, table_name, server_hash, file_seq, file_hash, file_size, row_count, status, file_path, error_log, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'), datetime('now')) """, ( "0-5", "2026-02-18 12:00:00", "transaksi_data", "serverhash", None, "", 25, 5, "SUCCESS", file_path, "", ), ) conn.commit() conn.close() model = ExportUploadModel(db_path=db_path) api = _UploadApiStub(ok=True) service = ExportUploadService( model=model, api_service=api, enabled_getter=lambda: True, batch_limit_getter=lambda: 10, timeout_getter=lambda: 10, retention_days_getter=lambda: 0, compile_check_enabled_getter=lambda: False, allowed_export_tables_getter=lambda: ["transaksi", "transaksi_data"], ) result = service.process_pending_uploads() self.assertEqual(result["uploaded"], 1) self.assertEqual(api.calls, 1) self.assertTrue(str(api.last_metadata.get("file_seq") or "").isdigit()) self.assertEqual(len(str(api.last_metadata.get("file_hash") or "")), 64) def test_upload_with_compile_check_marks_uploaded_when_complete(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = os.path.join(tmpdir, "upload_compile_ok.db") file_path = os.path.join(tmpdir, "compile_ok.gz") with open(file_path, "wb") as fh: fh.write(b"gzip-data") self._prepare_db(db_path) self._insert_flux_success(db_path, file_path) model = ExportUploadModel(db_path=db_path) api = _UploadApiStub( ok=True, compile_ok=True, compile_payload={ "status": 1, "compiled_transaksi": 1, "compiled_data": 1, "compiled_registry": 1, }, ) service = ExportUploadService( model=model, api_service=api, enabled_getter=lambda: True, batch_limit_getter=lambda: 10, timeout_getter=lambda: 10, retention_days_getter=lambda: 0, compile_check_enabled_getter=lambda: True, compile_poll_max_attempt_getter=lambda: 1, compile_poll_interval_getter=lambda: 0, compile_required_flags_getter=lambda: [ "compiled_transaksi", "compiled_data", "compiled_registry", ], ) result = service.process_pending_uploads() self.assertEqual(result["uploaded"], 1) self.assertEqual(result["retried"], 0) self.assertEqual(api.compile_calls, 1) def test_upload_with_compile_check_marks_uploaded_when_pending_non_blocking(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = os.path.join(tmpdir, "upload_compile_pending.db") file_path = os.path.join(tmpdir, "compile_pending.gz") with open(file_path, "wb") as fh: fh.write(b"gzip-data") self._prepare_db(db_path) self._insert_flux_success(db_path, file_path) model = ExportUploadModel(db_path=db_path) api = _UploadApiStub( ok=True, compile_ok=True, compile_payload={"status": 1, "compiled_transaksi": 0}, ) service = ExportUploadService( model=model, api_service=api, enabled_getter=lambda: True, batch_limit_getter=lambda: 10, timeout_getter=lambda: 10, max_attempt_getter=lambda: 5, backoff_base_getter=lambda: 1, backoff_factor_getter=lambda: 2, backoff_max_getter=lambda: 30, retention_days_getter=lambda: 0, compile_check_enabled_getter=lambda: True, compile_poll_max_attempt_getter=lambda: 1, compile_poll_interval_getter=lambda: 0, compile_required_flags_getter=lambda: [ "compiled_transaksi", "compiled_data", "compiled_registry", ], ) result = service.process_pending_uploads() # edited by glg # Compile status pending tetap non-blocking: # batas sukses POS adalah upload stream berhasil. self.assertEqual(result["uploaded"], 1) self.assertEqual(result["retried"], 0) self.assertEqual(api.compile_calls, 1) def test_upload_with_compile_check_accepts_per_key_done_state(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = os.path.join(tmpdir, "upload_compile_per_key_done.db") file_path = os.path.join(tmpdir, "compile_per_key_done.gz") with open(file_path, "wb") as fh: fh.write(b"gzip-data") self._prepare_db(db_path) self._insert_flux_success(db_path, file_path) model = ExportUploadModel(db_path=db_path) api = _UploadApiStub( ok=True, compile_ok=True, compile_payload={ "status": 1, "idempotency": { "per_key": { "found": 1, "status": "PROCESSED", "compile_state": "DONE", "compiled_transaksi": 0, "compiled_data": 0, "compiled_registry": 1, "is_compiled_full": 0, } }, }, ) service = ExportUploadService( model=model, api_service=api, enabled_getter=lambda: True, batch_limit_getter=lambda: 10, timeout_getter=lambda: 10, retention_days_getter=lambda: 0, compile_check_enabled_getter=lambda: True, compile_poll_max_attempt_getter=lambda: 1, compile_poll_interval_getter=lambda: 0, ) result = service.process_pending_uploads() self.assertEqual(result["uploaded"], 1) self.assertEqual(result["retried"], 0) self.assertEqual(api.compile_calls, 1) def test_upload_with_compile_check_accepts_per_key_required_flags(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = os.path.join(tmpdir, "upload_compile_required_flags.db") file_path = os.path.join(tmpdir, "compile_required_flags.gz") with open(file_path, "wb") as fh: fh.write(b"gzip-data") self._prepare_db(db_path) self._insert_flux_success(db_path, file_path) model = ExportUploadModel(db_path=db_path) api = _UploadApiStub( ok=True, compile_ok=True, compile_payload={ "status": 1, "idempotency": { "per_key": { "found": 1, "status": "PROCESSED", "compile_state": "PENDING", "compiled_transaksi": 0, "compiled_data": 0, "compiled_registry": 1, "is_compiled_full": 0, "required_flags": { "compiled_registry": 1, }, "required_stages": ["registry"], } }, }, ) service = ExportUploadService( model=model, api_service=api, enabled_getter=lambda: True, batch_limit_getter=lambda: 10, timeout_getter=lambda: 10, retention_days_getter=lambda: 0, compile_check_enabled_getter=lambda: True, compile_poll_max_attempt_getter=lambda: 1, compile_poll_interval_getter=lambda: 0, ) result = service.process_pending_uploads() self.assertEqual(result["uploaded"], 1) self.assertEqual(result["retried"], 0) self.assertEqual(api.compile_calls, 1) def test_housekeeping_cleans_old_empty_flux_rows(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = os.path.join(tmpdir, "upload_cleanup_empty.db") self._prepare_db(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ INSERT INTO export_flux ( batch_start, batch_end, table_name, server_hash, file_seq, file_hash, file_size, row_count, status, file_path, error_log, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now', '-10 day'), datetime('now', '-10 day')) """, ( "0-0", "2026-02-18 00:00:00", "settlement_history", "serverhash", None, "", 0, 0, "SUCCESS", None, "", ), ) conn.commit() conn.close() model = ExportUploadModel(db_path=db_path) service = ExportUploadService( model=model, api_service=_UploadApiStub(ok=True), enabled_getter=lambda: False, retention_days_getter=lambda: 0, empty_flux_retention_days_getter=lambda: 1, empty_flux_cleanup_limit_getter=lambda: 50, ) result = service.process_pending_uploads() self.assertEqual(int(result.get("cleaned_empty_flux") or 0), 1) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("SELECT COUNT(1) FROM export_flux") count = int(cur.fetchone()[0] or 0) conn.close() self.assertEqual(count, 0) if __name__ == "__main__": unittest.main()