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>
115 lines
4.7 KiB
Python
115 lines
4.7 KiB
Python
"""法规更新管理测试:检查更新、版本快照、更新文档、回滚。
|
|
|
|
用法: python3 -m unittest test_regulation -v
|
|
"""
|
|
import unittest
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from bossai import RegulationUpdateManager, KnowledgeStore
|
|
|
|
|
|
class RegulationUpdateTests(unittest.TestCase):
|
|
"""法规更新管理全流程测试。"""
|
|
|
|
def setUp(self):
|
|
self.mgr = RegulationUpdateManager()
|
|
self.store = KnowledgeStore()
|
|
# 创建测试文档
|
|
self.test_doc = self.store.add(
|
|
title="测试法规-更新管理",
|
|
content="原始内容:测试法规条款",
|
|
source="test",
|
|
doc_type="law",
|
|
effective_date="2020-01-01",
|
|
)
|
|
|
|
def tearDown(self):
|
|
"""清理测试文档。"""
|
|
import psycopg2
|
|
dsn = os.getenv("DATABASE_URL", "host=localhost port=5432 dbname=mcp user=postgres")
|
|
with psycopg2.connect(dsn) as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute("DELETE FROM knowledge_doc_versions WHERE doc_id = %s",
|
|
(self.test_doc["id"],))
|
|
cur.execute("DELETE FROM knowledge_docs WHERE id = %s",
|
|
(self.test_doc["id"],))
|
|
|
|
def test_check_updates_returns_list(self):
|
|
"""check_updates 返回需更新文档列表。"""
|
|
results = self.mgr.check_updates()
|
|
self.assertIsInstance(results, list)
|
|
# 测试文档(2020年生效)应在列表中
|
|
found = [r for r in results if r["id"] == self.test_doc["id"]]
|
|
self.assertTrue(len(found) > 0, "测试文档应在需更新列表中")
|
|
self.assertIn("生效超过1年", found[0]["reasons"][0])
|
|
|
|
def test_create_snapshot(self):
|
|
"""create_snapshot 创建版本快照。"""
|
|
snapshot = self.mgr.create_snapshot(self.test_doc["id"], reason="测试快照")
|
|
self.assertEqual(snapshot["doc_id"], self.test_doc["id"])
|
|
self.assertEqual(snapshot["version_num"], 1)
|
|
self.assertIn("created_at", snapshot)
|
|
|
|
def test_update_document_creates_snapshot(self):
|
|
"""update_document 自动创建更新前快照。"""
|
|
result = self.mgr.update_document(
|
|
doc_id=self.test_doc["id"],
|
|
content="更新后内容:新法规版本",
|
|
title="测试法规-已更新",
|
|
snapshot_reason="法规修订",
|
|
)
|
|
self.assertEqual(result["title"], "测试法规-已更新")
|
|
self.assertEqual(result["snapshot"]["version_num"], 1)
|
|
self.assertIn("created_at", result["snapshot"])
|
|
|
|
def test_list_versions(self):
|
|
"""list_versions 返回版本列表(降序)。"""
|
|
# 创建多个版本
|
|
self.mgr.update_document(self.test_doc["id"], content="v1 内容")
|
|
self.mgr.update_document(self.test_doc["id"], content="v2 内容")
|
|
versions = self.mgr.list_versions(self.test_doc["id"])
|
|
self.assertEqual(len(versions), 2)
|
|
# 降序:最新版本在前
|
|
self.assertEqual(versions[0]["version_num"], 2)
|
|
self.assertEqual(versions[1]["version_num"], 1)
|
|
|
|
def test_rollback(self):
|
|
"""rollback 回滚到指定版本。"""
|
|
# 更新两次
|
|
self.mgr.update_document(self.test_doc["id"], content="v1 内容")
|
|
self.mgr.update_document(self.test_doc["id"], content="v2 内容")
|
|
# 回滚到 v1
|
|
result = self.mgr.rollback(self.test_doc["id"], version_num=1)
|
|
self.assertEqual(result["rolled_back_to"], 1)
|
|
# 验证内容已恢复(直接查 DB)
|
|
import psycopg2
|
|
dsn = os.getenv("DATABASE_URL", "host=localhost port=5432 dbname=mcp user=postgres")
|
|
with psycopg2.connect(dsn) as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute("SELECT content FROM knowledge_docs WHERE id = %s",
|
|
(self.test_doc["id"],))
|
|
content = cur.fetchone()[0]
|
|
self.assertIn("原始内容", content) # v1 快照保存的是更新前的原始内容
|
|
|
|
def test_rollback_nonexistent_version(self):
|
|
"""回滚不存在的版本应报错。"""
|
|
with self.assertRaises(ValueError):
|
|
self.mgr.rollback(self.test_doc["id"], version_num=999)
|
|
|
|
def test_update_nonexistent_doc(self):
|
|
"""更新不存在的文档应报错。"""
|
|
with self.assertRaises(ValueError):
|
|
self.mgr.update_document(doc_id=99999, content="test")
|
|
|
|
def test_fetch_from_source_returns_empty(self):
|
|
"""fetch_from_source 预留接口返回空列表。"""
|
|
result = self.mgr.fetch_from_source("test")
|
|
self.assertEqual(result, [])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|