test(qa): F1/F2/F3 E2E — 拆书入 rules / 续写式链前文注入 / 模板库 CRUD

- F1: teardown generate→ingest→rules 真落行 + 负向预览不写库(不变量 #3)
- F2: continue_volume 两章第二章写章请求含第一章 accepted 正文(前文注入)+ draft_volume 回归守卫
- F3: templates POST→GET→DELETE 真 pg 回环 + title/body 空→422

真 pg + mock 网关零 token,镜像现有 tests/test_*_e2e.py 范式
This commit is contained in:
Yaojia Wang
2026-06-23 20:24:24 +02:00
parent cfe6a535b0
commit 5d8e619408
3 changed files with 655 additions and 0 deletions

137
tests/test_templates_e2e.py Normal file
View File

@@ -0,0 +1,137 @@
"""F3 提示词/模板库 端到端(真 pg无网关——模板库不调 LLM
证明单用户本地模板库 CRUD 闭环:创建→列出→删除真 pg 回环 + 输入校验title/body 空 → 422
owner_id 全程 stub单用户原型。无 pg → skip。
镜像现有 `tests/test_*_e2e.py` 范式:真 `get_sessionmaker`(无 pg skip+ `LifespanManager` +
`ASGITransport`DB 真源逐项断言。模板库不触达 LLM 网关,故无需 mock 适配器。
"""
from __future__ import annotations
import uuid
from collections.abc import AsyncIterator
import httpx
import pytest
from asgi_lifespan import LifespanManager
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from ww_db import get_sessionmaker
from ww_db.models import PromptTemplate
@pytest.fixture
async def e2e_sm() -> AsyncIterator[async_sessionmaker[AsyncSession]]:
"""真 DB session 工厂;无 pg 时跳过(每测试清缓存重建 engine、结束 dispose"""
get_sessionmaker.cache_clear()
maker = get_sessionmaker()
try:
async with maker() as probe:
await probe.execute(select(1))
except Exception:
pytest.skip("postgres not reachable")
yield maker
await maker.kw["bind"].dispose()
get_sessionmaker.cache_clear()
async def _cleanup(sm: async_sessionmaker[AsyncSession], ids: list[uuid.UUID]) -> None:
if not ids:
return
async with sm() as cleanup:
await cleanup.execute(delete(PromptTemplate).where(PromptTemplate.id.in_(ids)))
await cleanup.commit()
async def _template_count(sm: async_sessionmaker[AsyncSession], template_id: uuid.UUID) -> int:
async with sm() as s:
return len(
(await s.execute(select(PromptTemplate).where(PromptTemplate.id == template_id)))
.scalars()
.all()
)
async def test_templates_create_list_delete_roundtrip(
e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
"""用例 1POST → GET → DELETE 真 pg 回环(行真落 pg删除后表无行"""
from ww_api.main import create_app
app = create_app()
transport = httpx.ASGITransport(app=app)
created_id: uuid.UUID | None = None
try:
async with LifespanManager(app):
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
# 1) POST → 201模板真落 pg。
create_resp = await client.post(
"/templates",
json={
"title": "玄幻开篇模板",
"body": "请按照黄金三章节奏,开篇即上钩……",
"category": "开篇",
"tool_key": "opening",
},
)
assert create_resp.status_code == 201, create_resp.text
created = create_resp.json()
assert created["title"] == "玄幻开篇模板"
assert created["body"] == "请按照黄金三章节奏,开篇即上钩……"
assert created["category"] == "开篇"
assert created["tool_key"] == "opening"
created_id = uuid.UUID(created["id"])
# DB 真源:模板行真落 pgowner stub
assert await _template_count(e2e_sm, created_id) == 1
# 2) GET → 列表含刚建模板。
list_resp = await client.get("/templates")
assert list_resp.status_code == 200
listed = list_resp.json()
assert created["id"] in [t["id"] for t in listed]
mine = next(t for t in listed if t["id"] == created["id"])
assert mine["title"] == "玄幻开篇模板"
assert mine["body"] == "请按照黄金三章节奏,开篇即上钩……"
# 3) DELETE → 204DB 真源行被删。
del_resp = await client.delete(f"/templates/{created['id']}")
assert del_resp.status_code == 204
assert await _template_count(e2e_sm, created_id) == 0
# 删除后再列出不含该模板。
after = (await client.get("/templates")).json()
assert created["id"] not in [t["id"] for t in after]
# 4) 删除不存在 → 404 NOT_FOUND envelope。
missing = await client.delete(f"/templates/{uuid.uuid4()}")
assert missing.status_code == 404
assert missing.json()["error"]["code"] == "NOT_FOUND"
finally:
if created_id is not None:
await _cleanup(e2e_sm, [created_id])
async def test_templates_create_blank_title_returns_422(
e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
"""用例 2title 空 → 422schema min_length=1DB 无行落库。"""
from ww_api.main import create_app
app = create_app()
transport = httpx.ASGITransport(app=app)
async with LifespanManager(app):
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
before = len((await client.get("/templates")).json())
resp = await client.post("/templates", json={"title": "", "body": "非空正文"})
assert resp.status_code == 422, resp.text
# body 空亦 422守输入校验纪律
resp_body = await client.post("/templates", json={"title": "有标题", "body": ""})
assert resp_body.status_code == 422
# 两次 422 均未落库(列表数量不变)。
after = len((await client.get("/templates")).json())
assert after == before