3705945a7d
架构重构（候选 1-6）： - storage/repository.py 按域拆分为包（target/scenario/run/campaign/result） - storage/db.py 按域拆分为包（eval/campaign/file/model_config/intelligent_eval） - intelligent_eval/lifecycle.py 按状态机阶段拆分为包 - services/runs.py 编排逻辑下沉 - Campaigns.tsx 拆分为 campaigns/ 子组件 测试补全（候选 7）： 前端（+125 用例，107→232）： - utils/ 纯函数：date/campaignTime/ruleLabels/fileTree/fileFormat/colors - stores/tabStore 状态管理 - 核心组件：FormDrawer/PageWrapper/ChatBubble/GeneratedMessages/SectionHeader/StatCard/TurnList - 业务组件：CaseBlock/CaseDetail/RuleOverview/WindowTimeline/RunList/TabBar/CampaignRunTimeline - 文件管理：FileCategoryTree/FileTable - hooks：sessionReducer/useFiles/useRunSession 后端（+38 用例，916→954）： - targets API CRUD + 404 路径 - WebSocket 连接管理器 - proxy 头部重写（CSP/X-Frame-Options） - target 仓储 update 方法 - app 健康检查 + SPA 404 - scenarios 模板端点 + 404 - files API 边缘分支（404 场景 + 500 兜底） - files service update_category - 智能评估状态机迁移测试 门禁状态： - 前端：tsc 干净 + 232 passed - 后端：954 passed + ruff 全绿
118 lines
3.6 KiB
Python
118 lines
3.6 KiB
Python
"""Tests for file workflow transaction compensation and path safety."""
|
|
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from agenteval.services.files import FileManagementService
|
|
from agenteval.storage.file_repository import FileRecordRepository
|
|
from agenteval.storage.file_storage import FileStorage, InvalidStorageNameError
|
|
from sqlmodel import Session, SQLModel, create_engine
|
|
from starlette.datastructures import Headers, UploadFile
|
|
|
|
|
|
@pytest.fixture()
|
|
def service_context(tmp_path: Path):
|
|
from agenteval.storage.db import ( # noqa: F401
|
|
EvalResultDB,
|
|
EvalRunDB,
|
|
EvalTargetDB,
|
|
FileCategoryDB,
|
|
FileRecordDB,
|
|
ScenarioDB,
|
|
TurnDB,
|
|
)
|
|
|
|
engine = create_engine(
|
|
f"sqlite:///{tmp_path / 'file_service.db'}",
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
SQLModel.metadata.create_all(engine)
|
|
session = Session(engine)
|
|
storage = FileStorage(tmp_path / "files")
|
|
service = FileManagementService(session, storage, {"txt"}, 1)
|
|
yield service, session, storage
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
|
|
def _upload(content: bytes = b"content") -> UploadFile:
|
|
return UploadFile(
|
|
filename="test.txt",
|
|
file=BytesIO(content),
|
|
headers=Headers({"content-type": "text/plain"}),
|
|
)
|
|
|
|
|
|
async def test_upload_commit_failure_removes_physical_file(service_context, monkeypatch):
|
|
service, session, storage = service_context
|
|
|
|
def fail_commit():
|
|
raise RuntimeError("commit failed")
|
|
|
|
monkeypatch.setattr(session, "commit", fail_commit)
|
|
|
|
with pytest.raises(RuntimeError, match="commit failed"):
|
|
await service.upload(_upload())
|
|
|
|
assert list(storage.root.iterdir()) == []
|
|
assert FileRecordRepository(session).list_all() == []
|
|
|
|
|
|
def test_delete_commit_failure_restores_physical_file(service_context, monkeypatch):
|
|
service, session, storage = service_context
|
|
record = FileRecordRepository(session).create(
|
|
original_name="test.txt",
|
|
storage_name="stored.txt",
|
|
file_size=7,
|
|
mime_type="text/plain",
|
|
file_ext="txt",
|
|
)
|
|
session.commit()
|
|
physical_file = storage.path_for("stored.txt")
|
|
physical_file.write_bytes(b"content")
|
|
|
|
def fail_commit():
|
|
raise RuntimeError("commit failed")
|
|
|
|
monkeypatch.setattr(session, "commit", fail_commit)
|
|
|
|
with pytest.raises(RuntimeError, match="commit failed"):
|
|
service.delete_file(record.id)
|
|
|
|
assert physical_file.read_bytes() == b"content"
|
|
assert FileRecordRepository(session).get(record.id) is not None
|
|
|
|
|
|
@pytest.mark.parametrize("storage_name", ["../escape.txt", "folder/file.txt", ""])
|
|
def test_storage_rejects_names_outside_root(service_context, storage_name):
|
|
_, _, storage = service_context
|
|
|
|
with pytest.raises(InvalidStorageNameError):
|
|
storage.path_for(storage_name)
|
|
|
|
|
|
def test_update_category_renames_successfully(service_context):
|
|
service, session, _ = service_context
|
|
cat = service.create_category("原名")
|
|
|
|
updated = service.update_category(cat.id, "新名")
|
|
assert updated.name == "新名"
|
|
|
|
|
|
def test_update_category_empty_name_raises(service_context):
|
|
service, _, _ = service_context
|
|
cat = service.create_category("原名")
|
|
|
|
from agenteval.services.files import FileValidationError
|
|
with pytest.raises(FileValidationError):
|
|
service.update_category(cat.id, " ")
|
|
|
|
|
|
def test_update_category_nonexistent_raises(service_context):
|
|
service, _, _ = service_context
|
|
|
|
from agenteval.services.files import CategoryNotFoundError
|
|
with pytest.raises(CategoryNotFoundError):
|
|
service.update_category("nonexistent", "新名")
|