# edited by glg import unittest from datetime import datetime, timedelta from pypos.modules.dashboard.controllers.dashboard_controller import DashboardController class _DummyView: pass class _DummyAppController: pass class DashboardAutoSyncNotifyModeTests(unittest.TestCase): def _build_controller(self): return DashboardController(_DummyView(), _DummyAppController()) def test_extract_changed_tables_mendukung_rows_dan_samples(self): ctrl = self._build_controller() payload = { "row": 12, "data": { "price": { "rows": 5, "samples": [ {"entity_id": 1, "entity_name": "AOKA MIX", "changed_field": "harga_jual"}, ], }, "produk": { "rows": 7, "samples": [ {"entity_id": 2, "entity_name": "Nabati", "changed_field": "nama"}, ], }, }, } changed = ctrl._extract_changed_tables_from_check( check_response=payload, requested_tables=["price", "produk"], ) self.assertEqual(len(changed), 2) self.assertEqual(changed[0]["table"], "produk") self.assertEqual(changed[0]["rows"], 7) self.assertEqual(changed[1]["table"], "price") self.assertEqual(changed[1]["rows"], 5) self.assertEqual(changed[1]["samples"][0]["entity_name"], "AOKA MIX") def test_extract_row_total_fallback_ke_penjumlahan_tabel(self): ctrl = self._build_controller() changed = [ {"table": "price", "rows": 5}, {"table": "produk", "rows": 7}, ] total = ctrl._extract_row_total_from_check( check_response={"data": {"price": {"rows": 5}, "produk": {"rows": 7}}}, changed_tables=changed, ) self.assertEqual(total, 12) def test_prompt_message_memuat_contoh_data_berubah(self): ctrl = self._build_controller() changed = [ { "table": "price", "rows": 5, "label": "Harga Jual", "samples": [ {"entity_name": "AOKA MIX", "entity_id": "1", "changed_field": "harga_jual"}, ], } ] message = ctrl._build_auto_sync_prompt_message(5, changed) self.assertIn("Ada perubahan pada produk dari server.", message) self.assertIn("Produk yang berubah:", message) self.assertIn("AOKA MIX", message) def test_prompt_message_harga_lama_baru_tanpa_id(self): ctrl = self._build_controller() changed = [ { "table": "price", "rows": 1, "label": "Harga Jual", "samples": [ { "entity_name": "Le Minerale 600 ml", "entity_id": "139051", "changed_field": "last_update", "before_value": 2500, "after_value": 3000, }, ], } ] message = ctrl._build_auto_sync_prompt_message(1, changed) self.assertIn("Le Minerale 600 ml", message) self.assertIn("Produk yang berubah:", message) self.assertNotIn("harga", message.lower()) self.assertNotIn("ID 139051", message) def test_extract_samples_mendukung_harga_lama_baru(self): ctrl = self._build_controller() raw_value = { "samples": [ { "entity_name": "Toples 800 ml", "changed_field": "harga_jual", "old_value": 10000, "new_value": 12000, } ] } samples = ctrl._extract_check_update_samples(raw_value) self.assertEqual(len(samples), 1) self.assertEqual(samples[0]["before_value"], 10000) self.assertEqual(samples[0]["after_value"], 12000) def test_build_price_delta_samples_from_rows_menghasilkan_nama_dan_harga(self): ctrl = self._build_controller() ctrl._lookup_local_product_price_snapshot = ( lambda produk_id, cabang_id, limit=8: ("Le Minerale 600 ml", [2500, 2400]) ) rows = [ {"produk_id": 1001, "jenis_value": "harga_list", "nilai": 3000}, {"produk_id": 1001, "jenis_value": "harga_list", "nilai": 3000}, ] samples = ctrl._build_price_delta_samples_from_rows(rows, cabang_id=101, max_items=3) self.assertEqual(len(samples), 1) self.assertEqual(samples[0]["entity_name"], "Le Minerale 600 ml") self.assertEqual(samples[0]["before_value"], 2500) self.assertEqual(samples[0]["after_value"], 3000) def test_resolve_before_price_mengambil_harga_terakhir_yang_berbeda(self): ctrl = self._build_controller() before = ctrl._resolve_before_price([89000, 87000, 86000], 89000) self.assertEqual(before, 87000) def test_build_price_delta_samples_tidak_menampilkan_old_old(self): ctrl = self._build_controller() ctrl._lookup_local_product_price_snapshot = ( lambda produk_id, cabang_id, limit=8: ("1 Bangku LPT B", [89000, 87000]) ) rows = [{"produk_id": 2001, "jenis_value": "harga_list", "nilai": 89000}] samples = ctrl._build_price_delta_samples_from_rows(rows, cabang_id=101, max_items=1) self.assertEqual(len(samples), 1) self.assertEqual(samples[0]["before_value"], 87000) self.assertEqual(samples[0]["after_value"], 89000) def test_enrich_changed_tables_with_price_delta_saat_samples_server_kosong(self): ctrl = self._build_controller() ctrl._collect_price_delta_samples = lambda model, machine_id, cabang_id, max_items=3: [ { "entity_name": "Toples 800 ml", "entity_id": "1002", "changed_field": "harga_jual", "before_value": 10000, "after_value": 12000, } ] changed = [ {"table": "price", "rows": 2, "label": "Harga Jual", "samples": []}, {"table": "produk", "rows": 1, "label": "Produk", "samples": []}, ] out = ctrl._enrich_changed_tables_with_price_delta( changed_tables=changed, model=object(), machine_id="M1", cabang_id=101, ) self.assertEqual(len(out), 2) self.assertEqual(out[0]["table"], "price") self.assertEqual(out[0]["samples"][0]["entity_name"], "Toples 800 ml") self.assertEqual(out[0]["samples"][0]["before_value"], 10000) self.assertEqual(out[0]["samples"][0]["after_value"], 12000) def test_prompt_message_harga_string_lokal_diformat_benar(self): ctrl = self._build_controller() changed = [ { "table": "price", "rows": 1, "label": "Harga Jual", "samples": [ { "entity_name": "Produk Uji", "before_value": "2.500", "after_value": "12.000", }, ], } ] message = ctrl._build_auto_sync_prompt_message(1, changed) self.assertIn("Produk yang berubah:", message) self.assertIn("Produk Uji", message) self.assertNotIn("Rp", message) def test_nanti_menunda_notifikasi_signature_sama_sebelum_ttl(self): ctrl = self._build_controller() ctrl.config_service.get_auto_sync_notify_reminder_minutes = lambda: 10 prompt_calls = [] ctrl.ask_confirm = lambda *args, **kwargs: (prompt_calls.append(1) or False) payload = { "has_update": True, "row_total": 2, "changed_tables": [{"table": "price", "rows": 2, "label": "Harga Jual", "samples": []}], "signature": "sig-abc", } ctrl.on_auto_sync_update_detected(payload) ctrl.on_auto_sync_update_detected(payload) self.assertEqual(len(prompt_calls), 1) self.assertTrue(ctrl._is_auto_sync_update_snoozed("sig-abc")) def test_sinkron_sekarang_memaksa_prompt_meski_masih_snooze(self): ctrl = self._build_controller() ctrl.config_service.get_auto_sync_notify_reminder_minutes = lambda: 10 prompt_calls = [] ctrl.ask_confirm = lambda *args, **kwargs: (prompt_calls.append(1) or False) payload = { "has_update": True, "row_total": 2, "changed_tables": [{"table": "price", "rows": 2, "label": "Harga Jual", "samples": []}], "signature": "sig-xyz", } ctrl.on_auto_sync_update_detected(payload) handled = ctrl.handle_sync_now_request() self.assertTrue(handled) self.assertEqual(len(prompt_calls), 2) def test_notifikasi_muncul_lagi_setelah_ttl_habis(self): ctrl = self._build_controller() ctrl.config_service.get_auto_sync_notify_reminder_minutes = lambda: 10 prompt_calls = [] ctrl.ask_confirm = lambda *args, **kwargs: (prompt_calls.append(1) or False) payload = { "has_update": True, "row_total": 2, "changed_tables": [{"table": "price", "rows": 2, "label": "Harga Jual", "samples": []}], "signature": "sig-ttl", } ctrl.on_auto_sync_update_detected(payload) ctrl._auto_sync_update_snoozed_until = datetime.now() - timedelta(seconds=1) ctrl.on_auto_sync_update_detected(payload) self.assertEqual(len(prompt_calls), 2) if __name__ == "__main__": unittest.main()