# -*- coding: utf-8 -*- """会话身份改用昵称之后，身份必须稳：同一个人永远同一个 ID。""" import io from contextlib import redirect_stdout from unittest import TestCase, main, mock import numpy as np import session_name from session_name import NameReader, crop_to_ink, ink_bounds, normalize_name, session_id_for from test_support import queue_log_redirect def setUpModule(): queue_log_redirect.start() def tearDownModule(): queue_log_redirect.stop() def panel_with_text(width=200, height=40, box=(20, 10, 120, 30)) -> np.ndarray: """浅底深字的一块面板，box 是那行字占的矩形。""" image = np.full((height, width, 3), 245, dtype=np.uint8) x1, y1, x2, y2 = box image[y1:y2, x1:x2] = 30 return image class FakeEngine: """替身 OCR：按调用顺序吐出预先排好的（文本, 置信度）。""" def __init__(self, *results): self.results = list(results) self.calls = [] def __call__(self, image, **kwargs): self.calls.append((image.shape, kwargs)) if not self.results: return None, None text, score = self.results.pop(0) return [[text, score]], None def reader_with(*results) -> NameReader: reader = NameReader() reader._engine = FakeEngine(*results) return reader class NameNormalisationTest(TestCase): def test_the_title_and_the_list_row_agree_on_one_person(self): # 实测标题读出来带空格、列表行不带。不抹平的话同一个人两个 ID， # 又回到一人多档、客户收不到回复的老问题 self.assertEqual(normalize_name("高瑞 @微信"), normalize_name("高瑞@微信")) self.assertEqual( session_id_for("高瑞 @微信"), session_id_for("高瑞@微信") ) def test_full_width_characters_fold_to_the_same_id(self): self.assertEqual(session_id_for("高瑞＠微信"), session_id_for("高瑞@微信")) def test_the_id_is_a_plain_md5_hex(self): sid = session_id_for("高瑞@微信") self.assertEqual(len(sid), 32) self.assertTrue(all(ch in "0123456789abcdef" for ch in sid)) def test_two_different_people_never_share_an_id(self): self.assertNotEqual( session_id_for("高瑞@微信"), session_id_for("一个小迷糊@微信") ) def test_no_name_means_no_id(self): # 读不到名字时给个空 ID，调用方好判断"这一轮先别认"， # 而不是拿空字符串的 md5 去当成某个真实会话 self.assertEqual(session_id_for(""), "") self.assertEqual(session_id_for(" "), "") class InkCropTest(TestCase): def test_it_finds_the_line_of_text_in_the_panel(self): image = panel_with_text(box=(20, 10, 120, 30)) self.assertEqual(ink_bounds(image), (20, 10, 120, 30)) def test_a_blank_panel_has_no_text(self): blank = np.full((40, 200, 3), 245, dtype=np.uint8) self.assertIsNone(ink_bounds(blank)) self.assertIsNone(crop_to_ink(blank)) def test_the_crop_keeps_a_margin_around_the_glyphs(self): # 贴着字裁会切掉首字的一竖，实测「一个小迷糊@微信」会读成「个小迷糊@微信」 crop = crop_to_ink(panel_with_text(box=(20, 10, 120, 30)), pad=5) self.assertEqual(crop.shape[1], 110) self.assertEqual(crop.shape[0], 30) def test_the_crop_never_runs_off_the_panel(self): # 字顶到左右边框时，留白不能把裁剪框推到画面外面去 crop = crop_to_ink(panel_with_text(box=(0, 12, 200, 28)), pad=8) self.assertEqual(crop.shape[:2], (32, 200)) def test_a_dark_themed_panel_still_yields_its_text(self): image = np.full((40, 200, 3), 30, dtype=np.uint8) image[10:30, 20:120] = 240 self.assertEqual(ink_bounds(image), (20, 10, 120, 30)) class NameReadingTest(TestCase): def test_full_page_layout_ocr_returns_text_with_coordinates(self): class LayoutEngine: def __init__(self): self.kwargs = None def __call__(self, image, **kwargs): self.kwargs = kwargs return [ [[[10, 20], [70, 20], [70, 42], [10, 42]], "消息", 0.96], [[[80, 80], [150, 80], [150, 102], [80, 102]], "噪声", 0.20], ], None reader = NameReader() reader._engine = LayoutEngine() items = reader.read_layout(np.zeros((160, 240, 3), dtype=np.uint8)) self.assertEqual( items, [ { "text": "消息", "score": 0.96, "x1": 10, "y1": 20, "x2": 70, "y2": 42, } ], ) self.assertTrue(reader._engine.kwargs["use_det"]) self.assertFalse(reader._engine.kwargs["use_cls"]) def test_a_confident_read_becomes_the_session_identity(self): reader = reader_with(("高瑞@微信", 0.94)) self.assertEqual(reader.read(panel_with_text()), "高瑞@微信") def test_the_detector_is_skipped_so_it_stays_fast_enough_to_poll(self): # 带检测要 1.6 秒，放不进轮询；框自己算好了就别让模型再找一遍 reader = reader_with(("高瑞@微信", 0.94)) reader.read(panel_with_text()) _shape, kwargs = reader._engine.calls[0] self.assertFalse(kwargs["use_det"]) self.assertFalse(kwargs["use_cls"]) def test_a_shaky_read_is_refused_rather_than_guessed(self): # 实测「打卡」被读成「江卡」只有 0.708。宁可这一轮不认， # 也不能凭一个错名字开出一份新档案 reader = reader_with(("江卡", 0.70)) self.assertEqual(reader.read(panel_with_text()), "") def test_a_blank_panel_never_reaches_the_engine(self): reader = reader_with(("不该被调用", 0.99)) blank = np.full((40, 200, 3), 245, dtype=np.uint8) self.assertEqual(reader.read(blank), "") self.assertEqual(reader._engine.calls, []) def test_the_same_screen_is_only_read_once(self): # 同一块画面在一轮里会被反复问到；每次都重识别既慢，又可能这次读对 # 下次读错，会话身份就在同一轮里跳变 reader = reader_with(("高瑞@微信", 0.94)) image = panel_with_text() first = reader.read(image, signature=b"same-screen") second = reader.read(image, signature=b"same-screen") self.assertEqual(first, "高瑞@微信") self.assertEqual(second, "高瑞@微信") self.assertEqual(len(reader._engine.calls), 1) def test_a_different_screen_is_read_again(self): reader = reader_with(("高瑞@微信", 0.94), ("一个小迷糊@微信", 0.98)) image = panel_with_text() self.assertEqual(reader.read(image, signature=b"screen-a"), "高瑞@微信") self.assertEqual(reader.read(image, signature=b"screen-b"), "一个小迷糊@微信") def test_a_refused_read_is_not_cached_as_an_answer(self): # 低置信度不该被记成"这块画面就是没名字"，下一轮画面清楚了要能读出来 reader = reader_with(("江卡", 0.70), ("打卡记录", 0.95)) image = panel_with_text() self.assertEqual(reader.read(image, signature=b"row"), "") self.assertEqual(reader.read(image, signature=b"row"), "打卡记录") def test_the_cache_is_capped_so_a_day_long_run_cannot_eat_memory(self): # 每一行的未读态/选中态/悬停态各占一个条目，机器人一跑就是一整天 reader = NameReader() reader._engine = FakeEngine( *[("高瑞@微信", 0.94)] * (session_name.MAX_CACHE_ENTRIES + 40) ) image = panel_with_text() for index in range(session_name.MAX_CACHE_ENTRIES + 30): reader.read(image, signature=f"screen-{index}".encode()) self.assertLessEqual(len(reader._cache), session_name.MAX_CACHE_ENTRIES) self.assertGreater(len(reader._cache), 0) def test_a_missing_engine_reports_no_name_instead_of_crashing(self): reader = NameReader() reader._engine_failed = True self.assertFalse(reader.available) self.assertEqual(reader.read(panel_with_text()), "") def test_an_engine_that_throws_does_not_take_the_round_down(self): class Exploding: def __call__(self, image, **kwargs): raise RuntimeError("模型崩了") reader = NameReader() reader._engine = Exploding() with redirect_stdout(io.StringIO()): self.assertEqual(reader.read(panel_with_text()), "") class MisreadCorrectionTest(TestCase): def test_a_one_character_slip_is_pulled_back_to_the_known_contact(self): # 「行业资讯」被读成「亏业资讯」。不纠的话这一轮就成了另一个客户， # 上下文断掉、档案分家 reader = NameReader() reader.remember("行业资讯") with redirect_stdout(io.StringIO()): self.assertEqual(reader.canonical("亏业资讯"), "行业资讯") def test_a_genuinely_new_contact_is_not_folded_into_an_old_one(self): reader = NameReader() reader.remember("高瑞@微信") self.assertEqual(reader.canonical("一个小迷糊@微信"), "一个小迷糊@微信") def test_two_characters_apart_is_too_far_to_assume(self): reader = NameReader() reader.remember("甄养堂助理-江莉莉") self.assertEqual(reader.canonical("甄养堂助理-王菲菲"), "甄养堂助理-王菲菲") def test_short_names_are_never_corrected(self): # 「小王」和「小李」也只差一个字，但那是两个人 reader = NameReader() reader.remember("小王") self.assertEqual(reader.canonical("小李"), "小李") def test_the_contact_seen_most_often_wins_a_tie(self): # 偶发误读只会出现一两次，常客才是真身 reader = NameReader() reader._known = {"客户联系": 40, "客尸联系": 1} with redirect_stdout(io.StringIO()): self.assertEqual(reader.canonical("客户联糸"), "客户联系") def test_an_exact_known_name_is_returned_untouched(self): reader = NameReader() reader.remember("高瑞@微信") self.assertEqual(reader.canonical("高瑞@微信"), "高瑞@微信") def test_correction_keeps_the_id_stable_across_a_slip(self): # 这才是纠正的意义：读错一个字，会话 ID 依然不变 reader = NameReader() reader.remember("行业资讯") with redirect_stdout(io.StringIO()): corrected = reader.canonical("亏业资讯") self.assertEqual(session_id_for(corrected), session_id_for("行业资讯")) def test_a_corrected_read_does_not_register_the_wrong_spelling(self): reader = NameReader() reader.remember("行业资讯") with redirect_stdout(io.StringIO()): reader.read(panel_with_text(), signature=b"") reader._engine = FakeEngine(("亏业资讯", 0.89)) with redirect_stdout(io.StringIO()): reader.read(panel_with_text()) self.assertNotIn("亏业资讯", reader.known_names()) class BotIdentityReadingTest(TestCase): """机器人读「当前打开的是谁」，是进度提示和会话身份共同的入口。""" def _bot(self): from wechat_bot import WeChatBot bot = WeChatBot.__new__(WeChatBot) bot._progress_text = "" bot.progress_cb = None return bot def test_the_open_chat_is_named_from_its_title(self): bot = self._bot() bot._name_reader_instance = reader_with(("高瑞@微信", 0.94)) bot._chat_title_panel = lambda: panel_with_text() bot._chat_identity_signature = lambda: b"title-sig" self.assertEqual(bot._open_chat_display_name(), "高瑞@微信") def test_the_title_pixels_key_the_cache_so_one_chat_reads_once(self): # 同一个会话在一轮里会被问到好几次，每次都重识别既慢、又可能这次读对 # 下次读错，身份就在同一轮里跳变 bot = self._bot() reader = reader_with(("高瑞@微信", 0.94)) bot._name_reader_instance = reader bot._chat_title_panel = lambda: panel_with_text() bot._chat_identity_signature = lambda: b"title-sig" bot._open_chat_display_name() bot._open_chat_display_name() self.assertEqual(len(reader._engine.calls), 1) def test_a_missing_title_panel_names_nobody(self): bot = self._bot() bot._name_reader_instance = reader_with(("不该被调用", 0.99)) bot._chat_title_panel = lambda: None bot._chat_identity_signature = lambda: b"" self.assertEqual(bot._open_chat_display_name(), "") def test_the_progress_line_falls_back_to_a_fingerprint_stub(self): # 读不出名字时进度提示也不能变成空白，用户会以为它死了 bot = self._bot() bot._chat_title_panel = lambda: None bot._chat_identity_signature = lambda: b"" bot._name_reader_instance = reader_with() self.assertEqual(bot._session_label(bytes.fromhex("ab" * 20)), "会话 abababab") def test_the_progress_line_prefers_the_nickname(self): bot = self._bot() bot._name_reader_instance = reader_with(("一个小迷糊@微信", 0.98)) bot._chat_title_panel = lambda: panel_with_text() bot._chat_identity_signature = lambda: b"title-sig" self.assertEqual(bot._session_label(b"\x01" * 40), "一个小迷糊@微信") class RowIdentityTest(TestCase): """会话身份改由昵称派生：同一行在任何渲染下都必须是同一个人。""" def _bot(self, name_by_row=None): from wechat_bot import WeChatBot bot = WeChatBot.__new__(WeChatBot) bot.identity_by_name = True bot.scale = 1.0 bot.session_item_h = 64 bot._session_identity_trustworthy = lambda: True if name_by_row is not None: bot._row_display_name = ( lambda img, rel_y, row_center=False: name_by_row(rel_y) ) return bot @staticmethod def _row_image(width=230, height=64, bold=False, selected=False): """一行会话。未读是加粗、选中是蓝底白字——像素差别正是老 bug 的根源。""" background = (230, 120, 40) if selected else (245, 245, 245) image = np.full((height, width, 3), background, dtype=np.uint8) ink = 255 if selected else (10 if bold else 60) image[16:30, 70:150] = ink if bold: image[16:32, 70:154] = ink image[8:24, 210:226] = (60, 60, 220) # 未读红点 return image def test_the_same_contact_matches_across_unread_and_selected_renderings(self): # 这就是现场那句「实际打开对象与目标不一致，已停止后续发送」： # 记下目标时它是未读（加粗+红点），点开后它是选中（蓝底白字）， # 像素哈希算出两个身份，消息复制到了也不敢发 bot = self._bot(lambda rel_y: "一个小迷糊@微信") unread_fp = bot._session_fingerprint( self._row_image(bold=True), 32, row_center=True ) selected_fp = bot._session_fingerprint( self._row_image(selected=True), 32, row_center=True ) self.assertTrue(unread_fp) self.assertEqual(unread_fp, selected_fp) self.assertTrue(bot._session_fp_matches(unread_fp, selected_fp)) def test_two_different_contacts_stay_apart(self): bot = self._bot(lambda rel_y: "高瑞@微信" if rel_y < 50 else "一个小迷糊@微信") first = bot._session_fingerprint(self._row_image(), 32, row_center=True) second = bot._session_fingerprint(self._row_image(), 96, row_center=True) self.assertNotEqual(first, second) self.assertFalse(bot._session_fp_matches(first, second)) def test_the_session_id_is_exactly_the_nickname_md5(self): from wechat_bot import WeChatBot fp = WeChatBot._fp_from_name("高瑞@微信") self.assertEqual(len(fp), 40) self.assertEqual(WeChatBot.session_id_of(fp), session_id_for("高瑞@微信")) def test_the_title_and_the_row_produce_one_identity(self): # 标题里带空格、列表行不带；归一之后必须是同一个人 from wechat_bot import WeChatBot self.assertEqual( WeChatBot._fp_from_name("高瑞 @微信"), WeChatBot._fp_from_name("高瑞@微信"), ) def test_no_nickname_means_no_identity(self): from wechat_bot import WeChatBot self.assertEqual(WeChatBot._fp_from_name(""), b"") self.assertEqual(WeChatBot._fp_from_name(" "), b"") def test_an_unreadable_row_falls_back_to_the_pixel_path(self): # OCR 读不出来（空行、引擎缺失）时不能整个瘫掉，还得按老办法认 bot = self._bot(lambda rel_y: "") fallback = mock.Mock(return_value=b"pixel-path") bot._canonical_fp = fallback bot._raw_session_fingerprint = mock.Mock(return_value=b"raw") bot._session_name_fingerprint = mock.Mock(return_value=b"") bot._session_fingerprint(self._row_image(), 32, row_center=True) fallback.assert_called_once() def test_the_pixel_path_is_untouched_when_the_switch_is_off(self): bot = self._bot(lambda rel_y: "高瑞@微信") bot.identity_by_name = False bot._canonical_fp = mock.Mock(return_value=b"a" * 8) bot._raw_session_fingerprint = mock.Mock(return_value=b"raw") bot._session_name_fingerprint = mock.Mock(return_value=b"n" * 32) bot._canonical_session_fp = lambda raw: raw bot._session_render_fingerprint = mock.Mock(return_value=b"") fp = bot._session_fingerprint(self._row_image(), 32, row_center=True) self.assertEqual(fp, b"a" * 8 + b"n" * 32) class RowNameCropTest(TestCase): def _bot(self, width=460, item_h=128): from wechat_bot import WeChatBot bot = WeChatBot.__new__(WeChatBot) bot.scale = 2.0 bot.session_item_h = item_h bot._avatar_anchor = lambda img, y: y return bot def test_the_crop_skips_the_avatar_and_stops_before_the_timestamp(self): # 实测 460px 宽的列表：头像占到 112，昵称 136..318，时间戳更靠右。 # 时间戳一旦进框，「1分钟前」变「12:08」就会让同一个人换 md5 bot = self._bot() image = np.zeros((1208, 460, 3), dtype=np.uint8) panel = bot._row_name_panel(image, 78, row_center=True) self.assertIsNotNone(panel) left = int(460 * 0.26) right = int(460 * 0.78) self.assertEqual(panel.shape[1], right - left) self.assertGreater(left, 112) # 头像在外 self.assertLess(left, 136) # 昵称起点没被切 self.assertGreater(right, 318) # 最长的昵称还在 self.assertLess(right, 460) # 时间戳在外 def test_the_crop_follows_a_narrow_sidebar_instead_of_fixed_pixels(self): # 侧栏宽度是动态识别出来的，写死像素在窄侧栏下会切掉字 narrow = self._bot() image = np.zeros((600, 300, 3), dtype=np.uint8) panel = narrow._row_name_panel(image, 78, row_center=True) self.assertEqual(panel.shape[1], int(300 * 0.78) - int(300 * 0.26)) def test_the_crop_takes_the_name_line_not_the_message_preview(self): bot = self._bot() image = np.zeros((1208, 460, 3), dtype=np.uint8) panel = bot._row_name_panel(image, 200, row_center=True) self.assertLessEqual(panel.shape[0], int(128 * 0.34)) def test_a_row_too_small_to_hold_a_name_is_refused(self): bot = self._bot(item_h=8) image = np.zeros((40, 20, 3), dtype=np.uint8) self.assertIsNone(bot._row_name_panel(image, 10, row_center=True)) def test_a_missing_image_never_raises(self): bot = self._bot() self.assertIsNone(bot._row_name_panel(None, 10, row_center=True)) class ChatEdgeRuleTest(TestCase): """聊天区左边缘那条分隔线，是「对着同一句话连回三遍」的真正根因。""" @staticmethod def _grid(rows=36, cols=48): return np.zeros((rows, cols), dtype=bool) def _bot(self): from wechat_bot import WeChatBot return WeChatBot def test_a_full_height_line_at_the_left_edge_is_erased(self): # 会话列表和聊天区之间的分隔线被裁进了聊天区，网格最左一列每行都有内容 grid = self._grid() grid[3:, 0] = True cleaned = self._bot()._without_edge_rules(grid) self.assertFalse(cleaned.any()) def test_the_bottom_most_message_is_found_again_once_the_line_is_gone(self): # 有这条线时，「最底下有内容的行」永远是最后一行；判定区间落在空白处， # 只看见左边那条线，于是任何会话都被判成「最后一条是客户发的」 grid = self._grid() grid[3:, 0] = True # 边缘竖线 grid[17:21, 26:48] = True # 我方气泡，靠右 cleaned = self._bot()._without_edge_rules(grid) bottom = int(np.where(cleaned.any(axis=1))[0][-1]) self.assertEqual(bottom, 20) def test_a_real_bubble_touching_the_left_edge_is_kept(self): # 客户消息本来就靠左，不能连它一起抹掉 grid = self._grid() grid[17:21, 0:22] = True cleaned = self._bot()._without_edge_rules(grid) self.assertTrue(cleaned[17:21, 0].all()) def test_a_long_message_spanning_the_screen_is_not_mistaken_for_a_rule(self): # 中间列再长也不清；只有贴边且几乎贯穿全高的才算 UI 线条 grid = self._grid() grid[:, 20] = True cleaned = self._bot()._without_edge_rules(grid) self.assertTrue(cleaned[:, 20].all()) def test_a_rule_on_the_right_edge_is_erased_too(self): grid = self._grid() grid[:, -1] = True cleaned = self._bot()._without_edge_rules(grid) self.assertFalse(cleaned.any()) def test_an_empty_grid_survives_untouched(self): cleaned = self._bot()._without_edge_rules(self._grid()) self.assertFalse(cleaned.any()) def test_a_degenerate_grid_never_raises(self): bot = self._bot() self.assertIsNone(bot._without_edge_rules(None)) tiny = np.zeros((2, 4), dtype=bool) self.assertEqual(bot._without_edge_rules(tiny).shape, (2, 4)) def test_the_original_grid_is_not_mutated(self): grid = self._grid() grid[:, 0] = True self._bot()._without_edge_rules(grid) self.assertTrue(grid[:, 0].all()) class NoRepeatReplyTest(TestCase): """一条来消息只回一次。现场出过对着同一句「你好」连回三条的事故。""" def _bot(self, last_bubble_outgoing, pending=None): from wechat_bot import WeChatBot bot = WeChatBot.__new__(WeChatBot) bot._last_visible_bubble_is_outgoing = lambda: last_bubble_outgoing bot._pending_reply_sessions = dict(pending or {}) bot._pending_exchanges = {} bot._clear_reply_pending = mock.Mock() bot._commit_staged_exchange = mock.Mock() bot._remember_active_surface = mock.Mock() bot._pending_reply_state = lambda fp: bot._pending_reply_sessions.get(fp.hex()) return bot FP = bytes.fromhex("ab" * 20) def test_our_own_reply_on_screen_stops_another_one(self): # 客户只发了一句「你好」，回执没确认，任务留在队列里 —— 这时候绝不能 # 再生成一条措辞不同的问候发出去 bot = self._bot(True, {self.FP.hex(): {"staged_reply_text": "你好呀，我在呢"}}) with redirect_stdout(io.StringIO()): self.assertTrue(bot._nothing_left_to_answer(self.FP, "你好\n你好呀，我在呢")) def test_seeing_our_reply_settles_the_task_instead_of_retrying_forever(self): # 自己写的那句就在屏幕上，发送成功是确凿的：落档、清任务，别再重试 bot = self._bot(True, {self.FP.hex(): {"staged_reply_text": "你好呀，我在呢"}}) with redirect_stdout(io.StringIO()): bot._nothing_left_to_answer(self.FP, "你好\n你好呀，我在呢") bot._commit_staged_exchange.assert_called_once_with(self.FP.hex()) bot._clear_reply_pending.assert_called_once_with(self.FP) def test_a_human_reply_also_ends_the_task_without_archiving_our_draft(self): # 右边最后那句不是我们写的，说明人工接手回过了 bot = self._bot(True, {self.FP.hex(): {"staged_reply_text": "机器人拟的稿"}}) with redirect_stdout(io.StringIO()): self.assertTrue(bot._nothing_left_to_answer(self.FP, "你好\n我是人工，这就帮您看")) bot._commit_staged_exchange.assert_not_called() bot._clear_reply_pending.assert_called_once_with(self.FP) def test_a_waiting_customer_still_gets_answered(self): bot = self._bot(False, {self.FP.hex(): {"staged_reply_text": "上一轮的回复"}}) self.assertFalse(bot._nothing_left_to_answer(self.FP, "你好")) bot._clear_reply_pending.assert_not_called() def test_an_unreadable_bubble_side_never_silences_the_bot(self): # 宽气泡、深色主题判不出左右时宁可多回一句，不能因为看不清就把客户晾着 bot = self._bot(None, {self.FP.hex(): {"staged_reply_text": "上一轮的回复"}}) self.assertFalse(bot._nothing_left_to_answer(self.FP, "你好")) def test_an_idle_chat_with_no_task_is_simply_left_alone(self): bot = self._bot(True) self.assertTrue(bot._nothing_left_to_answer(self.FP, "")) bot._clear_reply_pending.assert_not_called() def test_the_guard_survives_a_blank_transcript(self): # 剪贴板没抓到文字时也不能把"我方最后发言"当成要回复 bot = self._bot(True, {self.FP.hex(): {"staged_reply_text": "你好呀"}}) with redirect_stdout(io.StringIO()): self.assertTrue(bot._nothing_left_to_answer(self.FP, "")) bot._commit_staged_exchange.assert_not_called() bot._clear_reply_pending.assert_called_once_with(self.FP) class ProgressReportingTest(TestCase): def _bot(self): from wechat_bot import WeChatBot bot = WeChatBot.__new__(WeChatBot) bot._progress_text = "" return bot def test_the_same_line_is_not_sent_twice(self): # 轮询每秒都在跑，重复内容会把界面刷爆 bot = self._bot() seen = [] bot.progress_cb = seen.append bot.report_progress("正在回复 高瑞@微信") bot.report_progress("正在回复 高瑞@微信") bot.report_progress("已回复 高瑞@微信") self.assertEqual(seen, ["正在回复 高瑞@微信", "已回复 高瑞@微信"]) def test_a_broken_callback_never_takes_the_round_down(self): bot = self._bot() def explode(_text): raise RuntimeError("界面没了") bot.progress_cb = explode bot.report_progress("正在回复 高瑞@微信") def test_a_headless_run_reports_to_nobody_without_complaining(self): bot = self._bot() bot.progress_cb = None bot.report_progress("正在回复 高瑞@微信") def test_a_partially_built_bot_still_accepts_progress(self): # 不少代码路径用 __new__ 造机器人，绕过了 __init__ from wechat_bot import WeChatBot bot = WeChatBot.__new__(WeChatBot) seen = [] bot.progress_cb = seen.append bot.report_progress("等待新消息") self.assertEqual(seen, ["等待新消息"]) class CorrectionDistanceTest(TestCase): def test_the_distance_helper_stops_early_instead_of_scanning_everything(self): far = session_name._edit_distance("abcdefgh", "zzzzzzzz", 1) self.assertGreater(far, 1) def test_identical_strings_are_zero_apart(self): self.assertEqual(session_name._edit_distance("高瑞@微信", "高瑞@微信", 1), 0) if __name__ == "__main__": main()