| import pathlib |
| import unittest |
|
|
|
|
| ROOT=pathlib.Path(__file__).resolve().parents[1] |
|
|
|
|
| class ConversationRecoveryTests(unittest.TestCase): |
| def test_backend_tracks_transcript_and_rebuilds_expired_thread(self): |
| source=(ROOT/"ui_server.py").read_text(encoding="utf-8") |
| self.assertIn("thread_recent_history={}", source) |
| self.assertIn("def _is_codewhale_missing_thread_error", source) |
| self.assertIn("def recover_codewhale_thread", source) |
| self.assertIn("def _format_recent_thread_history", source) |
|
|
| def test_chat_route_accepts_history_payload(self): |
| source=(ROOT/"routes"/"chat.py").read_text(encoding="utf-8") |
| self.assertIn( |
| "history:list[dict[str,str]]|None", |
| (ROOT/"ui_server.py").read_text(encoding="utf-8"), |
| ) |
| self.assertIn("thread_recent_history[x.thread_id]", source) |
|
|
| def test_followup_amendments_stay_in_data_runtime(self): |
| source=(ROOT/"ui_server.py").read_text(encoding="utf-8") |
| runtime=(ROOT/"services"/"chat_runtime.py").read_text(encoding="utf-8") |
| compressor=(ROOT/"services"/"context_compressor.py").read_text(encoding="utf-8") |
| self.assertIn("def _is_data_request_followup", source) |
| self.assertIn("from services.context_compressor import", runtime) |
| self.assertIn("COMPRESSED_DATA_REQUEST_CONTEXT", compressor) |
| self.assertIn("[AMENDMENT_DIRECTIVE]", runtime) |
|
|
| def test_frontend_sends_history_and_keeps_session_on_thread_expiry(self): |
| js=(ROOT/"static"/"js"/"app.js").read_text(encoding="utf-8") |
| self.assertIn("function historyPayload", js) |
| self.assertIn("history:historyBeforeSend", js) |
| self.assertIn("thread_recovered", js) |
| self.assertIn("重新发送时会自动重建上下文", js) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
