98 lines
3.7 KiB
Python
98 lines
3.7 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
from app.api.task_briefs import (
|
|
GenerateTaskBriefRequest,
|
|
TemplateCreateRequest,
|
|
create_task_brief_template,
|
|
generate_task_brief,
|
|
get_task_brief_contract_sync,
|
|
list_task_brief_center,
|
|
list_task_brief_runs,
|
|
list_task_brief_templates,
|
|
)
|
|
from app.models import Base, User
|
|
|
|
|
|
class TaskBriefApiTest(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self) -> None:
|
|
tmp = tempfile.NamedTemporaryFile(prefix="pmflow-task-brief-api-", suffix=".db", delete=False)
|
|
tmp.close()
|
|
self.db_path = tmp.name
|
|
self.engine = create_async_engine(f"sqlite+aiosqlite:///{self.db_path}", future=True)
|
|
self.session_factory = async_sessionmaker(self.engine, expire_on_commit=False, class_=AsyncSession)
|
|
async with self.engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
async def asyncTearDown(self) -> None:
|
|
await self.engine.dispose()
|
|
if getattr(self, "db_path", None):
|
|
try:
|
|
os.remove(self.db_path)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
async def test_template_generate_and_list(self) -> None:
|
|
async with self.session_factory() as db:
|
|
user = User(
|
|
tenant_id=1,
|
|
username="brief_admin",
|
|
email="brief_admin@example.com",
|
|
password_hash="hash",
|
|
name="模板管理员",
|
|
role="ADMIN",
|
|
status="active",
|
|
is_active=True,
|
|
)
|
|
db.add(user)
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
|
|
created = await create_task_brief_template(
|
|
body=TemplateCreateRequest(
|
|
name="默认增量模板",
|
|
description="用于 planner 增量任务包",
|
|
content="# Template Body",
|
|
),
|
|
db=db,
|
|
current_user=user,
|
|
)
|
|
self.assertEqual(created["code"], 0)
|
|
template_id = str(created["data"]["id"])
|
|
|
|
templates = await list_task_brief_templates(db=db, current_user=user)
|
|
self.assertEqual(templates["code"], 0)
|
|
self.assertGreaterEqual(len(templates["data"]["items"]), 1)
|
|
|
|
generated = await generate_task_brief(
|
|
body=GenerateTaskBriefRequest(
|
|
title="Signal Planner Increment",
|
|
template_id=template_id,
|
|
changed_files=[
|
|
"backend/app/services/signal_planner_service.py",
|
|
"frontend/src/pages/ModelRoutingConfigPage.tsx",
|
|
],
|
|
scope_summary="后端 Planner + 前端配置页联动",
|
|
),
|
|
db=db,
|
|
current_user=user,
|
|
)
|
|
self.assertEqual(generated["code"], 0)
|
|
self.assertEqual(generated["data"]["run"]["status"], "completed")
|
|
|
|
center = await list_task_brief_center(db=db, current_user=user)
|
|
self.assertEqual(center["code"], 0)
|
|
self.assertGreaterEqual(len(center["data"]["items"]), 1)
|
|
|
|
runs = await list_task_brief_runs(db=db, current_user=user)
|
|
self.assertEqual(runs["code"], 0)
|
|
self.assertGreaterEqual(len(runs["data"]["items"]), 1)
|
|
|
|
sync_board = await get_task_brief_contract_sync(db=db, current_user=user)
|
|
self.assertEqual(sync_board["code"], 0)
|
|
self.assertIn("all_synced", sync_board["data"])
|
|
self.assertIsInstance(sync_board["data"].get("files"), list)
|