"""灾难恢复演练:备份恢复验证 + 数据删除验证。 测试覆盖: 1. 备份恢复:pg_dump → 恢复到临时库 → 验证数据一致 2. 数据删除验证:销毁会话后 DB 无残留、摘要无正文、日志无敏感信息 3. 备份完整性:备份文件可解压、可读、包含预期表 用法: python3 -m unittest test_disaster_recovery -v """ import unittest import os import sys import subprocess import tempfile import json sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import psycopg2 def get_dsn(): return os.getenv("DATABASE_URL", "host=localhost port=5432 dbname=mcp user=postgres") class BackupRestoreTests(unittest.TestCase): """备份恢复验证。""" def test_backup_file_exists(self): """备份文件存在且非空。""" backup_dir = os.path.join(os.path.dirname(__file__), "backups") if not os.path.exists(backup_dir): self.skipTest("backups 目录不存在,先运行 ./backup.sh") files = [f for f in os.listdir(backup_dir) if f.endswith('.sql.gz')] self.assertGreater(len(files), 0, "无备份文件") def test_backup_can_decompress(self): """备份文件可解压。""" backup_dir = os.path.join(os.path.dirname(__file__), "backups") files = [f for f in os.listdir(backup_dir) if f.endswith('.sql.gz')] if not files: self.skipTest("无备份文件") latest = sorted(files)[-1] path = os.path.join(backup_dir, latest) # 解压到临时文件 with tempfile.NamedTemporaryFile(suffix='.sql', delete=False) as tmp: result = subprocess.run( ['gunzip', '-c', path], capture_output=True, timeout=30, ) self.assertEqual(result.returncode, 0, f"gunzip 失败: {result.stderr.decode()}") tmp.write(result.stdout) tmp_path = tmp.name # 验证 SQL 内容包含关键表 with open(tmp_path, encoding='utf-8', errors='replace') as f: sql = f.read() os.unlink(tmp_path) self.assertIn('knowledge_docs', sql) self.assertIn('user_settings', sql) self.assertIn('review_tickets', sql) def test_backup_and_restore_roundtrip(self): """备份 → 恢复到临时库 → 验证数据一致。""" dsn = get_dsn() # 1. 创建临时数据库 admin_dsn = dsn.replace("dbname=mcp", "dbname=postgres") test_db = "mcp_restore_test" try: conn = psycopg2.connect(admin_dsn) conn.autocommit = True with conn.cursor() as cur: cur.execute(f"DROP DATABASE IF EXISTS {test_db}") cur.execute(f"CREATE DATABASE {test_db}") conn.close() except Exception as e: self.skipTest(f"无法创建测试数据库: {e}") try: # 2. 备份当前库 with tempfile.NamedTemporaryFile(suffix='.sql', delete=False) as tmp: dump_path = tmp.name result = subprocess.run( ['pg_dump', '-h', 'localhost', '-U', 'postgres', '-d', 'mcp', '-f', dump_path], capture_output=True, timeout=30, ) self.assertEqual(result.returncode, 0, f"pg_dump 失败: {result.stderr.decode()}") # 3. 恢复到临时库 restore_dsn = dsn.replace("dbname=mcp", f"dbname={test_db}") result = subprocess.run( ['psql', '-h', 'localhost', '-U', 'postgres', '-d', test_db, '-f', dump_path], capture_output=True, timeout=30, ) # psql 恢复可能有非零退出码(警告),检查关键表存在即可 os.unlink(dump_path) # 4. 验证数据一致 with psycopg2.connect(restore_dsn) as conn: with conn.cursor() as cur: cur.execute("SELECT COUNT(*) FROM knowledge_docs") restored_count = cur.fetchone()[0] with psycopg2.connect(dsn) as conn: with conn.cursor() as cur: cur.execute("SELECT COUNT(*) FROM knowledge_docs") original_count = cur.fetchone()[0] self.assertEqual(restored_count, original_count, f"恢复后数据不一致: 原始 {original_count}, 恢复 {restored_count}") finally: # 清理临时库 try: with psycopg2.connect(admin_dsn) as conn: conn.autocommit = True with conn.cursor() as cur: cur.execute(f"DROP DATABASE IF EXISTS {test_db}") except Exception: pass class DataDeletionTests(unittest.TestCase): """数据删除验证:销毁后无残留。""" def test_destroyed_session_no_db_residue(self): """销毁会话后内存无消息残留。""" from bossai import SessionManager mgr = SessionManager() session = mgr.create('confidential') mgr.append(session.session_id, 'user', '包含敏感信息:手机号13812345678') mgr.append(session.session_id, 'assistant', '回复:身份证号110101199001011234') mgr.destroy(session.session_id) # 验证消息清空 self.assertEqual(len(session.messages), 0) # 验证 key_digest 清空 self.assertEqual(session.key_digest, "") # 验证标记销毁 self.assertTrue(session.destroyed) def test_action_summary_no_plaintext_pii(self): """脱敏摘要不含明文 PII。""" from bossai import ActionSummaryStore store = ActionSummaryStore() result = store.save("test-session", { 'risk_level': 'orange', 'conclusion': '客户手机13812345678需要联系', 'actions': ['拨打13812345678', '发送邮件至test@example.com'], }) summary_str = json.dumps(result['summary'], ensure_ascii=False) self.assertNotIn('13812345678', summary_str) self.assertIn('[手机号]', summary_str) def test_destroy_log_no_plaintext(self): """销毁日志不含正文内容。""" from bossai import SessionManager mgr = SessionManager() session = mgr.create('ephemeral') mgr.append(session.session_id, 'user', '这是秘密对话内容不应该出现在日志中') mgr.destroy(session.session_id) status = mgr.destroy_status(session.session_id) log_str = json.dumps(status['destroy_log'], ensure_ascii=False) # 日志不应包含正文 self.assertNotIn('秘密对话内容', log_str) # 日志应包含步骤 self.assertIn('steps', status['destroy_log'][0]) def test_temp_files_deleted(self): """销毁后临时文件物理删除。""" import tempfile from bossai import SessionManager mgr = SessionManager() session = mgr.create('confidential') tmp = tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') tmp.write(b'sensitive document content') tmp.close() self.assertTrue(os.path.exists(tmp.name)) mgr.register_temp_file(session.session_id, tmp.name) mgr.destroy(session.session_id) self.assertFalse(os.path.exists(tmp.name), "临时文件未被删除") def test_metrics_no_plaintext(self): """指标端点不含敏感正文。""" from fastapi.testclient import TestClient import api as api_module client = TestClient(api_module.app) api_module._rate_buckets.clear() r = client.get('/metrics') self.assertEqual(r.status_code, 200) metrics_str = json.dumps(r.json(), ensure_ascii=False) # 指标不应包含任何 PII import re phones = re.findall(r'1[3-9]\d{9}', metrics_str) self.assertEqual(len(phones), 0, f"指标中发现手机号: {phones}") if __name__ == '__main__': unittest.main(verbosity=2)