"""Local-only search regression tests; no account, network or message sending.""" import sqlite3 import tempfile import types import unittest from contextlib import closing from pathlib import Path from unittest.mock import Mock, patch import wxwork_message_browser as browser import wechat_gui_qt as gui class BrowserSearchTest(unittest.TestCase): def setUp(self): self.root = Path(self.enterContext(tempfile.TemporaryDirectory())) browser.clear_browser_cache() for account in ('1', '2'): folder = self.root / account folder.mkdir() with closing(sqlite3.connect(folder / 'message.db')) as db, db: db.execute('CREATE TABLE message_table(conversation_id, sender_id, send_time, content_type, content, server_id)') db.executemany('INSERT INTO message_table VALUES(?,?,?,?,?,?)', [ (f'M:{n}', str(n), n, 2, f'{account} needle {n}', str(n)) for n in range(120)]) with closing(sqlite3.connect(folder / 'user.db')) as db, db: db.execute('CREATE TABLE user_table(id, name)') db.executemany('INSERT INTO user_table VALUES(?,?)', [(str(n), f'客户{n}') for n in range(120)]) def search(self, **kwargs): return browser.search_browser_conversations(selected_account='1', cache_roots=[self.root], **kwargs) def test_search_never_loads_chat_media_or_initializes_keys(self): with patch.object(browser, '_messages', side_effect=AssertionError('chat loaded')), \ patch.object(browser, '_refresh_cache', side_effect=AssertionError('decryption')), \ patch.object(browser, 'detect_wxwork_dir', side_effect=AssertionError('source discovery')): result = self.search(query='客户11') self.assertEqual(result['conversationTotal'], 11) self.assertFalse(result['conversationHasMore']) self.assertNotIn('messages', result) self.assertEqual(self.search(query='2 needle')['conversations'], []) self.assertEqual(self.search(query='absent')['conversationTotal'], 0) def test_filtered_pagination_and_clear(self): result = self.search(query='needle', conversation_limit=50) self.assertEqual(len(result['conversations']), 50) self.assertEqual(result['conversationTotal'], 120) self.assertTrue(result['conversationHasMore']) result = self.search(query='needle', conversation_limit=150) self.assertEqual(len(result['conversations']), 120) self.assertFalse(result['conversationHasMore']) self.assertEqual(self.search(query='')['conversationTotal'], 120) def test_new_database_rows_invalidate_search_index(self): self.assertEqual(self.search(query='fresh')['conversationTotal'], 0) with closing(sqlite3.connect(self.root / '1/message.db')) as db, db: db.execute('INSERT INTO message_table VALUES(?,?,?,?,?,?)', ('M:new', 'new', 1000, 2, 'fresh', 'new')) self.assertEqual(self.search(query='fresh')['conversationTotal'], 1) def test_refresh_chat_outside_filter_keeps_exact_conversation(self): state = browser.load_browser_snapshot(selected_account='1', selected_conversation='M:2', query='客户119', cache_roots=[self.root], message_limit=40) self.assertEqual(state['selectedConversation'], 'M:2') self.assertEqual(state['conversationDetails']['id'], 'M:2') self.assertEqual(state['messages'][0]['content'], '1 needle 2') self.assertEqual(state['conversations'][0]['id'], 'M:119') def test_images_share_one_scan_until_expiry_or_refresh(self): media = self.root / 'Cache' media.mkdir() (media / 'first.png').write_bytes(b'\x89PNG\r\n\x1a\n') with patch.object(browser.time, 'monotonic', return_value=100), \ patch.object(Path, 'rglob', autospec=True, side_effect=Path.rglob) as scan: self.assertIn('first.png', browser._media_index(media)) browser._media_index(media) self.assertEqual(scan.call_count, 1) (media / 'next.png').write_bytes(b'\x89PNG\r\n\x1a\n') browser.clear_browser_cache() self.assertIn('next.png', browser._media_index(media)) self.assertEqual(scan.call_count, 2) with patch.object(browser.time, 'monotonic', return_value=106): (media / 'later.png').write_bytes(b'\x89PNG\r\n\x1a\n') self.assertIn('later.png', browser._media_index(media)) class SearchRequestsTest(unittest.TestCase): def setUp(self): self.jobs = [] self.responses = [] self.host = types.SimpleNamespace( _wechat_db_state=dict(selectedAccount='1', selectedConversation='M:open', query='', conversationLimit=50, messages=[{'id': 'keep'}], patientContext={'person': 'keep'}, orderSummary={'orders': 'keep'}), _wechat_db_load_serial=7, _wechat_db_query='', _wechat_db_selected_account='1', _wechat_db_selected_conversation='M:open', _schedule_console_push=Mock(), _push_console_state=Mock(), append_log=Mock(), _request_wechat_patient_context=Mock(), _refresh_wechat_database=Mock(), wechatDatabaseLoaded=types.SimpleNamespace(emit=self.responses.append)) for name in ('_search_wechat_database', '_start_wechat_database_search', '_wechat_database_loaded'): setattr(self.host, name, types.MethodType(getattr(gui.MainWindow, name), self.host)) def thread(*, target, **kwargs): return types.SimpleNamespace(start=lambda: self.jobs.append(target)) self.enterContext(patch.object(gui.threading, 'Thread', side_effect=thread)) self.lookup = self.enterContext(patch.object(browser, 'search_browser_conversations', side_effect=lambda **kw: dict(query=kw['query'], conversations=[{'id': kw['query']}], conversationLimit=kw['conversation_limit'], conversationTotal=1, conversationHasMore=False))) def finish_job(self): self.jobs.pop(0)() self.host._wechat_database_loaded(self.responses.pop(0)) def test_fast_typing_coalesces_and_discards_stale_results(self): self.host._search_wechat_database('a') self.host._search_wechat_database('ab') self.host._search_wechat_database('abc') self.assertEqual(len(self.jobs), 1) self.finish_job() self.assertNotIn('conversations', self.host._wechat_db_state) self.assertEqual(self.host._wechat_db_state['query'], 'abc') self.finish_job() self.assertEqual([c.kwargs['query'] for c in self.lookup.call_args_list], ['a', 'abc']) self.assertEqual(self.host._wechat_db_state['conversations'], [{'id': 'abc'}]) self.assertEqual(self.host._wechat_db_state['selectedConversation'], 'M:open') self.assertEqual(self.host._wechat_db_state['messages'], [{'id': 'keep'}]) self.assertEqual(self.host._wechat_db_state['patientContext'], {'person': 'keep'}) self.assertEqual(self.host._wechat_db_state['orderSummary'], {'orders': 'keep'}) self.assertEqual(self.host._wechat_db_load_serial, 7) self.host._request_wechat_patient_context.assert_not_called() self.host._refresh_wechat_database.assert_not_called() def test_switching_account_rejects_old_search(self): self.host._search_wechat_database('old') self.host._wechat_db_state.update(selectedAccount='2', query='new') self.finish_job() self.assertNotIn('conversations', self.host._wechat_db_state) self.assertEqual(self.host._wechat_db_state['selectedAccount'], '2') def test_search_during_full_load_waits_then_keeps_latest_input(self): self.host._wechat_db_state['loading'] = True self.host._search_wechat_database('first') self.host._search_wechat_database('last') self.assertEqual(self.jobs, []) self.host._wechat_database_loaded(dict(_serial=7, loading=False, selectedAccount='1', selectedConversation='M:open', query='', messages=[{'id': 'full'}])) self.assertEqual(self.host._wechat_db_state['query'], 'last') self.finish_job() self.assertEqual(self.host._wechat_db_state['query'], 'last') self.assertEqual(self.host._wechat_db_state['messages'], [{'id': 'full'}]) def test_search_error_does_not_erase_chat_and_can_retry(self): self.lookup.side_effect = ValueError('unavailable') self.host._search_wechat_database('fail') self.finish_job() self.assertIn('unavailable', self.host._wechat_db_state['searchError']) self.assertFalse(self.host._wechat_db_state['searching']) self.assertEqual(self.host._wechat_db_state['messages'], [{'id': 'keep'}]) self.host._search_wechat_database('fail') self.assertEqual(len(self.jobs), 1) def test_queued_navigation_keeps_newer_search_text(self): self.host._wechat_db_state['loading'] = True self.host._wechat_db_pending_request = dict(selected_account='2', query='old') self.host._search_wechat_database('latest') self.host._wechat_database_loaded(dict(_serial=7, loading=False, selectedAccount='1', selectedConversation='M:open', query='old')) self.host._refresh_wechat_database.assert_called_once_with( selected_account='2', query='latest', conversation_limit=50) if __name__ == '__main__': unittest.main()