from __future__ import annotations import argparse import importlib.util import json from pathlib import Path import tempfile import unittest SCRIPT_PATH = Path(__file__).resolve().parents[1] / "export_team_codex_conversations.py" SPEC = importlib.util.spec_from_file_location("team_conversation_exporter", SCRIPT_PATH) MODULE = importlib.util.module_from_spec(SPEC) assert SPEC.loader is not None SPEC.loader.exec_module(MODULE) class ExportTeamCodexConversationsTest(unittest.TestCase): def make_index(self, root: Path, threads: list[dict]) -> Path: path = root / "index.json" path.write_text( json.dumps( {"generated_at": "2026-08-10T19:00:00+08:00", "threads": threads}, ensure_ascii=False, ), encoding="utf-8", ) return path def args(self, index: Path, output: Path, collector: str = "jiangyang") -> argparse.Namespace: return argparse.Namespace( archive_index=index, workspace="zhctprompt", collector=collector, date="2026-08-10", output_root=output, dry_run=False, json=False, ) def sample_thread(self, thread_id: str = "thread-secret-1") -> dict: return { "thread_id": thread_id, "title": "排查接口", "workspace": "008-zhct/zhctprompt", "cwd": "/Users/alice/code/zhctprompt", "stage": "进行中", "source": "vscode", "started_at": "2026-08-10T10:00:00+08:00", "updated_at": "2026-08-10T11:00:00+08:00", "goal": "检查 10.0.0.8，手机号 13800138000", "plan": ["读取 /Users/alice/work", "使用 token=abc123456789012345"], "progress": "联系 a@example.com；Host：`prod.mysql.rds.aliyuncs.com` Username：`root` Password：`cpt2025@)!(` mysql -h prod.mysql.rds.aliyuncs.com -P3306 -uroot -p", "timeline": [ { "timestamp": "2026-08-10T10:00:00+08:00", "role": "user", "message": "password=hunter2 bearer abcdefghijklmnopqrstuvwxyz id=019f77ee-2fc7-75b1-a1f4-95f13a01d6fd /var/folders/a/private.png", } ], } def test_exports_only_matching_workspace_and_redacts_sensitive_values(self) -> None: with tempfile.TemporaryDirectory() as temp: root = Path(temp) matching = self.sample_thread() other = dict(self.sample_thread("thread-other"), workspace="other-project") index = self.make_index(root, [matching, other]) result = MODULE.export_conversations(self.args(index, root / "out")) self.assertEqual(result["matched_count"], 1) self.assertEqual(result["exported_count"], 1) files = list((root / "out/collectors/jiangyang/threads").glob("*.md")) self.assertEqual(len(files), 1) content = files[0].read_text(encoding="utf-8") self.assertNotIn("thread-secret-1", content) self.assertNotIn("13800138000", content) self.assertNotIn("10.0.0.8", content) self.assertNotIn("a@example.com", content) self.assertNotIn("hunter2", content) self.assertNotIn("cpt2025", content) self.assertNotIn("prod.mysql.rds.aliyuncs.com", content) self.assertNotIn("Username：`root`", content) self.assertNotIn("019f77ee-2fc7-75b1-a1f4-95f13a01d6fd", content) self.assertNotIn("/var/folders", content) self.assertNotIn("/Users/alice", content) self.assertIn("restricted-metadata-only", content) self.assertIn("[受限内容未进入共享 Git]", content) self.assertEqual(result["restricted_count"], 1) def test_sanitize_text_masks_credentials_without_context_classifier(self) -> None: raw = "Host：`prod.mysql.rds.aliyuncs.com` Username：`root` Password：`cpt2025@)!(` Note ID 是：ABCDEF0123456789ABCDEF0123456789" content, replacements = MODULE.sanitize_text(raw) self.assertGreaterEqual(replacements, 3) self.assertNotIn("prod.mysql.rds.aliyuncs.com", content) self.assertNotIn("root", content) self.assertNotIn("cpt2025", content) self.assertNotIn("ABCDEF0123456789ABCDEF0123456789", content) def test_hr_conversation_is_metadata_only(self) -> None: with tempfile.TemporaryDirectory() as temp: root = Path(temp) thread = self.sample_thread("hr-thread") thread.update( { "title": "新员工试用期考核表", "goal": "整理张三的培养计划", "plan": ["查看个人绩效"], "progress": "等待主管确认", "timeline": [ { "timestamp": "2026-08-10T10:00:00+08:00", "role": "user", "message": "请整理张三的新员工试用期材料", } ], } ) index = self.make_index(root, [thread]) result = MODULE.export_conversations(self.args(index, root / "out")) detail = next((root / "out/collectors/jiangyang/threads").glob("*.md")).read_text( encoding="utf-8" ) daily = Path(result["manifest_path"]).read_text(encoding="utf-8") self.assertNotIn("张三", detail) self.assertNotIn("张三", daily) self.assertNotIn("试用期考核表", detail) self.assertNotIn("试用期考核表", daily) self.assertIn("受限会话", detail) self.assertIn("受限内容未进入共享 Git", daily) def test_second_run_is_idempotent(self) -> None: with tempfile.TemporaryDirectory() as temp: root = Path(temp) index = self.make_index(root, [self.sample_thread()]) args = self.args(index, root / "out") MODULE.export_conversations(args) second = MODULE.export_conversations(args) self.assertEqual(second["counts"]["unchanged"], 1) def test_daily_file_contains_only_that_days_questions_and_answers(self) -> None: with tempfile.TemporaryDirectory() as temp: root = Path(temp) thread = self.sample_thread() thread["timeline"].append( { "timestamp": "2026-08-09T09:00:00+08:00", "role": "user", "message": "昨天的问题不应进入今天文件", } ) index = self.make_index(root, [thread]) result = MODULE.export_conversations(self.args(index, root / "out")) daily = Path(result["manifest_path"]).read_text(encoding="utf-8") self.assertIn("password=hunter2", thread["timeline"][0]["message"]) self.assertIn("[受限内容未进入共享 Git]", daily) self.assertNotIn("昨天的问题不应进入今天文件", daily) self.assertEqual(result["daily_question_count"], 1) self.assertEqual(result["daily_answer_count"], 0) self.assertEqual(result["daily_message_count"], 1) def test_collectors_use_separate_directories(self) -> None: with tempfile.TemporaryDirectory() as temp: root = Path(temp) index = self.make_index(root, [self.sample_thread()]) MODULE.export_conversations(self.args(index, root / "out", "jiangyang")) MODULE.export_conversations(self.args(index, root / "out", "sunqianqian")) self.assertTrue((root / "out/collectors/jiangyang/threads").is_dir()) self.assertTrue((root / "out/collectors/sunqianqian/threads").is_dir()) def test_missing_thread_id_fails_closed(self) -> None: with tempfile.TemporaryDirectory() as temp: root = Path(temp) thread = self.sample_thread() thread.pop("thread_id") index = self.make_index(root, [thread]) with self.assertRaises(ValueError): MODULE.export_conversations(self.args(index, root / "out")) if __name__ == "__main__": unittest.main()