528 lines
17 KiB
Python
528 lines
17 KiB
Python
"""
|
||
MACP 基础测试模块。
|
||
|
||
对核心服务层进行单元测试，验证各模块基本功能正确性。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.main import app
|
||
from app.models import (
|
||
Document,
|
||
DocumentStatus,
|
||
FileType,
|
||
GjbRule,
|
||
GjbStandard,
|
||
KnowledgeChunk,
|
||
KnowledgeLevel,
|
||
McpTool,
|
||
MemoryEntry,
|
||
MemoryType,
|
||
Skill,
|
||
Template,
|
||
TemplateType,
|
||
Workflow,
|
||
)
|
||
from app.services import (
|
||
DocumentService,
|
||
GjbValidatorService,
|
||
KnowledgeBaseService,
|
||
McpToolService,
|
||
MemoryService,
|
||
ModelService,
|
||
MultimodalService,
|
||
SandboxService,
|
||
SkillService,
|
||
TemplateService,
|
||
)
|
||
|
||
|
||
# =============================================================================
|
||
# 客户端 Fixture
|
||
# =============================================================================
|
||
|
||
@pytest.fixture
|
||
def client():
|
||
"""FastAPI 测试客户端"""
|
||
with TestClient(app) as c:
|
||
yield c
|
||
|
||
|
||
# =============================================================================
|
||
# 健康检查测试
|
||
# =============================================================================
|
||
|
||
class TestHealthCheck:
|
||
"""健康检查接口测试"""
|
||
|
||
def test_health_endpoint(self, client):
|
||
"""测试 /api/health 接口"""
|
||
response = client.get("/api/health")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["status"] == "ok"
|
||
assert data["service"] == "MACP Platform"
|
||
|
||
def test_root_endpoint(self, client):
|
||
"""测试根路径"""
|
||
response = client.get("/")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert "MACP" in data["message"]
|
||
|
||
|
||
# =============================================================================
|
||
# 模板管理服务测试（SRS-MACP_F-001）
|
||
# =============================================================================
|
||
|
||
class TestTemplateService:
|
||
"""文档模板管理服务测试"""
|
||
|
||
def setup_method(self):
|
||
self.service = TemplateService()
|
||
|
||
def test_list_templates(self):
|
||
"""测试获取模板列表"""
|
||
templates = self.service.list_templates()
|
||
assert len(templates) >= 2
|
||
assert any(t.name == "GJB438C 软件需求规格说明书模板" for t in templates)
|
||
|
||
def test_create_template(self):
|
||
"""测试创建模板"""
|
||
new_tpl = Template(
|
||
name="测试模板",
|
||
template_type=TemplateType.SRS,
|
||
gjb_standard=GjbStandard.GJB438C,
|
||
)
|
||
result = self.service.create_template(new_tpl)
|
||
assert result.id is not None
|
||
assert result.name == "测试模板"
|
||
|
||
def test_get_template_not_found(self):
|
||
"""测试获取不存在的模板"""
|
||
result = self.service.get_template("NONEXIST")
|
||
assert result is None
|
||
|
||
def test_delete_template(self):
|
||
"""测试删除模板"""
|
||
assert self.service.delete_template("TPL-001") is True
|
||
assert self.service.delete_template("NONEXIST") is False
|
||
|
||
|
||
# =============================================================================
|
||
# 文档管理服务测试
|
||
# =============================================================================
|
||
|
||
class TestDocumentService:
|
||
"""文档管理服务测试"""
|
||
|
||
def setup_method(self):
|
||
self.service = DocumentService()
|
||
|
||
def test_list_documents(self):
|
||
"""测试获取文档列表"""
|
||
docs = self.service.list_documents()
|
||
assert len(docs) >= 2
|
||
|
||
def test_create_document(self):
|
||
"""测试创建文档"""
|
||
new_doc = Document(
|
||
title="测试文档",
|
||
template_id="TPL-001",
|
||
)
|
||
result = self.service.create_document(new_doc)
|
||
assert result.id is not None
|
||
assert result.status == DocumentStatus.DRAFT
|
||
|
||
def test_get_document(self):
|
||
"""测试获取文档详情"""
|
||
doc = self.service.get_document("DOC-001")
|
||
assert doc is not None
|
||
assert doc.title == "某型雷达系统软件需求规格说明书"
|
||
|
||
|
||
# =============================================================================
|
||
# GJB合规验证服务测试（SRS-MACP_F-002）
|
||
# =============================================================================
|
||
|
||
class TestGjbValidatorService:
|
||
"""GJB合规验证服务测试"""
|
||
|
||
def setup_method(self):
|
||
self.service = GjbValidatorService()
|
||
self.doc_service = DocumentService()
|
||
|
||
def test_list_rules(self):
|
||
"""测试获取规则列表"""
|
||
rules = self.service.list_rules()
|
||
assert len(rules) >= 7
|
||
|
||
def test_validate_document(self):
|
||
"""测试文档合规验证"""
|
||
doc = self.doc_service.get_document("DOC-001")
|
||
assert doc is not None
|
||
report = self.service.validate_document(doc)
|
||
assert report.document_id == "DOC-001"
|
||
assert report.total_rules >= 7
|
||
assert isinstance(report.non_compliant_items, list)
|
||
|
||
|
||
# =============================================================================
|
||
# MCP工具管理服务测试（SRS-MACP_F-004）
|
||
# =============================================================================
|
||
|
||
class TestMcpToolService:
|
||
"""MCP工具管理服务测试"""
|
||
|
||
def setup_method(self):
|
||
self.service = McpToolService()
|
||
|
||
def test_list_tools(self):
|
||
"""测试获取MCP工具列表"""
|
||
tools = self.service.list_tools()
|
||
assert len(tools) >= 2
|
||
|
||
def test_register_tool(self):
|
||
"""测试注册MCP工具"""
|
||
new_tool = McpTool(
|
||
name="test_tool",
|
||
description="测试工具",
|
||
)
|
||
result = self.service.register_tool(new_tool)
|
||
assert result.id is not None
|
||
|
||
def test_debug_tool(self):
|
||
"""测试MCP工具调试"""
|
||
result = self.service.debug_tool("MCP-001", {"test": "data"})
|
||
assert result["success"] is True
|
||
|
||
def test_debug_tool_not_found(self):
|
||
"""测试调试不存在的工具"""
|
||
result = self.service.debug_tool("NONEXIST", {})
|
||
assert result["success"] is False
|
||
|
||
|
||
# =============================================================================
|
||
# Skills编排服务测试（SRS-MACP_F-005）
|
||
# =============================================================================
|
||
|
||
class TestSkillService:
|
||
"""Skills流程编排服务测试"""
|
||
|
||
def setup_method(self):
|
||
self.service = SkillService()
|
||
|
||
def test_list_skills(self):
|
||
"""测试获取Skills列表"""
|
||
skills = self.service.list_skills()
|
||
assert len(skills) >= 2
|
||
|
||
def test_create_skill(self):
|
||
"""测试创建Skill"""
|
||
new_skill = Skill(
|
||
name="测试Skill",
|
||
prompt_template="测试模板",
|
||
)
|
||
result = self.service.create_skill(new_skill)
|
||
assert result.id is not None
|
||
|
||
def test_execute_skill(self):
|
||
"""测试执行Skill"""
|
||
result = self.service.execute_skill("SK-001", {"document_id": "DOC-001"})
|
||
assert result["success"] is True
|
||
assert result["skill_id"] == "SK-001"
|
||
|
||
def test_list_workflows(self):
|
||
"""测试获取工作流列表"""
|
||
workflows = self.service.list_workflows()
|
||
assert len(workflows) >= 1
|
||
|
||
|
||
# =============================================================================
|
||
# 智能体记忆管理服务测试（SRS-MACP_F-006）
|
||
# =============================================================================
|
||
|
||
class TestMemoryService:
|
||
"""智能体记忆管理服务测试"""
|
||
|
||
def setup_method(self):
|
||
self.service = MemoryService()
|
||
|
||
def test_list_memories(self):
|
||
"""测试获取记忆列表"""
|
||
memories = self.service.list_memories()
|
||
assert len(memories) >= 2
|
||
|
||
def test_create_memory(self):
|
||
"""测试创建记忆"""
|
||
new_mem = MemoryEntry(
|
||
session_id="SESS-TEST",
|
||
summary="测试记忆",
|
||
original_dialogue="测试对话内容",
|
||
memory_type=MemoryType.SHORT_TERM,
|
||
)
|
||
result = self.service.create_memory(new_mem)
|
||
assert result.id is not None
|
||
|
||
def test_search_memories(self):
|
||
"""测试搜索记忆"""
|
||
results = self.service.search_memories("GJB438C")
|
||
assert len(results) >= 1
|
||
|
||
def test_delete_memory(self):
|
||
"""测试删除记忆"""
|
||
assert self.service.delete_memory("MEM-001") is True
|
||
assert self.service.delete_memory("NONEXIST") is False
|
||
|
||
|
||
# =============================================================================
|
||
# 知识库交互服务测试（SRS-MACP_F-008）
|
||
# =============================================================================
|
||
|
||
class TestKnowledgeBaseService:
|
||
"""知识库交互服务测试"""
|
||
|
||
def setup_method(self):
|
||
self.service = KnowledgeBaseService()
|
||
|
||
def test_list_chunks(self):
|
||
"""测试获取知识库片段"""
|
||
chunks = self.service.list_chunks()
|
||
assert len(chunks) >= 3
|
||
|
||
def test_list_chunks_by_level(self):
|
||
"""测试按层级过滤知识库片段"""
|
||
docs = self.service.list_chunks(KnowledgeLevel.DOCUMENT)
|
||
assert all(c.level == KnowledgeLevel.DOCUMENT for c in docs)
|
||
|
||
def test_search_chunks(self):
|
||
"""测试搜索知识库"""
|
||
results = self.service.search_chunks("GJB438C")
|
||
assert len(results) >= 1
|
||
|
||
def test_get_chunk_not_found(self):
|
||
"""测试获取不存在的知识库片段"""
|
||
result = self.service.get_chunk("NONEXIST")
|
||
assert result is None
|
||
|
||
|
||
# =============================================================================
|
||
# 大模型接入管理服务测试（SRS-MACP_F-009）
|
||
# =============================================================================
|
||
|
||
class TestModelService:
|
||
"""大模型接入管理服务测试"""
|
||
|
||
def setup_method(self):
|
||
self.service = ModelService()
|
||
|
||
def test_list_models(self):
|
||
"""测试获取模型配置列表"""
|
||
models = self.service.list_models()
|
||
assert len(models) >= 2
|
||
|
||
def test_get_primary_model(self):
|
||
"""测试获取主用模型"""
|
||
model = self.service.get_primary_model()
|
||
assert model is not None
|
||
assert model.is_primary is True
|
||
|
||
def test_health_check(self):
|
||
"""测试模型健康检查"""
|
||
result = self.service.health_check("MDL-001")
|
||
assert result["healthy"] is True
|
||
|
||
def test_failover(self):
|
||
"""测试模型故障转移"""
|
||
model = self.service.failover("MDL-001")
|
||
assert model is not None
|
||
assert model.is_primary is True
|
||
|
||
|
||
# =============================================================================
|
||
# 多模态处理服务测试（SRS-MACP_F-010）
|
||
# =============================================================================
|
||
|
||
class TestMultimodalService:
|
||
"""多模态输入处理服务测试"""
|
||
|
||
def setup_method(self):
|
||
self.service = MultimodalService()
|
||
|
||
def test_process_image(self):
|
||
"""测试图片处理"""
|
||
result = self.service.process_file("test.jpg", FileType.IMAGE)
|
||
assert result.ocr_text is not None
|
||
assert result.file_type == FileType.IMAGE
|
||
|
||
def test_process_audio(self):
|
||
"""测试音频处理"""
|
||
result = self.service.process_file("test.mp3", FileType.AUDIO)
|
||
assert result.asr_transcript is not None
|
||
assert result.file_type == FileType.AUDIO
|
||
|
||
def test_process_video(self):
|
||
"""测试视频处理"""
|
||
result = self.service.process_file("test.mp4", FileType.VIDEO)
|
||
assert result.ocr_text is not None
|
||
assert result.asr_transcript is not None
|
||
assert len(result.key_frames) > 0
|
||
|
||
def test_get_result(self):
|
||
"""测试获取处理结果"""
|
||
result = self.service.process_file("test.png", FileType.IMAGE)
|
||
fetched = self.service.get_result(result.id)
|
||
assert fetched is not None
|
||
assert fetched.id == result.id
|
||
|
||
|
||
# =============================================================================
|
||
# 安全沙箱执行服务测试（SRS-MACP_F-007）
|
||
# =============================================================================
|
||
|
||
class TestSandboxService:
|
||
"""安全沙箱执行服务测试"""
|
||
|
||
def setup_method(self):
|
||
self.service = SandboxService()
|
||
|
||
def test_execute_code(self):
|
||
"""测试沙箱执行代码"""
|
||
result = self.service.execute_code("print('hello')", "python")
|
||
assert result["status"] == "completed"
|
||
assert result["execution_id"] is not None
|
||
|
||
def test_get_execution_logs(self):
|
||
"""测试获取执行日志"""
|
||
self.service.execute_code("x = 1 + 1", "python")
|
||
logs = self.service.get_execution_logs()
|
||
assert len(logs) >= 1
|
||
|
||
|
||
# =============================================================================
|
||
# API 接口集成测试
|
||
# =============================================================================
|
||
|
||
class TestApiEndpoints:
|
||
"""API 接口集成测试"""
|
||
|
||
def test_get_templates_api(self, client):
|
||
"""测试获取模板列表接口"""
|
||
response = client.get("/api/v1/templates")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 200
|
||
assert len(data["data"]) >= 2
|
||
|
||
def test_create_template_api(self, client):
|
||
"""测试创建模板接口"""
|
||
payload = {
|
||
"name": "API测试模板",
|
||
"template_type": "SRS",
|
||
"gjb_standard": "GJB438C",
|
||
}
|
||
response = client.post("/api/v1/templates", json=payload)
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["message"] == "模板创建成功"
|
||
|
||
def test_get_documents_api(self, client):
|
||
"""测试获取文档列表接口"""
|
||
response = client.get("/api/v1/documents")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 200
|
||
|
||
def test_validate_document_api(self, client):
|
||
"""测试文档合规验证接口"""
|
||
response = client.post("/api/v1/documents/DOC-001/validate")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 200
|
||
assert "验证完成" in data["message"]
|
||
|
||
def test_get_rules_api(self, client):
|
||
"""测试获取规则列表接口"""
|
||
response = client.get("/api/v1/rules")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 200
|
||
|
||
def test_get_skills_api(self, client):
|
||
"""测试获取Skills列表接口"""
|
||
response = client.get("/api/v1/skills")
|
||
assert response.status_code == 200
|
||
|
||
def test_execute_skill_api(self, client):
|
||
"""测试执行Skill接口"""
|
||
response = client.post(
|
||
"/api/v1/skills/execute",
|
||
params={"skill_id": "SK-001"},
|
||
json={"document_id": "DOC-001"},
|
||
)
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["data"]["success"] is True
|
||
|
||
def test_get_memory_api(self, client):
|
||
"""测试获取记忆列表接口"""
|
||
response = client.get("/api/v1/memory")
|
||
assert response.status_code == 200
|
||
|
||
def test_search_knowledge_api(self, client):
|
||
"""测试搜索知识库接口"""
|
||
response = client.post(
|
||
"/api/v1/knowledge/search",
|
||
params={"query": "GJB438C", "top_k": 3},
|
||
)
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert len(data["data"]) >= 1
|
||
|
||
def test_get_models_api(self, client):
|
||
"""测试获取模型配置列表接口"""
|
||
response = client.get("/api/v1/models")
|
||
assert response.status_code == 200
|
||
|
||
def test_multimodal_process_api(self, client):
|
||
"""测试多模态处理接口"""
|
||
response = client.post(
|
||
"/api/v1/multimodal/process",
|
||
params={"file_name": "test.jpg", "file_type": "IMAGE"},
|
||
)
|
||
assert response.status_code == 200
|
||
|
||
def test_sandbox_execute_api(self, client):
|
||
"""测试沙箱执行接口"""
|
||
response = client.post(
|
||
"/api/v1/sandbox/execute",
|
||
params={"code": "print('hello')", "language": "python"},
|
||
)
|
||
assert response.status_code == 200
|
||
|
||
def test_mcp_tools_api(self, client):
|
||
"""测试获取MCP工具列表接口"""
|
||
response = client.get("/api/v1/mcp-tools")
|
||
assert response.status_code == 200
|
||
|
||
def test_404_error(self, client):
|
||
"""测试404错误"""
|
||
response = client.get("/api/v1/templates/NONEXIST")
|
||
assert response.status_code == 404
|
||
|
||
def test_get_workflows_api(self, client):
|
||
"""测试获取工作流列表接口"""
|
||
response = client.get("/api/v1/workflows")
|
||
assert response.status_code == 200
|
||
|
||
def test_get_mcp_tool_debug(self, client):
|
||
"""测试MCP工具调试接口"""
|
||
response = client.post(
|
||
"/api/v1/mcp-tools/MCP-001/debug",
|
||
json={"input": "test"},
|
||
)
|
||
assert response.status_code == 200
|