PMFlow/backend/tests/test_task_briefs_api.py

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, Integration, User
from app.services.task_brief_service import DEFAULT_TEMPLATE_FALLBACK


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)

    async def test_default_template_upgrade_and_placeholder_render(self) -> None:
        async with self.session_factory() as db:
            user = User(
                tenant_id=1,
                username="brief_admin_upgrade",
                email="brief_admin_upgrade@example.com",
                password_hash="hash",
                name="模板管理员",
                role="ADMIN",
                status="active",
                is_active=True,
            )
            db.add(user)
            await db.commit()
            await db.refresh(user)

            db.add(
                Integration(
                    tenant_id=1,
                    type="task_briefs",
                    is_active=True,
                    config={
                        "templates": [
                            {
                                "id": "tpl_default_signal_planner",
                                "name": "Signal Planner 默认模板",
                                "description": "系统初始化模板（可按需修改）",
                                "content": DEFAULT_TEMPLATE_FALLBACK,
                                "updated_at": "2026-01-01T00:00:00",
                                "updated_by": 0,
                                "created_at": "2026-01-01T00:00:00",
                            }
                        ],
                        "runs": [],
                        "briefs": [],
                    },
                )
            )
            await db.commit()

            templates = await list_task_brief_templates(db=db, current_user=user)
            self.assertEqual(templates["code"], 0)
            default_tpl = next((item for item in templates["data"]["items"] if item.get("id") == "tpl_default_signal_planner"), None)
            self.assertIsNotNone(default_tpl)
            self.assertIn("## Verification", str(default_tpl["content"]))
            self.assertNotEqual(str(default_tpl["content"]).strip(), DEFAULT_TEMPLATE_FALLBACK.strip())

            generated = await generate_task_brief(
                body=GenerateTaskBriefRequest(
                    title="Default Template Render",
                    changed_files=["backend/app/services/task_brief_service.py"],
                    scope_summary="验证默认模板占位符渲染",
                ),
                db=db,
                current_user=user,
            )
            self.assertEqual(generated["code"], 0)
            content = str(generated["data"]["brief"]["content"] or "")
            self.assertIn("backend/app/services/task_brief_service.py", content)
            self.assertIn("验证默认模板占位符渲染", content)
            self.assertNotIn("{{CHANGED_FILES}}", content)
            self.assertNotIn("{{SCOPE_SUMMARY}}", content)
            self.assertNotIn("{{GENERATED_AT}}", content)