import json import os import tempfile import threading import time import unittest from types import SimpleNamespace from unittest.mock import patch from modules.quantri import thongbao try: import fcntl except ImportError: # pragma: no cover - production hosting is Unix-based. fcntl = None class FakeClient: def __init__(self, groups=None, uid="bot-1", send_failures=None): self.uid = uid self.groups = { str(group_id): dict(record) for group_id, record in (groups or {}).items() } self.send_failures = { str(group_id): error for group_id, error in (send_failures or {}).items() } self.replies = [] self.sent_images = [] def fetchAllGroups(self): return SimpleNamespace(gridVerMap={group_id: 0 for group_id in self.groups}) def fetchGroupInfo(self, group_id): normalized_id = str(group_id) if normalized_id not in self.groups: raise RuntimeError(f"Unknown group: {normalized_id}") return SimpleNamespace( gridInfoMap={normalized_id: dict(self.groups[normalized_id])} ) def sendLocalImage(self, **kwargs): group_id = str(kwargs["thread_id"]) failure = self.send_failures.get(group_id) if failure is not None: if isinstance(failure, BaseException): raise failure raise RuntimeError(str(failure)) self.sent_images.append(kwargs) return SimpleNamespace(message_id=f"message-{group_id}") def replyMessage( self, message, reply_target, thread_id, thread_type, ttl=0 ): self.replies.append(message.text) def group_record( name, *, lock_send_msg=0, admin_ids=None, creator_id="owner-1", ): return { "name": name, "creatorId": creator_id, "adminIds": list(admin_ids or []), "setting": {"lockSendMsg": lock_send_msg}, } def campaign( campaign_id="campaign-1", *, admin_id="admin-1", status="ready", target_ids=None, target_names=None, delay_seconds=5, sent=None, failed=None, skipped_locked=None, in_flight=None, cancel_requested=False, ): target_ids = list(target_ids or []) target_names = dict(target_names or {group_id: group_id for group_id in target_ids}) return { "id": campaign_id, "status": status, "admin_id": admin_id, "content": "Khuyen mai hom nay", "image_path": "/tmp/promotion.jpg", "image_url": "https://example.test/promotion.jpg", "target_mode": "custom", "target_ids": target_ids, "target_names": target_names, "delay_seconds": delay_seconds, "sent": dict(sent or {}), "failed": dict(failed or {}), "skipped_locked": dict(skipped_locked or {}), "in_flight": dict(in_flight or {}), "attempts": 0, "confirm_token": None, "pending_action": None, "cancel_requested": cancel_requested, "created_at": "2026-09-16T08:00:00+07:00", "updated_at": "2026-09-16T08:00:00+07:00", } def state_with(*campaigns): return { "version": 1, "campaigns": {item["id"]: item for item in campaigns}, "active_by_admin": { str(item["admin_id"]): item["id"] for item in campaigns }, } class ThongBaoCommandTests(unittest.TestCase): def setUp(self): for campaign_id in list(thongbao._run_lock_handles): thongbao._release_broadcast_run(campaign_id) thongbao._active_runs.clear() self.temp_dir = tempfile.TemporaryDirectory() self.state_file = os.path.join(self.temp_dir.name, "thongbao_state.json") self.state_file_patch = patch.object( thongbao, "STATE_FILE", self.state_file ) self.state_file_patch.start() def tearDown(self): for campaign_id in list(thongbao._run_lock_handles): thongbao._release_broadcast_run(campaign_id) thongbao._active_runs.clear() self.state_file_patch.stop() self.temp_dir.cleanup() def save_campaign(self, item): thongbao.save_state(state_with(item)) def load_campaign(self, campaign_id="campaign-1"): return thongbao.load_state()["campaigns"][campaign_id] def test_command_is_admin_only_and_normalizes_numeric_admin_ids(self): client = FakeClient() with patch.object(thongbao, "ADMIN", ["100"]): thongbao.handle_thongbao_command( ".thongbao xem", SimpleNamespace(), "control-group", "GROUP", "not-admin", client, ) self.assertIn("không có quyền", client.replies[-1].casefold()) thongbao.handle_thongbao_command( ".thongbao xem", SimpleNamespace(), "control-group", "GROUP", 100, client, ) self.assertNotIn("không có quyền", client.replies[-1].casefold()) self.assertEqual(client.sent_images, []) def test_resolve_all_groups_returns_normalized_ids_and_names(self): client = FakeClient( { 101: group_record("Nhóm Một"), "002": group_record("Nhóm Hai"), } ) resolved = thongbao.resolve_all_groups(client) self.assertEqual(resolved, {"101": "Nhóm Một", "002": "Nhóm Hai"}) def test_resolve_custom_groups_deduplicates_and_rejects_unknown_groups(self): client = FakeClient( { "g1": group_record("Nhóm Một"), "g2": group_record("Nhóm Hai"), } ) resolved = thongbao.resolve_custom_groups(client, "g2, g1, g2") self.assertEqual(list(resolved), ["g2", "g1"]) self.assertEqual(resolved["g2"], "Nhóm Hai") with self.assertRaises(ValueError): thongbao.resolve_custom_groups(client, "g1, missing") def test_scope_commands_persist_all_or_custom_targets(self): item = campaign(target_ids=[]) self.save_campaign(item) client = FakeClient( { "g1": group_record("Nhóm Một"), "g2": group_record("Nhóm Hai"), "g3": group_record("Nhóm Ba"), } ) with patch.object(thongbao, "ADMIN", ["admin-1"]): thongbao.handle_thongbao_command( ".thongbao phamvi nhom g3,g1,g3", SimpleNamespace(), "control-group", "GROUP", "admin-1", client, ) custom = self.load_campaign() self.assertEqual(custom["target_ids"], ["g3", "g1"]) self.assertEqual( custom["target_names"], {"g3": "Nhóm Ba", "g1": "Nhóm Một"} ) thongbao.handle_thongbao_command( ".thongbao phamvi all", SimpleNamespace(), "control-group", "GROUP", "admin-1", client, ) all_groups = self.load_campaign() self.assertEqual(all_groups["target_mode"], "all") self.assertEqual(all_groups["target_ids"], ["g1", "g2", "g3"]) self.assertEqual(all_groups["target_names"]["g2"], "Nhóm Hai") def test_delay_accepts_only_integer_seconds_from_two_to_three_hundred(self): for raw, expected in (("2", 2), (17, 17), ("300", 300)): with self.subTest(raw=raw): self.assertEqual(thongbao.validate_delay(raw), expected) for raw in ("", "1", 1, "301", 301, "2.5", "abc", None, True): with self.subTest(raw=raw): with self.assertRaises(ValueError): thongbao.validate_delay(raw) def test_locked_non_admin_is_skipped_but_admin_and_creator_can_send(self): cases = ( (group_record("Mở", lock_send_msg="0"), False), (group_record("Khóa", lock_send_msg="1"), True), ( group_record( "Bot là admin", lock_send_msg=True, admin_ids=[12345] ), False, ), ( group_record( "Bot là chủ nhóm", lock_send_msg=1, creator_id=12345 ), False, ), ) for index, (record, expected) in enumerate(cases): with self.subTest(name=record["name"]): group_id = f"g{index}" client = FakeClient({group_id: record}, uid="12345") self.assertEqual( thongbao.is_group_locked_for_bot(client, group_id), expected ) def test_send_is_sequential_skips_locked_groups_and_reports_names(self): item = campaign( target_ids=["g1", "g2", "g3"], target_names={ "g1": "Nhóm Thành Công", "g2": "Nhóm Lỗi", "g3": "Nhóm Bị Khóa", }, delay_seconds=7, ) self.save_campaign(item) client = FakeClient( { "g1": group_record("Nhóm Thành Công"), "g2": group_record("Nhóm Lỗi"), "g3": group_record("Nhóm Bị Khóa", lock_send_msg=1), }, send_failures={"g2": "Zalo rejected the message"}, ) sleeps = [] report = thongbao.send_broadcast( client, "campaign-1", sleep_fn=sleeps.append, ) self.assertEqual( [str(call["thread_id"]) for call in client.sent_images], ["g1"] ) self.assertEqual(sleeps, [7]) self.assertEqual(report["sent_count"], 1) self.assertEqual(report["failed_count"], 0) self.assertEqual(report["uncertain_count"], 1) self.assertEqual(report["skipped_locked_count"], 1) rendered_report = json.dumps(report, ensure_ascii=False) self.assertIn("Nhóm Thành Công", rendered_report) self.assertIn("Nhóm Lỗi", rendered_report) self.assertIn("Nhóm Bị Khóa", rendered_report) saved = self.load_campaign() self.assertIn("g1", saved["sent"]) self.assertIn("g2", saved["in_flight"]) self.assertIn("g3", saved["skipped_locked"]) def test_success_and_uncertain_send_are_excluded_on_resume(self): item = campaign( target_ids=["g1", "g2", "g3"], target_names={"g1": "Một", "g2": "Hai", "g3": "Ba"}, delay_seconds=2, ) self.save_campaign(item) interrupted_client = FakeClient( { "g1": group_record("Một"), "g2": group_record("Hai"), "g3": group_record("Ba"), }, send_failures={"g2": KeyboardInterrupt("simulated process stop")}, ) with self.assertRaises(KeyboardInterrupt): thongbao.send_broadcast( interrupted_client, "campaign-1", sleep_fn=lambda _seconds: None, ) state_after_interruption = self.load_campaign() self.assertIn("g1", state_after_interruption["sent"]) self.assertIn("g2", state_after_interruption["in_flight"]) resumed_client = FakeClient( { "g1": group_record("Một"), "g2": group_record("Hai"), "g3": group_record("Ba"), } ) thongbao.send_broadcast( resumed_client, "campaign-1", sleep_fn=lambda _seconds: None, ) resumed_ids = [ str(call["thread_id"]) for call in resumed_client.sent_images ] self.assertEqual(resumed_ids, ["g3"]) self.assertNotIn("g1", resumed_ids) self.assertNotIn("g2", resumed_ids) def test_send_exception_stays_uncertain_and_retry_does_not_resend_it(self): item = campaign( target_ids=["g1"], target_names={"g1": "Nhóm Chưa Rõ"}, ) self.save_campaign(item) failed_client = FakeClient( {"g1": group_record("Nhóm Chưa Rõ")}, send_failures={"g1": RuntimeError("response lost after upload")}, ) report = thongbao.send_broadcast( failed_client, "campaign-1", sleep_fn=lambda _seconds: None, ) saved = self.load_campaign() self.assertIn("g1", saved["in_flight"]) self.assertNotIn("g1", saved["failed"]) self.assertEqual(report["uncertain_count"], 1) self.assertEqual( thongbao.build_pending_targets(saved, retry_failed=True), [] ) retry_client = FakeClient({"g1": group_record("Nhóm Chưa Rõ")}) thongbao.send_broadcast( retry_client, "campaign-1", retry_failed=True, sleep_fn=lambda _seconds: None, ) self.assertEqual(retry_client.sent_images, []) def test_concurrent_workers_allow_only_one_active_broadcast(self): item = campaign( target_ids=["g1"], target_names={"g1": "Nhóm Một"}, ) self.save_campaign(item) send_started = threading.Event() release_send = threading.Event() class ConcurrentClient(FakeClient): def sendLocalImage(self, **kwargs): send_started.set() if not release_send.wait(timeout=2): raise RuntimeError("test send timed out") return super().sendLocalImage(**kwargs) client = ConcurrentClient({"g1": group_record("Nhóm Một")}) errors = [] def run_worker(): try: thongbao.send_broadcast( client, "campaign-1", sleep_fn=lambda _seconds: None, ) except BaseException as error: errors.append(error) first_worker = threading.Thread(target=run_worker) second_worker = threading.Thread(target=run_worker) first_worker.start() self.assertTrue(send_started.wait(timeout=1)) second_worker.start() second_worker.join(timeout=1) release_send.set() first_worker.join(timeout=3) self.assertEqual(len(errors), 1) self.assertIsInstance(errors[0], ValueError) self.assertFalse(first_worker.is_alive()) self.assertFalse(second_worker.is_alive()) self.assertEqual( [str(call["thread_id"]) for call in client.sent_images], ["g1"] ) @unittest.skipIf(fcntl is None, "fcntl is required for the cross-process lock") def test_external_process_lock_blocks_a_second_broadcast_claim(self): lock_path = f"{self.state_file}.run.lock" with open(lock_path, "a+", encoding="utf-8") as external_lock: fcntl.flock(external_lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) self.assertTrue(thongbao.has_active_broadcast()) self.assertTrue( thongbao._campaign_is_running( "campaign-1", campaign(status="ready", target_ids=[]), ) ) with self.assertRaises(ValueError): thongbao._claim_broadcast_run("campaign-1") self.assertNotIn("campaign-1", thongbao._active_runs) fcntl.flock(external_lock.fileno(), fcntl.LOCK_UN) @unittest.skipIf(fcntl is None, "fcntl is required for the cross-process lock") def test_external_state_lock_serializes_state_transactions(self): self.save_campaign(campaign(target_ids=["g1"])) lock_path = f"{self.state_file}.lock" reader_started = threading.Event() reader_finished = threading.Event() loaded_states = [] def read_state(): reader_started.set() loaded_states.append(thongbao.load_state()) reader_finished.set() with open(lock_path, "a+", encoding="utf-8") as external_lock: fcntl.flock(external_lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) reader = threading.Thread(target=read_state) reader.start() self.assertTrue(reader_started.wait(timeout=1)) self.assertFalse(reader_finished.wait(timeout=0.05)) fcntl.flock(external_lock.fileno(), fcntl.LOCK_UN) self.assertTrue(reader_finished.wait(timeout=1)) reader.join(timeout=1) self.assertEqual( loaded_states[0]["campaigns"]["campaign-1"]["target_ids"], ["g1"], ) def test_preclaimed_run_remains_owned_until_caller_releases_it(self): item = campaign(target_ids=[]) self.save_campaign(item) client = FakeClient() thongbao._claim_broadcast_run("campaign-1") thongbao.send_broadcast( client, "campaign-1", run_claimed=True, sleep_fn=lambda _seconds: None, ) self.assertIn("campaign-1", thongbao._active_runs) thongbao._release_broadcast_run("campaign-1") def test_scope_resolution_cannot_overwrite_campaign_that_just_started(self): item = campaign( status="awaiting_confirmation", target_ids=["old"], target_names={"old": "Nhóm Cũ"}, ) item["confirm_token"] = "ABC123" item["confirm_expires_at"] = time.time() + 60 item["pending_action"] = "send" self.save_campaign(item) resolve_started = threading.Event() release_resolve = threading.Event() class SlowScopeClient(FakeClient): def fetchAllGroups(self): resolve_started.set() if not release_resolve.wait(timeout=2): raise RuntimeError("test scope resolution timed out") return super().fetchAllGroups() client = SlowScopeClient({"new": group_record("Nhóm Mới")}) worker = threading.Thread( target=thongbao.handle_thongbao_command, args=( ".thongbao phamvi all", SimpleNamespace(), "control-group", "GROUP", "admin-1", client, ), ) with patch.object(thongbao, "ADMIN", ["admin-1"]): worker.start() self.assertTrue(resolve_started.wait(timeout=1)) self.assertEqual( thongbao.consume_confirmation("campaign-1", "ABC123"), "send", ) release_resolve.set() worker.join(timeout=2) saved = self.load_campaign() self.assertFalse(worker.is_alive()) self.assertEqual(saved["status"], "running") self.assertEqual(saved["target_ids"], ["old"]) self.assertIn("không thể đổi phạm vi", client.replies[-1].casefold()) thongbao._release_broadcast_run("campaign-1") def test_delay_change_cannot_overwrite_campaign_that_just_started(self): item = campaign( status="awaiting_confirmation", target_ids=["g1"], target_names={"g1": "Nhóm Một"}, delay_seconds=5, ) item["confirm_token"] = "ABC123" item["confirm_expires_at"] = time.time() + 60 item["pending_action"] = "send" self.save_campaign(item) validation_started = threading.Event() release_validation = threading.Event() real_validate_delay = thongbao.validate_delay def slow_validate_delay(raw_value): validation_started.set() if not release_validation.wait(timeout=2): raise RuntimeError("test delay validation timed out") return real_validate_delay(raw_value) client = FakeClient() worker = threading.Thread( target=thongbao.handle_thongbao_command, args=( ".thongbao nghi 20", SimpleNamespace(), "control-group", "GROUP", "admin-1", client, ), ) with patch.object(thongbao, "ADMIN", ["admin-1"]), patch.object( thongbao, "validate_delay", side_effect=slow_validate_delay ): worker.start() self.assertTrue(validation_started.wait(timeout=1)) self.assertEqual( thongbao.consume_confirmation("campaign-1", "ABC123"), "send", ) release_validation.set() worker.join(timeout=2) saved = self.load_campaign() self.assertFalse(worker.is_alive()) self.assertEqual(saved["status"], "running") self.assertEqual(saved["delay_seconds"], 5) self.assertIn("không thể đổi thời gian nghỉ", client.replies[-1].casefold()) thongbao._release_broadcast_run("campaign-1") def test_send_confirmation_cannot_replace_a_campaign_that_just_started(self): item = campaign( status="awaiting_confirmation", target_ids=["g1"], target_names={"g1": "Nhóm Một"}, ) item["confirm_token"] = "OLD123" item["confirm_expires_at"] = time.time() + 60 item["pending_action"] = "send" self.save_campaign(item) pending_check_started = threading.Event() release_pending_check = threading.Event() real_build_pending_targets = thongbao.build_pending_targets first_check = True def slow_build_pending_targets(item, retry_failed=False): nonlocal first_check if first_check: first_check = False pending_check_started.set() if not release_pending_check.wait(timeout=2): raise RuntimeError("test pending target check timed out") return real_build_pending_targets(item, retry_failed=retry_failed) client = FakeClient() worker = threading.Thread( target=thongbao.handle_thongbao_command, args=( ".thongbao gui", SimpleNamespace(), "control-group", "GROUP", "admin-1", client, ), ) with patch.object(thongbao, "ADMIN", ["admin-1"]), patch.object( thongbao, "build_pending_targets", side_effect=slow_build_pending_targets, ): worker.start() self.assertTrue(pending_check_started.wait(timeout=1)) self.assertEqual( thongbao.consume_confirmation("campaign-1", "OLD123"), "send", ) release_pending_check.set() worker.join(timeout=2) saved = self.load_campaign() self.assertFalse(worker.is_alive()) self.assertEqual(saved["status"], "running") self.assertIsNone(saved["confirm_token"]) self.assertIsNone(saved["pending_action"]) self.assertIn("đang được gửi", client.replies[-1].casefold()) thongbao._release_broadcast_run("campaign-1") def test_retry_confirmation_cannot_replace_a_campaign_that_just_started(self): item = campaign( status="awaiting_confirmation", target_ids=["g1"], target_names={"g1": "Nhóm Một"}, failed={ "g1": { "name": "Nhóm Một", "reason": "safe pre-send failure", "retryable": True, } }, ) item["confirm_token"] = "OLD123" item["confirm_expires_at"] = time.time() + 60 item["pending_action"] = "retry" self.save_campaign(item) pending_check_started = threading.Event() release_pending_check = threading.Event() real_build_pending_targets = thongbao.build_pending_targets first_check = True def slow_build_pending_targets(item, retry_failed=False): nonlocal first_check if first_check: first_check = False pending_check_started.set() if not release_pending_check.wait(timeout=2): raise RuntimeError("test retry target check timed out") return real_build_pending_targets(item, retry_failed=retry_failed) client = FakeClient() worker = threading.Thread( target=thongbao.handle_thongbao_command, args=( ".thongbao retry", SimpleNamespace(), "control-group", "GROUP", "admin-1", client, ), ) with patch.object(thongbao, "ADMIN", ["admin-1"]), patch.object( thongbao, "build_pending_targets", side_effect=slow_build_pending_targets, ): worker.start() self.assertTrue(pending_check_started.wait(timeout=1)) self.assertEqual( thongbao.consume_confirmation("campaign-1", "OLD123"), "retry", ) release_pending_check.set() worker.join(timeout=2) saved = self.load_campaign() self.assertFalse(worker.is_alive()) self.assertEqual(saved["status"], "running") self.assertIsNone(saved["confirm_token"]) self.assertIsNone(saved["pending_action"]) self.assertIn("đang được gửi", client.replies[-1].casefold()) thongbao._release_broadcast_run("campaign-1") def test_confirmation_token_is_consumed_atomically_only_once(self): item = campaign(target_ids=["g1"], target_names={"g1": "Nhóm Một"}) item["status"] = "awaiting_confirmation" item["confirm_token"] = "ABC123" item["confirm_expires_at"] = time.time() + 60 item["pending_action"] = "send" self.save_campaign(item) actions = [] errors = [] def consume_token(): try: actions.append( thongbao.consume_confirmation("campaign-1", "ABC123") ) except ValueError as error: errors.append(str(error)) workers = [threading.Thread(target=consume_token) for _ in range(2)] for worker in workers: worker.start() for worker in workers: worker.join(timeout=2) self.assertEqual(actions, ["send"]) self.assertEqual(len(errors), 1) self.assertIn("xác nhận", errors[0].casefold()) def test_stale_running_campaign_recovers_as_interrupted_after_restart(self): item = campaign( status="running", target_ids=["g1", "g2"], target_names={"g1": "Một", "g2": "Hai"}, sent={"g1": {"name": "Một"}}, ) self.save_campaign(item) client = FakeClient() with patch.object(thongbao, "ADMIN", ["admin-1"]): thongbao.handle_thongbao_command( ".thongbao trangthai", SimpleNamespace(), "control-group", "GROUP", "admin-1", client, ) saved = self.load_campaign() self.assertEqual(saved["status"], "interrupted") self.assertIn("interrupted", client.replies[-1]) def test_cancel_requested_before_worker_start_is_not_cleared(self): item = campaign( status="running", target_ids=["g1"], target_names={"g1": "Nhóm Một"}, cancel_requested=True, ) self.save_campaign(item) client = FakeClient({"g1": group_record("Nhóm Một")}) report = thongbao.send_broadcast( client, "campaign-1", sleep_fn=lambda _seconds: None, ) self.assertEqual(client.sent_images, []) self.assertEqual(report["status"], "cancelled") def test_retry_that_finds_group_locked_clears_previous_failure(self): item = campaign( status="partial", target_ids=["g1"], target_names={"g1": "Nhóm Một"}, failed={ "g1": { "name": "Nhóm Một", "reason": "temporary", "retryable": True, } }, ) self.save_campaign(item) client = FakeClient( {"g1": group_record("Nhóm Một", lock_send_msg=1)} ) report = thongbao.send_broadcast( client, "campaign-1", retry_failed=True, sleep_fn=lambda _seconds: None, ) saved = self.load_campaign() self.assertNotIn("g1", saved["failed"]) self.assertIn("g1", saved["skipped_locked"]) self.assertEqual(report["status"], "completed_with_skips") def test_legacy_failure_without_retryable_flag_is_not_retried(self): item = campaign( status="partial", target_ids=["g1"], target_names={"g1": "Nhóm Một"}, failed={"g1": {"name": "Nhóm Một", "reason": "legacy error"}}, ) self.assertEqual( thongbao.build_pending_targets(item, retry_failed=True), [] ) def test_legacy_ambiguous_failure_loads_as_uncertain_delivery(self): item = campaign( status="partial", target_ids=["g1"], target_names={"g1": "Nhóm Một"}, failed={ "g1": { "name": "Nhóm Một", "reason": "legacy send error", } }, ) with open(self.state_file, "w", encoding="utf-8") as state_file: json.dump(state_with(item), state_file) loaded = self.load_campaign() self.assertNotIn("g1", loaded["failed"]) self.assertIn("g1", loaded["in_flight"]) self.assertTrue(loaded["in_flight"]["g1"]["delivery_uncertain"]) self.assertEqual( thongbao.build_pending_targets(loaded, retry_failed=True), [] ) def test_legacy_uncertain_marker_overrides_retryable_failure_fields(self): cases = ( { "reason": "legacy send error", "retryable": True, "delivery_uncertain": True, }, { "reason": "Không đọc được thông tin nhóm: legacy timeout", "delivery_uncertain": True, }, ) for index, failed_entry in enumerate(cases): with self.subTest(index=index): item = campaign( status="partial", target_ids=["g1"], target_names={"g1": "Nhóm Một"}, failed={"g1": {"name": "Nhóm Một", **failed_entry}}, ) with open(self.state_file, "w", encoding="utf-8") as state_file: json.dump(state_with(item), state_file) loaded = self.load_campaign() self.assertNotIn("g1", loaded["failed"]) self.assertIn("g1", loaded["in_flight"]) self.assertEqual( thongbao.build_pending_targets(loaded, retry_failed=True), [], ) client = FakeClient({"g1": group_record("Nhóm Một")}) thongbao.send_broadcast( client, "campaign-1", retry_failed=True, sleep_fn=lambda _seconds: None, ) self.assertEqual(client.sent_images, []) def test_retry_send_error_moves_group_from_failed_to_uncertain(self): item = campaign( status="partial", target_ids=["g1"], target_names={"g1": "Nhóm Một"}, failed={ "g1": { "name": "Nhóm Một", "reason": "safe pre-send failure", "retryable": True, } }, ) self.save_campaign(item) client = FakeClient( {"g1": group_record("Nhóm Một")}, send_failures={"g1": RuntimeError("response lost")}, ) report = thongbao.send_broadcast( client, "campaign-1", retry_failed=True, sleep_fn=lambda _seconds: None, ) saved = self.load_campaign() self.assertNotIn("g1", saved["failed"]) self.assertIn("g1", saved["in_flight"]) self.assertEqual(report["total_failed_count"], 0) self.assertEqual(report["uncertain_count"], 1) def test_admin_can_mark_uncertain_group_as_delivered_without_resending(self): item = campaign( status="interrupted", target_ids=["g1"], target_names={"g1": "Nhóm Một"}, in_flight={ "g1": {"name": "Nhóm Một", "reason": "response lost"} }, ) self.save_campaign(item) client = FakeClient() with patch.object(thongbao, "ADMIN", ["admin-1"]): thongbao.handle_thongbao_command( ".thongbao xuly g1 dagui", SimpleNamespace(), "control-group", "GROUP", "admin-1", client, ) saved = self.load_campaign() self.assertNotIn("g1", saved["in_flight"]) self.assertIn("g1", saved["sent"]) self.assertEqual(saved["status"], "completed") self.assertEqual(thongbao.build_pending_targets(saved), []) self.assertEqual(thongbao.build_pending_targets(saved, retry_failed=True), []) def test_admin_can_mark_uncertain_group_as_not_delivered_then_retry(self): item = campaign( status="interrupted", target_ids=["g1"], target_names={"g1": "Nhóm Một"}, in_flight={ "g1": {"name": "Nhóm Một", "reason": "response lost"} }, ) self.save_campaign(item) client = FakeClient() with patch.object(thongbao, "ADMIN", ["admin-1"]): thongbao.handle_thongbao_command( ".thongbao xuly g1 chuagui", SimpleNamespace(), "control-group", "GROUP", "admin-1", client, ) saved = self.load_campaign() self.assertNotIn("g1", saved["in_flight"]) self.assertTrue(saved["failed"]["g1"]["retryable"]) self.assertEqual(saved["status"], "partial") self.assertEqual( thongbao.build_pending_targets(saved, retry_failed=True), ["g1"] ) @unittest.skipIf(fcntl is None, "fcntl is required for the cross-process lock") def test_external_broadcast_blocks_uncertain_delivery_resolution(self): item = campaign( status="interrupted", target_ids=["g1"], target_names={"g1": "Nhóm Một"}, in_flight={ "g1": {"name": "Nhóm Một", "reason": "response pending"} }, ) self.save_campaign(item) lock_path = f"{self.state_file}.run.lock" with open(lock_path, "a+", encoding="utf-8") as external_lock: fcntl.flock(external_lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) with self.assertRaises(ValueError): thongbao.resolve_uncertain_delivery( "campaign-1", "g1", delivered=False, ) fcntl.flock(external_lock.fileno(), fcntl.LOCK_UN) saved = self.load_campaign() self.assertIn("g1", saved["in_flight"]) self.assertNotIn("g1", saved["failed"]) def test_cancelled_campaign_with_uncertain_groups_cannot_be_replaced(self): item = campaign( status="cancelled", target_ids=["g1"], target_names={"g1": "Nhóm Một"}, in_flight={"g1": {"name": "Nhóm Một"}}, ) self.assertFalse(thongbao._can_replace_campaign(item)) def test_corrupted_delivery_ledger_fails_closed(self): with open(self.state_file, "w", encoding="utf-8") as state_file: state_file.write("{not-json") with self.assertRaises(RuntimeError): thongbao.load_state() with open(self.state_file, "w", encoding="utf-8") as state_file: json.dump({}, state_file) with self.assertRaises(RuntimeError): thongbao.load_state() def test_corrupted_delivery_bucket_fails_closed(self): item = campaign( target_ids=["g1"], target_names={"g1": "Nhóm Một"}, sent={"g1": {"name": "Nhóm Một"}}, ) damaged_state = state_with(item) damaged_state["campaigns"]["campaign-1"]["sent"] = ["g1"] with open(self.state_file, "w", encoding="utf-8") as state_file: json.dump(damaged_state, state_file) with self.assertRaises(RuntimeError): thongbao.load_state() def test_missing_sent_ledger_fails_closed(self): damaged_state = state_with(campaign(target_ids=["g1"])) del damaged_state["campaigns"]["campaign-1"]["sent"] with open(self.state_file, "w", encoding="utf-8") as state_file: json.dump(damaged_state, state_file) with self.assertRaises(RuntimeError): thongbao.load_state() def test_campaign_history_without_active_mapping_fails_closed(self): damaged_state = state_with(campaign(target_ids=["g1"])) damaged_state["active_by_admin"] = {} with open(self.state_file, "w", encoding="utf-8") as state_file: json.dump(damaged_state, state_file) with self.assertRaises(RuntimeError): thongbao.load_state() def test_unresolved_orphan_campaign_fails_closed(self): active = campaign( campaign_id="campaign-active", status="completed", target_ids=[], ) orphan = campaign( campaign_id="campaign-orphan", status="interrupted", target_ids=["g1"], target_names={"g1": "Nhóm Một"}, in_flight={"g1": {"name": "Nhóm Một"}}, ) damaged_state = { "version": 1, "campaigns": { active["id"]: active, orphan["id"]: orphan, }, "active_by_admin": {"admin-1": active["id"]}, } with open(self.state_file, "w", encoding="utf-8") as state_file: json.dump(damaged_state, state_file) with self.assertRaises(RuntimeError): thongbao.load_state() with open(self.state_file, "w", encoding="utf-8") as state_file: json.dump([], state_file) with self.assertRaises(RuntimeError): thongbao.load_state() def test_reply_image_url_is_extracted_and_create_command_saves_draft(self): client = FakeClient() message_object = SimpleNamespace( quote=SimpleNamespace( attach=json.dumps( {"hdUrl": "https:\\/\\/photo.zdn.vn\\/promotion.jpg"} ) ) ) with patch.object(thongbao, "ADMIN", ["admin-1"]), patch.object( thongbao, "download_campaign_image", return_value=("/tmp/cached-promotion.jpg", 1200, 800), ) as download_image: thongbao.handle_thongbao_command( ".thongbao tao 5 Khuyến mãi cuối tuần", message_object, "control-group", "GROUP", "admin-1", client, ) saved_state = thongbao.load_state() campaign_id = saved_state["active_by_admin"]["admin-1"] saved = saved_state["campaigns"][campaign_id] self.assertEqual(saved["content"], "Khuyến mãi cuối tuần") self.assertEqual(saved["delay_seconds"], 5) self.assertEqual(saved["image_path"], "/tmp/cached-promotion.jpg") self.assertEqual((saved["image_width"], saved["image_height"]), (1200, 800)) download_image.assert_called_once_with( "https://photo.zdn.vn/promotion.jpg", campaign_id ) def test_slow_create_cannot_replace_a_newer_campaign(self): self.save_campaign(campaign(status="completed", target_ids=[])) download_started = threading.Event() release_download = threading.Event() def slow_download(_image_url, _campaign_id): download_started.set() if not release_download.wait(timeout=2): raise RuntimeError("test image download timed out") return ("/tmp/late-promotion.jpg", 1200, 800) client = FakeClient() message_object = SimpleNamespace( quote=SimpleNamespace( attach=json.dumps( {"hdUrl": "https:\\/\\/photo.zdn.vn\\/promotion.jpg"} ) ) ) worker = threading.Thread( target=thongbao.handle_thongbao_command, args=( ".thongbao tao 5 Khuyến mãi đến muộn", message_object, "control-group", "GROUP", "admin-1", client, ), ) with patch.object(thongbao, "ADMIN", ["admin-1"]), patch.object( thongbao, "_new_campaign_id", return_value="campaign-late" ), patch.object( thongbao, "download_campaign_image", side_effect=slow_download, ): worker.start() self.assertTrue(download_started.wait(timeout=1)) newer_campaign = campaign( campaign_id="campaign-newer", status="draft", target_ids=[], ) state = thongbao.load_state() state["campaigns"]["campaign-newer"] = newer_campaign state["active_by_admin"]["admin-1"] = "campaign-newer" thongbao.save_state(state) release_download.set() worker.join(timeout=2) saved_state = thongbao.load_state() self.assertFalse(worker.is_alive()) self.assertEqual( saved_state["active_by_admin"]["admin-1"], "campaign-newer", ) self.assertNotIn("campaign-late", saved_state["campaigns"]) self.assertIn("chiến dịch đã thay đổi", client.replies[-1].casefold()) def test_reply_image_url_supports_attachment_lists(self): message_object = SimpleNamespace( quote=SimpleNamespace( attach=None, qmsgAttach=json.dumps( [{"normalUrl": "https://photo.zdn.vn/list-image.png"}] ), ) ) self.assertEqual( thongbao.extract_quoted_image_url(message_object), "https://photo.zdn.vn/list-image.png", ) def test_retry_targets_only_prior_failures_and_never_sent_groups(self): item = campaign( target_ids=["g1", "g2", "g3", "g4"], target_names={ "g1": "Đã gửi Một", "g2": "Cần thử lại", "g3": "Bị khóa", "g4": "Đã gửi Bốn", }, sent={ "g1": {"name": "Đã gửi Một"}, "g4": {"name": "Đã gửi Bốn"}, }, failed={ "g1": {"name": "Đã gửi Một", "reason": "stale failure"}, "g2": { "name": "Cần thử lại", "reason": "temporary error", "retryable": True, }, }, skipped_locked={ "g3": {"name": "Bị khóa", "reason": "lockSendMsg"} }, ) self.save_campaign(item) self.assertEqual( thongbao.build_pending_targets(item, retry_failed=True), ["g2"] ) client = FakeClient( { "g1": group_record("Đã gửi Một"), "g2": group_record("Cần thử lại"), "g3": group_record("Bị khóa", lock_send_msg=1), "g4": group_record("Đã gửi Bốn"), } ) report = thongbao.send_broadcast( client, "campaign-1", retry_failed=True, sleep_fn=lambda _seconds: None, ) self.assertEqual( [str(call["thread_id"]) for call in client.sent_images], ["g2"] ) self.assertEqual(report["sent_count"], 1) saved = self.load_campaign() self.assertIn("g2", saved["sent"]) self.assertNotIn("g2", saved["failed"]) self.assertIn("g1", saved["sent"]) self.assertNotIn("g1", saved["failed"]) self.assertIn("g4", saved["sent"]) self.assertEqual(saved["status"], "completed_with_skips") def test_gui_and_retry_require_matching_confirmation_tokens(self): item = campaign( target_ids=["g1"], target_names={"g1": "Nhóm Một"}, ) self.save_campaign(item) client = FakeClient({"g1": group_record("Nhóm Một")}) send_started = threading.Event() calls = [] def fake_send_broadcast( passed_client, campaign_id, retry_failed=False, sleep_fn=time.sleep, run_claimed=False, ): calls.append((passed_client, campaign_id, retry_failed)) send_started.set() return { "sent_count": 0, "failed_count": 0, "skipped_locked_count": 0, } with patch.object(thongbao, "ADMIN", ["admin-1"]), patch.object( thongbao, "send_broadcast", side_effect=fake_send_broadcast ): thongbao.handle_thongbao_command( ".thongbao gui", SimpleNamespace(), "control-group", "GROUP", "admin-1", client, ) self.assertFalse(send_started.is_set()) send_token = self.load_campaign()["confirm_token"] self.assertTrue(send_token) thongbao.handle_thongbao_command( ".thongbao xacnhan sai-token", SimpleNamespace(), "control-group", "GROUP", "admin-1", client, ) self.assertFalse(send_started.is_set()) thongbao.handle_thongbao_command( f".thongbao xacnhan {send_token}", SimpleNamespace(), "control-group", "GROUP", "admin-1", client, ) self.assertTrue(send_started.wait(timeout=1)) self.assertEqual(calls[-1][1:], ("campaign-1", False)) retry_campaign = self.load_campaign() retry_campaign["failed"] = { "g1": { "name": "Nhóm Một", "reason": "temporary", "retryable": True, } } retry_campaign["status"] = "partial" self.save_campaign(retry_campaign) send_started.clear() thongbao.handle_thongbao_command( ".thongbao retry", SimpleNamespace(), "control-group", "GROUP", "admin-1", client, ) self.assertFalse(send_started.is_set()) retry_state = self.load_campaign() retry_token = retry_state["confirm_token"] self.assertTrue(retry_token) self.assertEqual(retry_state["pending_action"], "retry") thongbao.handle_thongbao_command( f".thongbao xacnhan {retry_token}", SimpleNamespace(), "control-group", "GROUP", "admin-1", client, ) self.assertTrue(send_started.wait(timeout=1)) self.assertEqual(calls[-1][1:], ("campaign-1", True)) if __name__ == "__main__": unittest.main()