后端: - 新持久化包 research_templates(表 ai_writing_research_templates,create_all 建表,注册 models/__init__),deps 接 research_template_store - 路由 /api/ai-writing/research-templates CRUD(在 ai_writing.py) - 内置范例 网摘/周报/要讯(DEFAULT_RESEARCH_TEMPLATES + seed_defaults,表空时由 list 接口幂等播种) - tests/test_research_templates.py(9 passed) 前端: - api/ai-writing-research-templates.ts + hooks/useResearchTemplates.ts - 配置页 ResearchTemplatesConfigPage(/page/canvas/research-templates)CRUD - ResearchSidebar 研究模板改用研究模板实体;点击回填整张表单并直接 startWriting 进入写作;顶部齿轮进配置页 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
125 lines
4.5 KiB
Python
125 lines
4.5 KiB
Python
"""Tests for the research-template store (课题研究模板,需求 0616 #5).
|
|
|
|
A research template bundles writing-form presets (article type / outline /
|
|
material source / word count / audience / strict mode / topic) for one-click
|
|
pre-fill. Covers create (with defaults) / list ordering / get / partial update /
|
|
delete.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from deerflow.persistence.base import Base
|
|
from deerflow.persistence.research_templates import ResearchTemplateRepository
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def store(tmp_path):
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'rt.db'}")
|
|
async with engine.begin() as conn:
|
|
import deerflow.persistence.research_templates.model # noqa: F401
|
|
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
sf = async_sessionmaker(engine, expire_on_commit=False)
|
|
return ResearchTemplateRepository(sf)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_applies_defaults(store: ResearchTemplateRepository):
|
|
row = await store.create(name="深度分析模板")
|
|
assert row["name"] == "深度分析模板"
|
|
assert row["material_source"] == "general"
|
|
assert row["word_count"] == 800
|
|
assert row["strict_mode"] is False
|
|
assert row["article_type_key"] is None
|
|
assert row["id"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_full_then_get_roundtrips(store: ResearchTemplateRepository):
|
|
created = await store.create(
|
|
name="舆情研究",
|
|
description="围绕舆情的研究模板",
|
|
article_type_key="analysis",
|
|
outline="一、背景\n二、分析\n三、结论",
|
|
material_source="notebook",
|
|
word_count=1500,
|
|
audience="决策者",
|
|
strict_mode=True,
|
|
topic="某品牌近期舆情",
|
|
sort_order=2,
|
|
)
|
|
fetched = await store.get(created["id"])
|
|
assert fetched is not None
|
|
assert fetched["article_type_key"] == "analysis"
|
|
assert fetched["outline"].startswith("一、背景")
|
|
assert fetched["material_source"] == "notebook"
|
|
assert fetched["word_count"] == 1500
|
|
assert fetched["audience"] == "决策者"
|
|
assert fetched["strict_mode"] is True
|
|
assert fetched["topic"] == "某品牌近期舆情"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_orders_by_sort_order(store: ResearchTemplateRepository):
|
|
await store.create(name="B", sort_order=2)
|
|
await store.create(name="A", sort_order=1)
|
|
rows = await store.list()
|
|
assert [r["name"] for r in rows] == ["A", "B"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_partial_keeps_other_fields(store: ResearchTemplateRepository):
|
|
created = await store.create(name="模板", word_count=800, strict_mode=False)
|
|
updated = await store.update(created["id"], word_count=2000, strict_mode=True)
|
|
assert updated is not None
|
|
assert updated["word_count"] == 2000
|
|
assert updated["strict_mode"] is True
|
|
assert updated["name"] == "模板" # untouched
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_missing_returns_none(store: ResearchTemplateRepository):
|
|
assert await store.update("nope", name="x") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete(store: ResearchTemplateRepository):
|
|
created = await store.create(name="待删")
|
|
assert await store.delete(created["id"]) is True
|
|
assert await store.get(created["id"]) is None
|
|
assert await store.delete(created["id"]) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_seed_defaults_inserts_builtins_when_empty(store: ResearchTemplateRepository):
|
|
inserted = await store.seed_defaults()
|
|
assert inserted == 3
|
|
rows = await store.list()
|
|
names = [r["name"] for r in rows]
|
|
assert names == ["网摘", "周报", "要讯"] # sort_order 0/1/2
|
|
# 内置模板内容完整(范例质量)。
|
|
by_name = {r["name"]: r for r in rows}
|
|
assert by_name["网摘"]["article_type_key"] == "news"
|
|
assert by_name["周报"]["strict_mode"] is True
|
|
assert by_name["要讯"]["outline"]
|
|
assert all(r["topic"] for r in rows)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_seed_defaults_is_idempotent(store: ResearchTemplateRepository):
|
|
assert await store.seed_defaults() == 3
|
|
# 已有模板 → 再次播种跳过,不重复、不覆盖。
|
|
assert await store.seed_defaults() == 0
|
|
assert len(await store.list()) == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_seed_defaults_skips_when_user_has_templates(store: ResearchTemplateRepository):
|
|
await store.create(name="用户自建")
|
|
assert await store.seed_defaults() == 0
|
|
assert [r["name"] for r in await store.list()] == ["用户自建"]
|