# edited by glg import threading import unittest from pypos.core.utils import worker_pool_utils from pypos.modules.penjualan.controllers.history_transaksi_controller import ( HistoryTransaksiController, ) from pypos.modules.penjualan.controllers.load_transaksi_controller import ( LoadTransaksiController, ) from pypos.modules.penjualan.controllers.settlement_controller import SettlementController from pypos.modules.customer.controllers.customer_search_controller import CustomerSearchController class _HistoryViewFake: def __init__(self): self.rows_rendered = [] self.detail_payload = None def tampilkan_history_rows(self, rows): self.rows_rendered = list(rows or []) def tampilkan_detail_transaksi(self, header_data, detail_rows, voucher_info, return_items): self.detail_payload = { "header_data": header_data, "detail_rows": list(detail_rows or []), "voucher_info": dict(voucher_info or {}), "return_items": list(return_items or []), } class _SettlementViewFake: def __init__(self): self.transaksi_rendered = [] self.non_tunai_rendered = {} self.history_rendered = [] self.rekap_text = None self.uang_input = None def render_transaksi_list(self, rows): self.transaksi_rendered = list(rows or []) def tampilkan_info_non_tunai(self, summary): self.non_tunai_rendered = dict(summary or {}) def tampilkan_history(self, history): self.history_rendered = list(history or []) def tampilkan_info_transaksi(self, _nilai): return None class _SettlementStateServiceFake: def __init__(self): self._transaksi_data = [] self._total_non_tunai = 0.0 self._summary = {} def set_transaksi_data(self, value): self._transaksi_data = list(value or []) def get_transaksi_data(self): return list(self._transaksi_data or []) def set_non_tunai_summary(self, summary, total_non_tunai): self._summary = dict(summary or {}) self._total_non_tunai = float(total_non_tunai or 0.0) def get_total_non_tunai(self): return float(self._total_non_tunai or 0.0) def has_transaksi_data(self): return bool(self._transaksi_data) def get_total_settlement(self): return 0.0 class _SettlementViewDataServiceFake: def calculate_total_non_tunai(self, summary): return float((summary or {}).get("total_non_tunai", 0.0) or 0.0) def build_rekap_text(self, _rekap): return "" class _LoadTableFake: def __init__(self): self.clear_selection_called = False def clearSelection(self): self.clear_selection_called = True class _LoadViewFake: def __init__(self): self.rows_rendered = [] self.preview_reset_called = False self.preview_rows = [] self.table = _LoadTableFake() def render_transaksi_list(self, rows): self.rows_rendered = list(rows or []) def reset_preview(self): self.preview_reset_called = True def tampilkan_preview_detail(self, rows): self.preview_rows = list(rows or []) class _CustomerSearchViewFake: def __init__(self): self.rows = [] self.offset = 0 self.pagination_payload = {} def render_customers(self, customers, current_offset): self.rows = list(customers or []) self.offset = int(current_offset or 0) def update_pagination(self, **kwargs): self.pagination_payload = dict(kwargs or {}) class AsyncRefreshStaleGuardTests(unittest.TestCase): def test_history_controller_ignores_stale_history_payload(self): controller = HistoryTransaksiController.__new__(HistoryTransaksiController) controller._history_request_seq = 0 controller._history_request_lock = threading.Lock() controller.view = _HistoryViewFake() controller.log_info = lambda *_args, **_kwargs: None latest_request_id = controller._next_history_request_id() controller._on_history_payload_ready( latest_request_id - 1, {"rows": [{"id": "lama"}], "query_elapsed_ms": 0.0}, ) self.assertEqual(controller.view.rows_rendered, []) controller._on_history_payload_ready( latest_request_id, {"rows": [{"id": "baru"}], "query_elapsed_ms": 0.0}, ) self.assertEqual(controller.view.rows_rendered, [{"id": "baru"}]) def test_settlement_controller_ignores_stale_refresh_payload(self): controller = SettlementController.__new__(SettlementController) controller._refresh_request_seq = 0 controller._refresh_request_lock = threading.Lock() controller.state_service = _SettlementStateServiceFake() controller.view_data_service = _SettlementViewDataServiceFake() controller.view = _SettlementViewFake() controller._view_supports_rekap = lambda: False controller._update_settle_button_state = lambda *_args, **_kwargs: None controller.log_info = lambda *_args, **_kwargs: None latest_request_id = controller._next_refresh_request_id() stale_payload = { "transaksi_data": [{"tanggal": "2026-04-02"}], "non_tunai_summary": {"total_non_tunai": 1000}, "history": [{"tanggal": "2026-04-02"}], "include_rekap": False, "transaksi_query_elapsed_ms": 0.0, "rekap_query_elapsed_ms": 0.0, "non_tunai_query_elapsed_ms": 0.0, "history_query_elapsed_ms": 0.0, } controller._on_refresh_payload_ready(latest_request_id - 1, stale_payload) self.assertEqual(controller.view.transaksi_rendered, []) self.assertEqual(controller.view.history_rendered, []) controller._on_refresh_payload_ready(latest_request_id, stale_payload) self.assertEqual(controller.view.transaksi_rendered, [{"tanggal": "2026-04-02"}]) self.assertEqual(controller.view.history_rendered, [{"tanggal": "2026-04-02"}]) self.assertEqual(controller.view.non_tunai_rendered, {"total_non_tunai": 1000}) def test_settlement_controller_history_only_mode_updates_history(self): controller = SettlementController.__new__(SettlementController) controller._refresh_request_seq = 0 controller._refresh_request_lock = threading.Lock() controller.state_service = _SettlementStateServiceFake() controller.view_data_service = _SettlementViewDataServiceFake() controller.view = _SettlementViewFake() controller._view_supports_rekap = lambda: False controller._update_settle_button_state = lambda *_args, **_kwargs: None controller.log_info = lambda *_args, **_kwargs: None latest_request_id = controller._next_refresh_request_id() controller._on_refresh_payload_ready( latest_request_id, { "mode": "history_only", "history": [{"tanggal": "2026-04-03"}], "history_query_elapsed_ms": 0.0, }, ) self.assertEqual(controller.view.history_rendered, [{"tanggal": "2026-04-03"}]) self.assertEqual(controller.view.transaksi_rendered, []) def test_load_transaksi_controller_ignores_stale_list_payload(self): controller = LoadTransaksiController.__new__(LoadTransaksiController) controller._list_request_seq = 0 controller._list_request_lock = threading.Lock() controller.view = _LoadViewFake() controller._row_contexts = [] controller._selected_context = None controller.transaksi_terpilih_id = 99 latest_request_id = controller._next_list_request_id() controller._on_list_payload_ready( latest_request_id - 1, { "display_rows": [["old"]], "row_contexts": [{"source": "local"}], }, ) self.assertEqual(controller.view.rows_rendered, []) self.assertFalse(controller.view.preview_reset_called) self.assertFalse(controller.view.table.clear_selection_called) controller._on_list_payload_ready( latest_request_id, { "display_rows": [["new"]], "row_contexts": [{"source": "api"}], }, ) self.assertEqual(controller.view.rows_rendered, [["new"]]) self.assertEqual(controller._row_contexts, [{"source": "api"}]) self.assertIsNone(controller._selected_context) self.assertIsNone(controller.transaksi_terpilih_id) self.assertTrue(controller.view.preview_reset_called) self.assertTrue(controller.view.table.clear_selection_called) def test_customer_search_controller_ignores_stale_reload_payload(self): controller = CustomerSearchController.__new__(CustomerSearchController) controller._reload_request_seq = 0 controller._reload_request_lock = threading.Lock() controller.page_size = 10 controller.current_offset = 0 controller.total_count = 0 controller.customer_list = [] controller.view = _CustomerSearchViewFake() latest_request_id = controller._next_reload_request_id() controller._on_page_payload_ready( latest_request_id - 1, { "total_count": 2, "current_offset": 0, "customer_list": [{"id": 100}], }, ) self.assertEqual(controller.customer_list, []) self.assertEqual(controller.view.rows, []) controller._on_page_payload_ready( latest_request_id, { "total_count": 2, "current_offset": 0, "customer_list": [{"id": 200}], }, ) self.assertEqual(controller.customer_list, [{"id": 200}]) self.assertEqual(controller.view.rows, [{"id": 200}]) self.assertEqual(int(controller.total_count), 2) def test_load_transaksi_preview_fallback_sync_saat_queue_preview_penuh(self): controller = LoadTransaksiController.__new__(LoadTransaksiController) controller._preview_request_seq = 1 controller._preview_request_lock = threading.Lock() controller.transaksi_terpilih_id = 33 controller.view = _LoadViewFake() controller._fetch_preview_payload = lambda parsed_id: { "parsed_id": int(parsed_id), "detail_rows": [{"id": int(parsed_id)}], } original_sem = worker_pool_utils._UI_POOL_PENDING_SEMAPHORES.get("preview") try: worker_pool_utils._UI_POOL_PENDING_SEMAPHORES["preview"] = threading.Semaphore(0) controller._run_async_preview_fetch(request_id=1, parsed_id=33) self.assertEqual(controller.view.preview_rows, [{"id": 33}]) finally: if original_sem is not None: worker_pool_utils._UI_POOL_PENDING_SEMAPHORES["preview"] = original_sem def test_history_detail_fallback_sync_saat_queue_preview_penuh(self): controller = HistoryTransaksiController.__new__(HistoryTransaksiController) controller._detail_request_seq = 1 controller._detail_request_lock = threading.Lock() controller.view = _HistoryViewFake() controller._fetch_detail_payload = lambda transaksi_id_text: { "header_data": {"id": transaksi_id_text}, "detail_rows": [{"produk": "A"}], "voucher_info": {"kode": "VCR-1"}, "return_items": [{"produk": "B"}], } original_sem = worker_pool_utils._UI_POOL_PENDING_SEMAPHORES.get("preview") try: worker_pool_utils._UI_POOL_PENDING_SEMAPHORES["preview"] = threading.Semaphore(0) controller._run_async_detail_fetch(request_id=1, transaksi_id_text="33") self.assertIsNotNone(controller.view.detail_payload) self.assertEqual(controller.view.detail_payload["detail_rows"], [{"produk": "A"}]) finally: if original_sem is not None: worker_pool_utils._UI_POOL_PENDING_SEMAPHORES["preview"] = original_sem if __name__ == "__main__": unittest.main()