"""阅后即焚测试:超时、退出、重复销毁、销毁后访问、崩溃恢复、临时文件清理。 测试覆盖: - 临时会话(ephemeral)5 分钟超时后自动销毁 - confidential 24 小时超时 - 手动销毁后消息清空 - 重复销毁返回 False - 销毁后无法再提问 - 销毁后 key_digest 为空 - 销毁日志记录完整步骤 - 临时文件在销毁时被清理 - 销毁状态可查询 - purge_expired 自动清理过期会话 - 销毁统计可追踪 - 崩溃恢复:新 SessionManager 不保留旧会话(内存模式) 用法: python3 -m unittest test_burn_after_reading -v """ import unittest import os import sys import tempfile from datetime import datetime, timedelta, timezone sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from bossai import SessionManager, PrivacySession class BurnAfterReadingTests(unittest.TestCase): """阅后即焚全路径测试。""" def test_ephemeral_session_auto_destroy_on_timeout(self): """临时会话超时后自动销毁。""" # 用可控时钟 start = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc) clock = {'time': start} def mock_clock(): return clock['time'] mgr = SessionManager(clock=mock_clock) session = mgr.create('ephemeral') # 5 分钟 TTL self.assertFalse(session.destroyed) # 写入消息 mgr.append(session.session_id, 'user', '秘密信息12345') self.assertEqual(len(session.messages), 1) # 前进 6 分钟(超过 5 分钟 TTL) clock['time'] = start + timedelta(minutes=6) # purge_expired 应自动销毁 removed = mgr.purge_expired() self.assertEqual(removed, 1) self.assertTrue(session.destroyed) self.assertEqual(len(session.messages), 0) def test_confidential_session_24h_timeout(self): """confidential 会话 24 小时后超时。""" start = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc) clock = {'time': start} def mock_clock(): return clock['time'] mgr = SessionManager(clock=mock_clock) session = mgr.create('confidential') # 24 小时 TTL self.assertFalse(session.destroyed) # 23 小时后仍可用 clock['time'] = start + timedelta(hours=23) mgr.append(session.session_id, 'user', '测试') self.assertEqual(len(session.messages), 1) # 25 小时后自动销毁 clock['time'] = start + timedelta(hours=25) removed = mgr.purge_expired() self.assertEqual(removed, 1) self.assertTrue(session.destroyed) def test_manual_destroy_clears_messages(self): """手动销毁后消息清空。""" mgr = SessionManager() session = mgr.create('confidential') mgr.append(session.session_id, 'user', '敏感对话内容') mgr.append(session.session_id, 'assistant', '回复内容') result = mgr.destroy(session.session_id) self.assertTrue(result) self.assertTrue(session.destroyed) self.assertEqual(len(session.messages), 0) def test_manual_destroy_clears_key_digest(self): """销毁后 key_digest 为空。""" mgr = SessionManager() session = mgr.create('ephemeral') self.assertNotEqual(session.key_digest, "") mgr.destroy(session.session_id) self.assertEqual(session.key_digest, "") def test_double_destroy_returns_false(self): """重复销毁返回 False。""" mgr = SessionManager() session = mgr.create('ephemeral') first = mgr.destroy(session.session_id) self.assertTrue(first) second = mgr.destroy(session.session_id) self.assertFalse(second) def test_destroyed_session_cannot_append(self): """销毁后无法再写入消息。""" mgr = SessionManager() session = mgr.create('confidential') mgr.destroy(session.session_id) with self.assertRaises(KeyError): mgr.append(session.session_id, 'user', '销毁后写入') def test_destroy_log_records_steps(self): """销毁日志记录完整步骤。""" mgr = SessionManager() session = mgr.create('ephemeral') mgr.append(session.session_id, 'user', '测试') mgr.destroy(session.session_id) status = mgr.destroy_status(session.session_id) self.assertIsNotNone(status) self.assertTrue(status['destroyed']) self.assertGreater(len(status['destroy_log']), 0) # 日志应包含步骤记录 log = status['destroy_log'][0] self.assertIn('steps', log) # 验证关键步骤 steps = log['steps'] self.assertIn('messages_cleared', steps) self.assertIn('key_destroyed', steps) self.assertIn('marked_destroyed', steps) def test_temp_files_cleaned_on_destroy(self): """销毁时临时文件被清理。""" mgr = SessionManager() session = mgr.create('confidential') # 创建临时文件 tmp1 = tempfile.NamedTemporaryFile(delete=False, suffix='.txt') tmp1.write(b'secret1') tmp1.close() tmp2 = tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') tmp2.write(b'secret2') tmp2.close() mgr.register_temp_file(session.session_id, tmp1.name) mgr.register_temp_file(session.session_id, tmp2.name) self.assertTrue(os.path.exists(tmp1.name)) self.assertTrue(os.path.exists(tmp2.name)) mgr.destroy(session.session_id) self.assertFalse(os.path.exists(tmp1.name)) self.assertFalse(os.path.exists(tmp2.name)) self.assertEqual(len(session.temp_files), 0) def test_destroy_status_query(self): """销毁状态可查询。""" mgr = SessionManager() session = mgr.create('ephemeral') # 销毁前查询 status = mgr.destroy_status(session.session_id) self.assertFalse(status['destroyed']) mgr.destroy(session.session_id) # 销毁后查询 status = mgr.destroy_status(session.session_id) self.assertTrue(status['destroyed']) def test_destroy_status_nonexistent(self): """查询不存在的会话返回 None。""" mgr = SessionManager() status = mgr.destroy_status('nonexistent-id') self.assertIsNone(status) def test_purge_expired_only_destroys_expired(self): """purge_expired 只销毁过期会话,不影响活跃会话。""" start = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc) clock = {'time': start} def mock_clock(): return clock['time'] mgr = SessionManager(clock=mock_clock) s1 = mgr.create('ephemeral') # 5 分钟 s2 = mgr.create('confidential') # 24 小时 # 前进 6 分钟 clock['time'] = start + timedelta(minutes=6) removed = mgr.purge_expired() self.assertEqual(removed, 1) self.assertTrue(s1.destroyed) self.assertFalse(s2.destroyed) def test_destroy_stats_tracking(self): """销毁统计可追踪。""" mgr = SessionManager() s1 = mgr.create('ephemeral') s2 = mgr.create('ephemeral') mgr.destroy(s1.session_id) mgr.destroy(s2.session_id) stats = mgr.destroy_stats() self.assertEqual(stats['success'], 2) self.assertEqual(stats['failed'], 0) def test_emergency_stop_destroys_session(self): """急停销毁会话并阻止后续写入。""" mgr = SessionManager() session = mgr.create('confidential') mgr.append(session.session_id, 'user', '紧急情况') result = mgr.emergency_stop(session.session_id) self.assertTrue(result) self.assertTrue(session.destroyed) self.assertEqual(len(session.messages), 0) # 急停后无法写入 with self.assertRaises(KeyError): mgr.append(session.session_id, 'user', '急停后写入') def test_crash_recovery_no_persistence(self): """崩溃恢复:新 SessionManager 不保留旧会话(内存模式)。 这是内存模式的预期行为:进程崩溃后会话全部丢失。 生产环境应使用加密持久化存储。 """ mgr1 = SessionManager() session = mgr1.create('confidential') mgr1.append(session.session_id, 'user', '重要数据') # 模拟崩溃:新建 SessionManager mgr2 = SessionManager() status = mgr2.destroy_status(session.session_id) self.assertIsNone(status) # 旧会话不存在 def test_expired_session_auto_destroy_on_access(self): """过期会话在访问时自动销毁。""" start = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc) clock = {'time': start} def mock_clock(): return clock['time'] mgr = SessionManager(clock=mock_clock) session = mgr.create('ephemeral') # 超时 clock['time'] = start + timedelta(minutes=10) # 访问应触发自动销毁 with self.assertRaises(KeyError): mgr.append(session.session_id, 'user', '超时后访问') # 验证已自动销毁 self.assertTrue(session.destroyed) def test_normal_session_long_ttl(self): """normal 会话有超长 TTL(不主动过期)。""" start = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc) clock = {'time': start} def mock_clock(): return clock['time'] mgr = SessionManager(clock=mock_clock) session = mgr.create('normal') # 前进 1 年 clock['time'] = start + timedelta(days=365) removed = mgr.purge_expired() self.assertEqual(removed, 0) self.assertFalse(session.destroyed) if __name__ == '__main__': unittest.main(verbosity=2)