# -*- coding: utf-8 -*- """回复队列的执行记录：写得下、读得出、坏了也不许把机器人带崩。""" import json import os import tempfile import time from unittest import TestCase, SkipTest, mock from queue_log import QueueLog import wechat_bot from wechat_bot import WeChatBot def _log(directory: str) -> QueueLog: return QueueLog(os.path.join(directory, "queue_events.json")) class QueueLogStoreTest(TestCase): def test_a_recorded_step_comes_back_with_everything_it_was_given(self): with tempfile.TemporaryDirectory() as directory: log = _log(directory) log.append("abc123", "高瑞@微信", "已发送", "好的，这就帮您看") entry = log.recent()[0] self.assertEqual(entry["session_id"], "abc123") self.assertEqual(entry["name"], "高瑞@微信") self.assertEqual(entry["event"], "已发送") self.assertEqual(entry["detail"], "好的，这就帮您看") self.assertGreater(entry["ts"], 0) def test_the_newest_step_is_the_one_you_see_first(self): """排查总是从"刚才发生了什么"开始，最新的必须在最上面。""" with tempfile.TemporaryDirectory() as directory: log = _log(directory) for event in ("入队", "开始处理", "已发送"): log.append("abc123", "高瑞@微信", event) self.assertEqual( [item["event"] for item in log.recent()], ["已发送", "开始处理", "入队"], ) def test_a_long_running_bot_never_grows_the_file_without_bound(self): with tempfile.TemporaryDirectory() as directory: log = _log(directory) log.MAX_EVENTS = 5 for index in range(12): log.append("abc123", "高瑞@微信", "已发送", str(index)) kept = log.recent(50) self.assertEqual(len(kept), 5) # 留下的必须是最近那几条，不能是最早的 self.assertEqual([item["detail"] for item in kept], ["11", "10", "9", "8", "7"]) def test_asking_for_fewer_steps_gives_exactly_that_many(self): with tempfile.TemporaryDirectory() as directory: log = _log(directory) for index in range(6): log.append("abc123", "高瑞@微信", "已发送", str(index)) self.assertEqual(len(log.recent(2)), 2) def test_reading_an_empty_history_is_not_an_error(self): with tempfile.TemporaryDirectory() as directory: self.assertEqual(_log(directory).recent(), []) def test_a_corrupted_history_starts_over_instead_of_crashing(self): """记录只是排查线索，它坏掉不能反过来让界面和机器人一起崩。""" with tempfile.TemporaryDirectory() as directory: path = os.path.join(directory, "queue_events.json") with open(path, "w", encoding="utf-8") as handle: handle.write("{这不是 JSON") log = QueueLog(path) self.assertEqual(log.recent(), []) self.assertTrue(log.append("abc123", "高瑞@微信", "入队")) self.assertEqual(len(log.recent()), 1) def test_junk_rows_are_dropped_rather_than_handed_to_the_table(self): with tempfile.TemporaryDirectory() as directory: path = os.path.join(directory, "queue_events.json") with open(path, "w", encoding="utf-8") as handle: json.dump({"events": ["坏行", {"event": "入队"}, 7]}, handle) self.assertEqual([item["event"] for item in QueueLog(path).recent()], ["入队"]) def test_a_finished_write_leaves_no_temp_files_behind(self): """写入走临时文件加原子替换，界面才不会读到写了一半的 JSON。""" with tempfile.TemporaryDirectory() as directory: log = _log(directory) log.append("abc123", "高瑞@微信", "入队") leftovers = [name for name in os.listdir(directory) if name.endswith(".tmp")] self.assertEqual(leftovers, []) def test_a_failed_write_reports_failure_instead_of_raising(self): with tempfile.TemporaryDirectory() as directory: log = _log(directory) with mock.patch.object(log, "_write_unlocked", side_effect=OSError("盘满了")): self.assertFalse(log.append("abc123", "高瑞@微信", "入队")) def test_clearing_wipes_the_history_and_survives_a_reread(self): with tempfile.TemporaryDirectory() as directory: path = os.path.join(directory, "queue_events.json") log = QueueLog(path) log.append("abc123", "高瑞@微信", "入队") self.assertTrue(log.clear()) self.assertEqual(QueueLog(path).recent(), []) def test_a_novel_length_reply_does_not_bloat_every_row(self): with tempfile.TemporaryDirectory() as directory: log = _log(directory) log.append("abc123", "高瑞@微信", "已发送", "啊" * 5000) self.assertLessEqual(len(log.recent()[0]["detail"]), 200) class BotQueueRecordingTest(TestCase): """机器人把队列里的每一步记下来，且记录出问题时不许影响回复。""" @staticmethod def _bot(directory: str) -> WeChatBot: bot = WeChatBot.__new__(WeChatBot) bot._pending_reply_sessions = {} bot._pending_reply_path = os.path.join(directory, "pending.json") bot._queue_log_instance = _log(directory) return bot def test_a_new_task_writes_one_enqueue_line(self): with tempfile.TemporaryDirectory() as directory: bot = self._bot(directory) fp = b"\x01" * 40 with mock.patch.object(WeChatBot, "_live_render_ids_for", return_value=()), \ mock.patch.object(WeChatBot, "_session_label", return_value="高瑞@微信"), \ mock.patch.object(WeChatBot, "_persist_pending_replies", return_value=True): bot._mark_reply_pending(fp, confirmed_unread=True) entries = bot.queue_log.recent() self.assertEqual([item["event"] for item in entries], ["入队"]) self.assertEqual(entries[0]["name"], "高瑞@微信") def test_the_same_task_touched_again_does_not_queue_twice(self): """一轮里同一个会话会被反复标记，每次都记一条"入队"就没法看了。""" with tempfile.TemporaryDirectory() as directory: bot = self._bot(directory) fp = b"\x01" * 40 with mock.patch.object(WeChatBot, "_live_render_ids_for", return_value=()), \ mock.patch.object(WeChatBot, "_session_label", return_value="高瑞@微信"), \ mock.patch.object(WeChatBot, "_persist_pending_replies", return_value=True): bot._mark_reply_pending(fp, confirmed_unread=True) bot._mark_reply_pending(fp, batch_ready=True) bot._mark_reply_pending(fp) self.assertEqual(len(bot.queue_log.recent()), 1) def test_the_nickname_is_kept_on_the_task_for_later_steps(self): """后面几步都要报出人名，不能每一步再去 OCR 一遍。""" with tempfile.TemporaryDirectory() as directory: bot = self._bot(directory) fp = b"\x01" * 40 with mock.patch.object(WeChatBot, "_live_render_ids_for", return_value=()), \ mock.patch.object(WeChatBot, "_session_label", return_value="高瑞@微信"), \ mock.patch.object(WeChatBot, "_persist_pending_replies", return_value=True): bot._mark_reply_pending(fp) self.assertEqual( bot._pending_reply_sessions[fp.hex()]["display_name"], "高瑞@微信" ) def test_an_unreadable_nickname_is_never_stored_as_the_customer(self): """读不出昵称时 `_session_label` 返回「会话 xxxx」，那是占位不是人名。""" with tempfile.TemporaryDirectory() as directory: bot = self._bot(directory) fp = b"\x01" * 40 with mock.patch.object(WeChatBot, "_live_render_ids_for", return_value=()), \ mock.patch.object(WeChatBot, "_session_label", return_value="会话 0101"), \ mock.patch.object(WeChatBot, "_persist_pending_replies", return_value=True): bot._mark_reply_pending(fp) self.assertNotIn("display_name", bot._pending_reply_sessions[fp.hex()]) def test_a_broken_recorder_never_stops_a_reply(self): with tempfile.TemporaryDirectory() as directory: bot = self._bot(directory) with mock.patch.object( bot._queue_log_instance, "append", side_effect=RuntimeError("炸了") ): bot._log_queue_event(b"\x01" * 40, "已发送", "在的") def test_a_recorded_step_carries_the_session_id_the_ui_shows(self): with tempfile.TemporaryDirectory() as directory: bot = self._bot(directory) fp = WeChatBot._fp_from_name("高瑞@微信") bot._pending_reply_sessions[fp.hex()] = {"display_name": "高瑞@微信"} bot._log_queue_event(fp, "已发送", "在的") self.assertEqual( bot.queue_log.recent()[0]["session_id"], WeChatBot.session_id_of(fp) ) def test_a_recorded_step_is_also_pushed_to_the_floating_capsule(self): with tempfile.TemporaryDirectory() as directory: bot = self._bot(directory) fp = WeChatBot._fp_from_name("高瑞@微信") bot._pending_reply_sessions[fp.hex()] = {"display_name": "高瑞@微信"} bot.progress_cb = mock.Mock() bot._log_queue_event(fp, "开始处理", "已打开会话，正在读取消息") bot.progress_cb.assert_called_once_with("正在读取消息 · 高瑞@微信") def test_an_atomic_operation_is_visible_in_capsule_and_execution_table(self): with tempfile.TemporaryDirectory() as directory: bot = self._bot(directory) fp = WeChatBot._fp_from_name("高瑞@微信") bot._pending_reply_sessions[fp.hex()] = {"display_name": "高瑞@微信"} bot.progress_cb = mock.Mock() bot.report_operation( "会话视觉定位", 2, 4, "检测聊天区、右侧工具抽屉和输入框", fp, ) bot.progress_cb.assert_called_once_with( "会话视觉定位 2/4 · 检测聊天区、右侧工具抽屉和输入框 · 高瑞@微信" ) entry = bot.queue_log.recent()[0] self.assertEqual(entry["event"], "会话视觉定位 2/4") self.assertEqual(entry["detail"], "检测聊天区、右侧工具抽屉和输入框") def test_visual_model_is_named_explicitly_in_capsule_progress(self): self.assertEqual( WeChatBot._queue_event_progress_text( "高瑞@微信", "调用模型", "视觉模型｜客户发来图片" ), "正在调用视觉模型 · 高瑞@微信", ) def test_the_nickname_survives_a_restart(self): """重启后队列界面还要认得出这些任务是谁的。""" with tempfile.TemporaryDirectory() as directory: path = os.path.join(directory, "pending.json") bot = self._bot(directory) bot._pending_reply_path = path fp = b"\x01" * 40 with mock.patch.object(WeChatBot, "_live_render_ids_for", return_value=()), \ mock.patch.object(WeChatBot, "_session_label", return_value="高瑞@微信"): bot._mark_reply_pending(fp, confirmed_unread=True) restored = WeChatBot.__new__(WeChatBot) restored._pending_reply_path = path self.assertEqual( restored._load_pending_replies()[fp.hex()]["display_name"], "高瑞@微信" ) class SendFailureReasonTest(TestCase): """失败要写清是哪一道闸拦的，光说"发送未成功"等于什么都没说。""" @staticmethod def _bot(directory: str, state: dict | None = None) -> WeChatBot: bot = WeChatBot.__new__(WeChatBot) bot._pending_reply_sessions = {(b"\x01" * 40).hex(): state} if state else {} bot._pending_reply_path = os.path.join(directory, "pending.json") bot._queue_log_instance = _log(directory) return bot def test_the_gate_that_blocked_the_send_is_named_in_the_record(self): with tempfile.TemporaryDirectory() as directory: bot = self._bot(directory, {"display_name": "高瑞@微信"}) fp = b"\x01" * 40 bot._deny_send("输入框里有人工草稿「稍等」，已保留不动") bot._log_send_outcome(fp, False) entry = bot.queue_log.recent()[0] self.assertEqual(entry["event"], "失败") self.assertIn("人工草稿", entry["detail"]) def test_a_pressed_enter_awaiting_proof_is_not_branded_a_failure(self): """回车已经按下去，只是还没认出那条消息——多半其实发出去了。""" with tempfile.TemporaryDirectory() as directory: state = {"display_name": "高瑞@微信", "send_state": "uncertain"} bot = self._bot(directory, state) fp = b"\x01" * 40 bot._deny_send("已按下发送但屏幕上还看不到这条，转入后台对账") bot._log_send_outcome(fp, False) self.assertEqual(bot.queue_log.recent()[0]["event"], "发送待核对") def test_a_successful_send_records_what_was_actually_said(self): with tempfile.TemporaryDirectory() as directory: bot = self._bot(directory, {"display_name": "高瑞@微信"}) bot._log_send_outcome(b"\x01" * 40, True, "在的，您慢慢说。") entry = bot.queue_log.recent()[0] self.assertEqual(entry["event"], "已发送") self.assertEqual(entry["detail"], "在的，您慢慢说。") def test_an_unnamed_reason_still_produces_a_usable_line(self): with tempfile.TemporaryDirectory() as directory: bot = self._bot(directory, {"display_name": "高瑞@微信"}) bot._last_send_failure_reason = "" bot._log_send_outcome(b"\x01" * 40, False) self.assertTrue(bot.queue_log.recent()[0]["detail"]) def test_a_new_send_forgets_the_previous_rounds_reason(self): """上一轮的原因留到这一轮，会把人引到早就修好的问题上。""" bot = WeChatBot.__new__(WeChatBot) bot._deny_send("上一轮：检测到人工草稿") bot._security_gate_visible = mock.Mock(return_value=True) bot.send_reply("在的", expected_fp=None) self.assertIn("安全验证", bot.last_send_failure_reason()) def test_every_gate_reason_is_human_readable_not_a_code(self): bot = WeChatBot.__new__(WeChatBot) bot._deny_send("发送限频未过，等待后仍不允许发送") reason = bot.last_send_failure_reason() self.assertNotIn("_", reason) self.assertGreater(len(reason), 4) class ModelCallRecordTest(TestCase): """记录里要看得出模型收到的是什么，否则"答非所问"根本无从查起。""" def test_a_plain_question_is_quoted_in_the_record(self): self.assertEqual( WeChatBot._model_call_summary("在不在", "在不在"), "客户说：在不在" ) def test_an_image_message_says_so_instead_of_looking_empty(self): summary = WeChatBot._model_call_summary("", "", {"image"}) self.assertIn("图片", summary) def test_an_image_with_a_caption_keeps_both(self): summary = WeChatBot._model_call_summary("这个多少钱", "", {"image"}) self.assertIn("图片", summary) self.assertIn("这个多少钱", summary) def test_several_media_kinds_are_all_named(self): summary = WeChatBot._model_call_summary("", "", {"image", "voice"}) self.assertIn("图片", summary) self.assertIn("语音", summary) def test_a_wall_of_text_is_trimmed_to_stay_readable(self): summary = WeChatBot._model_call_summary("啊" * 500, "") self.assertLessEqual(len(summary), 100) self.assertTrue(summary.endswith("…")) def test_a_screenshot_only_round_is_described_rather_than_left_blank(self): self.assertIn("截图", WeChatBot._model_call_summary("", "")) def test_line_breaks_never_break_the_table_row(self): summary = WeChatBot._model_call_summary("第一行\n第二行", "") self.assertNotIn("\n", summary) class GenerationGapTest(TestCase): """模型没交出回复时，队列里不能出现"入队之后再无下文"的断头任务。""" @staticmethod def _bot(directory: str) -> WeChatBot: bot = WeChatBot.__new__(WeChatBot) bot._pending_reply_sessions = {} bot._pending_reply_path = os.path.join(directory, "pending.json") bot._queue_log_instance = _log(directory) return bot def test_a_silent_model_still_leaves_a_line_in_the_record(self): with tempfile.TemporaryDirectory() as directory: bot = self._bot(directory) bot._generation_logged = False bot._log_generation_gap(b"\x01" * 40) self.assertEqual(bot.queue_log.recent()[0]["event"], "跳过") def test_a_model_that_already_explained_itself_is_not_repeated(self): with tempfile.TemporaryDirectory() as directory: bot = self._bot(directory) bot._log_queue_event(b"\x01" * 40, "跳过", "模型判定没有新消息") bot._log_generation_gap(b"\x01" * 40) self.assertEqual(len(bot.queue_log.recent()), 1) def test_a_recorded_reply_also_counts_as_the_model_speaking_up(self): with tempfile.TemporaryDirectory() as directory: bot = self._bot(directory) bot._log_queue_event(b"\x01" * 40, "生成回复", "在的") bot._log_generation_gap(b"\x01" * 40) self.assertEqual(len(bot.queue_log.recent()), 1) class QueuePageReadingTest(TestCase): """队列页面怎么把磁盘上那两个文件讲成人话。""" @classmethod def setUpClass(cls): try: from wechat_gui_qt import QueuePage except Exception as exc: # PySide6 缺失/无显示环境时跳过 raise SkipTest(f"PySide6 不可用: {exc}") cls.page = QueuePage def test_a_task_being_sent_reads_as_sending_not_as_queued(self): self.assertEqual(self.page._state_text({"send_state": "sending"}), "正在发送") def test_a_task_whose_receipt_is_unconfirmed_says_so(self): """这个状态正是"回了没回"说不清的时候，界面上必须看得见。""" self.assertEqual(self.page._state_text({"send_state": "uncertain"}), "发送待核对") def test_a_batched_task_reads_as_generating(self): self.assertEqual(self.page._state_text({"batch_ready": True}), "正在生成回复") def test_a_fresh_unread_task_reads_as_waiting(self): self.assertEqual(self.page._state_text({"confirmed_unread": True}), "等待处理") def test_a_bare_task_still_gets_a_readable_status(self): self.assertEqual(self.page._state_text({}), "排队中") def test_a_short_wait_is_shown_in_seconds(self): import time as _time self.assertTrue(self.page._waited_text(_time.time() - 12).endswith("秒")) def test_a_long_wait_is_shown_in_hours_so_a_stuck_task_stands_out(self): import time as _time self.assertIn("小时", self.page._waited_text(_time.time() - 7200)) def test_a_task_with_no_timestamp_never_shows_a_bogus_wait(self): self.assertEqual(self.page._waited_text(None), "--") class QueuePageTableTest(TestCase): """页面真渲染出来之后，表格里到底是什么。""" @classmethod def setUpClass(cls): try: from PySide6.QtWidgets import QApplication import wechat_gui_qt cls.app = QApplication.instance() or QApplication([]) cls.module = wechat_gui_qt except Exception as exc: # PySide6 缺失/无显示环境时跳过 raise SkipTest(f"PySide6 不可用: {exc}") def _page(self, directory: str, pending: dict, log: QueueLog): from pathlib import Path with open(os.path.join(directory, "pending_replies.json"), "w", encoding="utf-8") as handle: json.dump(pending, handle, ensure_ascii=False) with mock.patch.object(self.module, "SCRIPT_DIR", Path(directory)), \ mock.patch("queue_log.QueueLog", lambda *a, **k: log): page = self.module.QueuePage() page.refresh_data() return page @staticmethod def _column(table, column: int) -> list[str]: return [ table.item(row, column).text() if table.item(row, column) else "" for row in range(table.rowCount()) ] def test_the_queue_is_shown_oldest_first_so_nobody_is_starved(self): """界面上的顺序就是实际服务顺序，先来的必须排在前面。""" with tempfile.TemporaryDirectory() as directory: now = time.time() pending = { "aa" * 20: {"display_name": "后来的", "created_at": now - 5}, "bb" * 20: {"display_name": "先来的", "created_at": now - 500}, } page = self._page(directory, pending, _log(directory)) self.assertEqual(self._column(page.queue_table, 1), ["先来的", "后来的"]) self.assertEqual(page.waiting_metric.value.text(), "2") def test_an_empty_queue_says_so_instead_of_showing_a_blank_slab(self): with tempfile.TemporaryDirectory() as directory: page = self._page(directory, {}, _log(directory)) self.assertEqual(page.queue_table.rowCount(), 1) self.assertIn("没有排队", page.queue_table.item(0, 0).text()) self.assertEqual(page.waiting_metric.value.text(), "0") def test_a_queue_that_drains_replaces_its_rows_with_the_empty_notice(self): """刷新是原地重画的，上一轮的行不清干净就会留在屏幕上骗人。""" with tempfile.TemporaryDirectory() as directory: from pathlib import Path log = _log(directory) path = os.path.join(directory, "pending_replies.json") page = self._page( directory, {"aa" * 20: {"display_name": "高瑞@微信", "created_at": time.time()}}, log, ) self.assertEqual(page.queue_table.rowCount(), 1) with open(path, "w", encoding="utf-8") as handle: json.dump({}, handle) with mock.patch.object(self.module, "SCRIPT_DIR", Path(directory)), \ mock.patch("queue_log.QueueLog", lambda *a, **k: log): page.refresh_data() self.assertIn("没有排队", page.queue_table.item(0, 0).text()) def test_a_task_whose_nickname_never_read_still_takes_a_visible_slot(self): """认不出是谁也得让人看见它在排队，否则就成了看不见的积压。""" with tempfile.TemporaryDirectory() as directory: pending = {"cc" * 20: {"created_at": time.time(), "send_state": "uncertain"}} page = self._page(directory, pending, _log(directory)) self.assertEqual(page.queue_table.rowCount(), 1) self.assertIn("未识别", page.queue_table.item(0, 1).text()) def test_todays_sends_and_failures_are_counted_from_the_history(self): with tempfile.TemporaryDirectory() as directory: log = _log(directory) for event in ("已发送", "已发送", "失败", "跳过", "入队"): log.append("abc", "高瑞@微信", event) page = self._page(directory, {}, log) self.assertEqual(page.sent_metric.value.text(), "2") self.assertEqual(page.failed_metric.value.text(), "1") def test_yesterdays_sends_do_not_inflate_todays_count(self): with tempfile.TemporaryDirectory() as directory: path = os.path.join(directory, "queue_events.json") stale = time.time() - 3 * 24 * 3600 with open(path, "w", encoding="utf-8") as handle: json.dump( {"events": [{"ts": stale, "name": "高瑞@微信", "event": "已发送"}]}, handle, ) page = self._page(directory, {}, QueueLog(path)) self.assertEqual(page.history_table.rowCount(), 1) self.assertEqual(page.sent_metric.value.text(), "0") def test_a_task_with_a_junk_timestamp_still_appears_in_the_queue(self): """一条坏数据不能让整页读不出来——那正是最需要看它的时候。""" with tempfile.TemporaryDirectory() as directory: pending = { "aa" * 20: {"display_name": "正常的", "created_at": time.time()}, "bb" * 20: {"display_name": "坏时间", "created_at": "昨天"}, } page = self._page(directory, pending, _log(directory)) self.assertEqual(self._column(page.queue_table, 1), ["正常的", "坏时间"]) self.assertEqual(page.queue_table.item(1, 3).text(), "--") def test_a_stopped_listener_explains_why_the_wait_keeps_growing(self): """停着的时候等待时长照涨，不说明白会被当成机器人卡死了。""" with tempfile.TemporaryDirectory() as directory: pending = {"aa" * 20: {"display_name": "高瑞@微信", "created_at": time.time()}} page = self._page(directory, pending, _log(directory)) page.set_running(False) self.assertTrue(page.queue_notice.isVisibleTo(page)) self.assertIn("监听已停止", page.queue_notice.text()) def test_a_running_listener_does_not_nag(self): with tempfile.TemporaryDirectory() as directory: pending = {"aa" * 20: {"display_name": "高瑞@微信", "created_at": time.time()}} page = self._page(directory, pending, _log(directory)) page.set_running(True) self.assertFalse(page.queue_notice.isVisibleTo(page)) def test_an_empty_queue_never_shows_the_stopped_warning(self): with tempfile.TemporaryDirectory() as directory: page = self._page(directory, {}, _log(directory)) page.set_running(False) self.assertFalse(page.queue_notice.isVisibleTo(page)) def test_a_task_awaiting_proof_is_coloured_apart_from_a_real_failure(self): with tempfile.TemporaryDirectory() as directory: log = _log(directory) log.append("abc", "高瑞@微信", "发送待核对", "已按下发送，正在核对") log.append("abc", "高瑞@微信", "失败", "输入框里有人工草稿") page = self._page(directory, {}, log) colours = { page.history_table.item(row, 2).text(): page.history_table.item(row, 2).foreground().color().name() for row in range(page.history_table.rowCount()) } self.assertNotEqual(colours["发送待核对"], colours["失败"]) def test_awaiting_proof_is_not_counted_as_a_failure_today(self): """把"多半已经发出去了"记成失败，会让人以为出了大问题。""" with tempfile.TemporaryDirectory() as directory: log = _log(directory) log.append("abc", "高瑞@微信", "发送待核对", "已按下发送，正在核对") page = self._page(directory, {}, log) self.assertEqual(page.failed_metric.value.text(), "0") def test_an_empty_history_says_so_too(self): with tempfile.TemporaryDirectory() as directory: page = self._page(directory, {}, _log(directory)) self.assertEqual(page.history_table.rowCount(), 1) self.assertIn("还没有", page.history_table.item(0, 0).text()) def test_a_missing_pending_file_is_not_reported_as_an_error(self): """第一次启动时这个文件根本不存在，不该在日志里吓人一跳。""" from pathlib import Path with tempfile.TemporaryDirectory() as directory: log = _log(directory) complaints = [] with mock.patch.object(self.module, "SCRIPT_DIR", Path(directory)), \ mock.patch("queue_log.QueueLog", lambda *a, **k: log): page = self.module.QueuePage() page.logMessage.connect(lambda text, tag: complaints.append(text)) page.refresh_data() self.assertEqual(complaints, [])