171 lines
9.3 KiB
Python
171 lines
9.3 KiB
Python
"""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()
|