selfreleaseandDevin
5c21e8fcd0
包含功能: - 法务参谋/经营决策/生活助理/情绪中心四大AI对话模块 - 专家咨询系统(用户选定→专家确认→咨询→回复→评价) - 角色登录(用户/专家/管理员)+ ProtectedRoute 权限控制 - 管理后台(用户管理/专家管理/AI使用统计/专家活跃度) - 专家工作台(待确认服务/待回复咨询/已回复咨询/服务中用户) - 服务关系取消(用户和专家均可取消,需自己确认) - 全页面移动端响应式适配 - 知识库RAG检索 - 对话历史(普通/保密模式) Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
293 lines
9.7 KiB
Python
293 lines
9.7 KiB
Python
"""阅后即焚测试:超时、退出、重复销毁、销毁后访问、崩溃恢复、临时文件清理。
|
|
|
|
测试覆盖:
|
|
- 临时会话(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)
|