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 范式
217 lines
8.9 KiB
Python
217 lines
8.9 KiB
Python
"""F1 拆书 teardown 落库成 rules 端到端（真 pg + mock 网关零 token）。
|
||
|
||
证明拆书入库闭环（不变量 #3：入库经白名单 gate；预览不写库）：
|
||
1. teardown `generate` → 200 预览（BookTeardownResult）；断言 `rules` 表零新增（预览不写库）。
|
||
2. teardown `ingest`（teardown 产物）→ 201；`rules` 表真落行（拆书结论拍平成可读规则条目）。
|
||
|
||
镜像 `tests/test_t6_toolbox_e2e.py`：真 `Gateway` + 据 `req.output_schema` 分支返 parsed 的
|
||
假适配器（绝不联网）+ 真 `SqlAlchemyLedgerSink`。注入缝 `get_tier_gateway_builder`（generate
|
||
调 `build_gateway(spec.tier)`；rules ingest 无 continuity 预检故不调网关）。无 pg → skip。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
from collections.abc import AsyncIterator, Awaitable, Callable
|
||
from typing import Annotated
|
||
|
||
import httpx
|
||
import pytest
|
||
from asgi_lifespan import LifespanManager
|
||
from fastapi import Depends
|
||
from sqlalchemy import delete, func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||
from ww_agents import BookTeardownResult
|
||
from ww_db import get_session, get_sessionmaker
|
||
from ww_db.models import Project, Rule, UsageLedger
|
||
from ww_llm_gateway import Gateway, SqlAlchemyLedgerSink, resolve_route
|
||
from ww_llm_gateway.adapters.base import (
|
||
Capabilities,
|
||
ProviderResult,
|
||
ProviderUsage,
|
||
StreamChunk,
|
||
)
|
||
from ww_llm_gateway.types import LlmRequest, Tier
|
||
|
||
_PROVIDER = "deepseek"
|
||
_USAGE = ProviderUsage(input_tokens=29, output_tokens=13)
|
||
|
||
# teardown 假产物：themes/archetypes/structure/hooks 全非空（拍平成 4 条 rules）。
|
||
_TEARDOWN = BookTeardownResult(
|
||
themes=["逆袭打脸", "守护苍生"],
|
||
archetypes=["扮猪吃虎的主角", "口嫌体正的女主"],
|
||
structure="开篇废柴受辱 → 奇遇得宝 → 步步攀升 → 终战封神",
|
||
hooks=["每章末留悬念", "三章一小高潮"],
|
||
)
|
||
|
||
|
||
class _FakeTeardownAdapter:
|
||
"""实现 `ProviderAdapter`：仅 BookTeardownResult 一路（拆书 generate），绝不联网。"""
|
||
|
||
provider = _PROVIDER
|
||
|
||
def capabilities(self) -> Capabilities:
|
||
return Capabilities(structured_output=True)
|
||
|
||
async def complete(self, req: LlmRequest, model: str) -> ProviderResult:
|
||
if req.output_schema is BookTeardownResult:
|
||
return ProviderResult(text=_TEARDOWN.model_dump_json(), parsed=_TEARDOWN, usage=_USAGE)
|
||
raise AssertionError(f"unexpected output_schema in F1 fake adapter: {req.output_schema!r}")
|
||
|
||
async def stream(self, req: LlmRequest, model: str) -> AsyncIterator[StreamChunk]:
|
||
yield StreamChunk(usage=_USAGE)
|
||
raise AssertionError("teardown generation must not stream")
|
||
|
||
|
||
@pytest.fixture
|
||
async def e2e_sm() -> AsyncIterator[async_sessionmaker[AsyncSession]]:
|
||
"""真 DB session 工厂；无 pg 时跳过。"""
|
||
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()
|
||
|
||
|
||
def _tier_builder_override(
|
||
adapter: _FakeTeardownAdapter,
|
||
) -> Callable[[AsyncSession], Callable[[Tier], Awaitable[Gateway]]]:
|
||
def _override(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> Callable[[Tier], Awaitable[Gateway]]:
|
||
async def _build(_tier: Tier) -> Gateway:
|
||
return Gateway(
|
||
adapters={adapter.provider: adapter},
|
||
ledger=SqlAlchemyLedgerSink(session),
|
||
resolver=resolve_route,
|
||
)
|
||
|
||
return _build
|
||
|
||
return _override
|
||
|
||
|
||
async def _rule_count(sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> int:
|
||
async with sm() as s:
|
||
return int(
|
||
(
|
||
await s.execute(
|
||
select(func.count()).select_from(Rule).where(Rule.project_id == project_uuid)
|
||
)
|
||
).scalar_one()
|
||
)
|
||
|
||
|
||
async def _cleanup(sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> None:
|
||
async with sm() as cleanup:
|
||
await cleanup.execute(delete(UsageLedger).where(UsageLedger.project_id == project_uuid))
|
||
await cleanup.execute(delete(Rule).where(Rule.project_id == project_uuid))
|
||
await cleanup.execute(delete(Project).where(Project.id == project_uuid))
|
||
await cleanup.commit()
|
||
|
||
|
||
async def test_teardown_generate_preview_does_not_write_rules(
|
||
e2e_sm: async_sessionmaker[AsyncSession],
|
||
) -> None:
|
||
"""用例 1（负向）：teardown generate → 200 预览；rules 表零新增（不变量 #3 预览不写库）。"""
|
||
from ww_api.main import create_app
|
||
from ww_api.services.project_deps import get_tier_gateway_builder
|
||
|
||
app = create_app()
|
||
app.dependency_overrides[get_tier_gateway_builder] = _tier_builder_override(
|
||
_FakeTeardownAdapter()
|
||
)
|
||
transport = httpx.ASGITransport(app=app)
|
||
project_uuid: uuid.UUID | None = None
|
||
try:
|
||
async with LifespanManager(app):
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||
create_resp = await client.post("/projects", json={"title": "F1 拆书预览作品"})
|
||
assert create_resp.status_code == 201
|
||
project_id = create_resp.json()["id"]
|
||
project_uuid = uuid.UUID(project_id)
|
||
|
||
gen_resp = await client.post(
|
||
f"/projects/{project_id}/skills/teardown/generate",
|
||
json={"text": "第一章……（待拆解的样章正文）", "kind": "某爆款玄幻"},
|
||
)
|
||
assert gen_resp.status_code == 200, gen_resp.text
|
||
body = gen_resp.json()
|
||
assert body["tool_key"] == "teardown"
|
||
assert body["output_kind"] == "BookTeardownResult"
|
||
assert body["preview"]["themes"] == ["逆袭打脸", "守护苍生"]
|
||
|
||
# DB 真源：预览不写 rules 业务表（守不变量 #3）。
|
||
assert project_uuid is not None
|
||
assert await _rule_count(e2e_sm, project_uuid) == 0
|
||
finally:
|
||
if project_uuid is not None:
|
||
await _cleanup(e2e_sm, project_uuid)
|
||
|
||
|
||
async def test_teardown_ingest_persists_rules(
|
||
e2e_sm: async_sessionmaker[AsyncSession],
|
||
) -> None:
|
||
"""用例 2：teardown ingest → 201；rules 表真落行（拆书结论拍平成可读规则条目）。"""
|
||
from ww_api.main import create_app
|
||
from ww_api.services.project_deps import get_tier_gateway_builder
|
||
|
||
app = create_app()
|
||
# ingest（rules 无 continuity 预检）不触达网关；override 防真凭据探测路径。
|
||
app.dependency_overrides[get_tier_gateway_builder] = _tier_builder_override(
|
||
_FakeTeardownAdapter()
|
||
)
|
||
transport = httpx.ASGITransport(app=app)
|
||
project_uuid: uuid.UUID | None = None
|
||
try:
|
||
async with LifespanManager(app):
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||
create_resp = await client.post("/projects", json={"title": "F1 拆书入库作品"})
|
||
assert create_resp.status_code == 201
|
||
project_id = create_resp.json()["id"]
|
||
project_uuid = uuid.UUID(project_id)
|
||
|
||
assert await _rule_count(e2e_sm, project_uuid) == 0
|
||
|
||
ingest_resp = await client.post(
|
||
f"/projects/{project_id}/skills/teardown/ingest",
|
||
json={
|
||
"teardown": {
|
||
"themes": _TEARDOWN.themes,
|
||
"archetypes": _TEARDOWN.archetypes,
|
||
"structure": _TEARDOWN.structure,
|
||
"hooks": _TEARDOWN.hooks,
|
||
}
|
||
},
|
||
)
|
||
assert ingest_resp.status_code == 201, ingest_resp.text
|
||
ok = ingest_resp.json()
|
||
assert ok["table"] == "rules"
|
||
# 四个非空字段 → 四条 rules（确定性顺序）。
|
||
assert ok["created"] == ["themes", "archetypes", "structure", "hooks"]
|
||
assert ok["rejected_tables"] == []
|
||
|
||
# DB 真源：rules 行真落 pg（拍平成可读条目，内容含拆书结论）。
|
||
assert project_uuid is not None
|
||
assert await _rule_count(e2e_sm, project_uuid) == 4
|
||
async with e2e_sm() as verify:
|
||
rows = (
|
||
(await verify.execute(select(Rule).where(Rule.project_id == project_uuid)))
|
||
.scalars()
|
||
.all()
|
||
)
|
||
blob = "\n".join(r.content for r in rows)
|
||
# 拍平内容须含拆书结论（断真落行 + 形变保留）。
|
||
assert "逆袭打脸" in blob
|
||
assert "扮猪吃虎的主角" in blob
|
||
assert "开篇废柴受辱" in blob
|
||
assert "每章末留悬念" in blob
|
||
finally:
|
||
if project_uuid is not None:
|
||
await _cleanup(e2e_sm, project_uuid)
|