5d8e619408
- 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 范式
138 lines
5.6 KiB
Python
138 lines
5.6 KiB
Python
"""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:
|
||
"""用例 1：POST → 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 真源：模板行真落 pg（owner 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 → 204；DB 真源行被删。
|
||
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:
|
||
"""用例 2：title 空 → 422（schema min_length=1）；DB 无行落库。"""
|
||
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
|