345cc73965
P0-1 SqlCredentialStore/save_draft 由自提交改 flush，端点/服务统一 commit （新增 CredentialStore.commit() 统一提交点；token 刷新落库显式提交）； 补多凭据一请求中途失败整体回滚集成测试。 P0-2 启动校验 _fernet(enc_key) 快速失败 + catch-all Exception → ErrorEnvelope； credential_enc_key 改 SecretStr。 P0-3 run_job 异常分类：AppError 存 code+message，其余存通用文案不泄 str(exc)。 P0-4 评审/正文 SSE 失败先发 error 事件，尾部 commit 包 try/except。 P1-4 max_version 加 FOR UPDATE 行锁消除 TOCTOU。 P1-5 scan_overdue 谓词下推 + 批量 UPDATE RETURNING。 P1-10 移除 OAuth user_code 日志。 P2 provider_deps 改调网关 build_adapter；accept_service Committable Protocol； CORS 白名单收窄；request_id 安全字符集白名单；stdlib 日志接管；读端点 404 校验； httpx timeout；测试用合法 Fernet key；类型化响应模型(JobResponse/DimensionEntry/ ReviewConflictView/selling_points)+路由 ErrorEnvelope responses（供 codegen）。
572 lines
25 KiB
Python
572 lines
25 KiB
Python
"""M4 端到端：学文风(指纹+证据,版本化) + 漂移打分(第四审) + 回炉（真实 DB，零 token）。
|
||
|
||
证明 M4 闭环（DoD：学文风指纹+证据；漂移打分作第四审真持久化；回炉返新旧 diff）：
|
||
用例 1 · 学文风闭环（jobs 异步长任务，M4-c）
|
||
`POST /projects` → `POST .../style`(samples) → **202 `{job_id}`**（写一行 job 返回）；
|
||
BackgroundTask 经 `run_job` 自建独立 session 跑 `run_style_extraction`(analyst 假适配器
|
||
产 `StyleFingerprintResult`)→拆指纹→`SqlStyleFingerprintWriteRepo.append`(真落 pg)。
|
||
ASGITransport 下 `await client.post` 等 background task 跑完（同 M3 时序坑）；override
|
||
`get_session_factory`→`e2e_sm`(真 sessionmaker/同测试 engine/loop)，并 monkeypatch
|
||
`style.build_gateway_for_tier`→真 Gateway 包假适配器(用 background task 的真 session 记账)。
|
||
→ `GET /jobs/{id}` status=done(result.version/dims_count) → 断言 `style_fingerprint` 真有行
|
||
(dimensions_json/evidence_json/version)。再 `mode="update"` 学一次 → version+1。
|
||
用例 2 · 写章第四审（drift scoring）
|
||
学文风后 `assemble` 的 `stable_core` 含指纹 → 第四审能在 review_context 拿指纹对照。
|
||
`POST .../chapters/N/review`(SSE) → 断言 `section{name:"style"}` 帧 + `style{score,segments}`
|
||
帧真发；`GET .../reviews` 留痕行 `style`(dict) 已填；DB 真源 `chapter_reviews.style` 真落库
|
||
(score + segments)。四审齐（continuity/foreshadow/pace/style）无 error。
|
||
用例 3 · 回炉（refine，非持久 / 不变量 #3）
|
||
`POST .../chapters/N/refine` → 200 `{original, refined}`；断言 `refined != original`
|
||
（writer 假适配器据 `req.output_schema is None` 返改写串）+ 不写库。
|
||
|
||
确定性 & 零成本（同 M1/M2/M3）：真实 `Gateway` + 多档位假适配器（据 `req.output_schema`
|
||
分支返回固定 `parsed`/`text`，绝不联网）+ 真实 `SqlAlchemyLedgerSink`。无 pg → skip。
|
||
|
||
坑（见 memory/gotchas）：
|
||
- 学文风 BackgroundTask 在 `run_job` 自建独立 session 上**自造网关**(`build_gateway_for_tier`,
|
||
从凭据建真 OpenAI 适配器) → E2E 无真凭据，须 monkeypatch `style.build_gateway_for_tier`
|
||
返真 `Gateway`(包假适配器, ledger 绑入参 session)；写侧 `SqlStyleFingerprintWriteRepo` 保持
|
||
真实 → 指纹真落 pg。`run_job` 默认 `repo_factory`→真 `SqlJobRepo`（落真 pg job 行），不动。
|
||
- `get_session_factory` 必 override → `e2e_sm`（真 sessionmaker，同测试 engine/loop），否则
|
||
默认 `get_sessionmaker()` 另建 engine 绑别的 loop（跨 loop 坑）。
|
||
- `get_sessionmaker` engine 绑首个 loop → 每 DB 测试清缓存重建、结束 dispose（同 M3）。
|
||
- ASGITransport 默认 `raise_app_exceptions=True`；SSE 解析 `event:`/`data:` 块；snake_case。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import uuid
|
||
from collections.abc import AsyncIterator, Callable
|
||
from typing import Annotated
|
||
|
||
import httpx
|
||
import pytest
|
||
from asgi_lifespan import LifespanManager
|
||
from fastapi import Depends
|
||
from sqlalchemy import delete, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||
from ww_agents import (
|
||
ContinuityReview,
|
||
ForeshadowReview,
|
||
ForeshadowSuggestion,
|
||
PaceIssue,
|
||
PaceReview,
|
||
StyleDimension,
|
||
StyleDriftReview,
|
||
StyleDriftSegment,
|
||
StyleFingerprintResult,
|
||
)
|
||
from ww_api.services.digest_extraction import ChapterDigestFacts
|
||
from ww_db import get_session, get_sessionmaker
|
||
from ww_db.models import (
|
||
Chapter,
|
||
ChapterDigest,
|
||
ChapterReview,
|
||
Foreshadow,
|
||
Job,
|
||
Outline,
|
||
Project,
|
||
StyleFingerprint,
|
||
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
|
||
|
||
# 三档位（writer/analyst/light）默认都路由到 deepseek（config.tier_defaults）。
|
||
_PROVIDER = "deepseek"
|
||
|
||
# 确定性流式写章 token。
|
||
_TOKENS = ["第", "一", "章", "：", "文风漂移之时。"]
|
||
|
||
# 学文风假产物：两维指纹，每维带原文证据（StyleFingerprintResult 提取轨）。
|
||
_FINGERPRINT = StyleFingerprintResult(
|
||
dimensions=[
|
||
StyleDimension(name="句长节奏", value="短句为主", evidence=["他来了。", "她走了。"]),
|
||
StyleDimension(name="叙事人称", value="第三人称", evidence=["他看着远方的山。"]),
|
||
]
|
||
)
|
||
|
||
# 第四审（文风漂移）假产物：整体相似度 + 一个低相似段。
|
||
_STYLE_DRIFT = StyleDriftReview(
|
||
score=72,
|
||
segments=[StyleDriftSegment(idx=2, score=40, label="机翻腔")],
|
||
)
|
||
|
||
# 三审固定产物（与 M3 一致，让四审图跑通）。
|
||
_FORESHADOW_REVIEW = ForeshadowReview(
|
||
planted=[ForeshadowSuggestion(code="F9", title="新埋：山门禁制", where="第2段")],
|
||
resolved=[],
|
||
)
|
||
_PACE_REVIEW = PaceReview(
|
||
water=[PaceIssue(where="第3段", reason="信息密度低")],
|
||
hook=True,
|
||
beat_map=[1, 3, 5, 2],
|
||
)
|
||
|
||
# 回炉改写标记（writer 纯文本返回）：保证 refined != original。
|
||
_REFINE_PREFIX = "【已回炉】"
|
||
|
||
# 各档位假用量（喂记账；证明各档位调用各自落库）。
|
||
_USAGE = {
|
||
"writer": ProviderUsage(input_tokens=11, output_tokens=5),
|
||
"analyst": ProviderUsage(input_tokens=23, output_tokens=7),
|
||
"light": ProviderUsage(input_tokens=17, output_tokens=9),
|
||
}
|
||
|
||
|
||
def _req_text(req: LlmRequest) -> str:
|
||
"""把 `LlmRequest.input`（str | list[Block]）压成纯文本（供回炉改写引用原文）。"""
|
||
if isinstance(req.input, str):
|
||
return req.input
|
||
return "\n".join(block.text for block in req.input)
|
||
|
||
|
||
class _FakeMultiTierAdapter:
|
||
"""实现 `ProviderAdapter` Protocol：按 `req.output_schema` 分支返回 parsed/text，绝不联网。
|
||
|
||
- `complete()`：据 `req.output_schema` 路由：
|
||
`StyleFingerprintResult`→学文风提取（analyst）；四审 schema
|
||
（`ContinuityReview`/`ForeshadowReview`/`PaceReview`/`StyleDriftReview`）→各审；
|
||
`output_schema is None`→**回炉纯文本**（writer，返改写串，保证 refined≠original）；
|
||
其余视作 digest 提炼（light）。
|
||
- `stream()`：吐固定写章 token（writer/draft），末尾喂 writer 用量。
|
||
每路带 `ProviderUsage` → 网关据此各落一条 usage_ledger。
|
||
"""
|
||
|
||
provider = _PROVIDER
|
||
|
||
def capabilities(self) -> Capabilities:
|
||
return Capabilities(structured_output=True)
|
||
|
||
async def complete(self, req: LlmRequest, model: str) -> ProviderResult:
|
||
schema = req.output_schema
|
||
if schema is None:
|
||
# 回炉：writer 纯文本重写选中段（保证 refined != original）。
|
||
return ProviderResult(
|
||
text=_REFINE_PREFIX + _req_text(req),
|
||
parsed=None,
|
||
usage=_USAGE["writer"],
|
||
)
|
||
if schema is StyleFingerprintResult:
|
||
return ProviderResult(
|
||
text=_FINGERPRINT.model_dump_json(),
|
||
parsed=_FINGERPRINT,
|
||
usage=_USAGE["analyst"],
|
||
)
|
||
if schema is StyleDriftReview:
|
||
return ProviderResult(
|
||
text=_STYLE_DRIFT.model_dump_json(),
|
||
parsed=_STYLE_DRIFT,
|
||
usage=_USAGE["light"],
|
||
)
|
||
if schema is ContinuityReview:
|
||
review = ContinuityReview(conflicts=[]) # 零冲突
|
||
return ProviderResult(
|
||
text=review.model_dump_json(), parsed=review, usage=_USAGE["analyst"]
|
||
)
|
||
if schema is ForeshadowReview:
|
||
return ProviderResult(
|
||
text=_FORESHADOW_REVIEW.model_dump_json(),
|
||
parsed=_FORESHADOW_REVIEW,
|
||
usage=_USAGE["analyst"],
|
||
)
|
||
if schema is PaceReview:
|
||
return ProviderResult(
|
||
text=_PACE_REVIEW.model_dump_json(), parsed=_PACE_REVIEW, usage=_USAGE["light"]
|
||
)
|
||
# 否则视为 digest 提炼（ChapterDigestFacts）。
|
||
facts = ChapterDigestFacts(summary="终稿摘要", events=["事件"], locations=["地点"])
|
||
return ProviderResult(text=facts.model_dump_json(), parsed=facts, usage=_USAGE["light"])
|
||
|
||
async def stream(self, req: LlmRequest, model: str) -> AsyncIterator[StreamChunk]:
|
||
for token in _TOKENS:
|
||
yield StreamChunk(text=token)
|
||
yield StreamChunk(usage=_USAGE["writer"])
|
||
|
||
|
||
@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()
|
||
|
||
|
||
def _parse_sse(raw: str) -> list[tuple[str, str]]:
|
||
"""把 text/event-stream 原文解析为 `(event, data)` 帧列表。"""
|
||
frames: list[tuple[str, str]] = []
|
||
event: str | None = None
|
||
data: str | None = None
|
||
for line in raw.splitlines():
|
||
if line.startswith("event:"):
|
||
event = line[len("event:") :].strip()
|
||
elif line.startswith("data:"):
|
||
data = line[len("data:") :].strip()
|
||
elif line == "":
|
||
if event is not None and data is not None:
|
||
frames.append((event, data))
|
||
event, data = None, None
|
||
if event is not None and data is not None:
|
||
frames.append((event, data))
|
||
return frames
|
||
|
||
|
||
def _gateway_override_factory(
|
||
adapter: _FakeMultiTierAdapter,
|
||
) -> Callable[[AsyncSession], Gateway]:
|
||
"""构造依赖覆盖：真实 Gateway + 给定假适配器 + 真实 ledger（请求 session）。"""
|
||
|
||
def _override(session: Annotated[AsyncSession, Depends(get_session)]) -> Gateway:
|
||
return Gateway(
|
||
adapters={_PROVIDER: adapter},
|
||
ledger=SqlAlchemyLedgerSink(session),
|
||
resolver=resolve_route,
|
||
)
|
||
|
||
return _override
|
||
|
||
|
||
def _patched_build_gateway(
|
||
adapter: _FakeMultiTierAdapter,
|
||
) -> Callable[..., object]:
|
||
"""构造 monkeypatch 替身：`style.build_gateway_for_tier`→真 Gateway 包假适配器。
|
||
|
||
学文风 BackgroundTask 在 `run_job` 自建的独立 session 上自造网关（从凭据建真 OpenAI
|
||
适配器）——E2E 无真凭据，故替换成返真 `Gateway`(包假适配器, ledger 绑 background task
|
||
的真 session)。写侧 `SqlStyleFingerprintWriteRepo` + `run_job` 默认 `SqlJobRepo` 保持
|
||
真实 → 指纹/job 真落 pg。
|
||
"""
|
||
|
||
async def _build(session: AsyncSession, store: object, tier: str) -> Gateway:
|
||
return Gateway(
|
||
adapters={_PROVIDER: adapter},
|
||
ledger=SqlAlchemyLedgerSink(session),
|
||
resolver=resolve_route,
|
||
)
|
||
|
||
return _build
|
||
|
||
|
||
async def _cleanup(e2e_sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> None:
|
||
"""按 FK 顺序清理（多数子表经 FK CASCADE；显式删可独立行）。"""
|
||
async with e2e_sm() as cleanup:
|
||
await cleanup.execute(delete(UsageLedger).where(UsageLedger.project_id == project_uuid))
|
||
await cleanup.execute(delete(Job).where(Job.project_id == project_uuid))
|
||
await cleanup.execute(
|
||
delete(StyleFingerprint).where(StyleFingerprint.project_id == project_uuid)
|
||
)
|
||
await cleanup.execute(delete(ChapterDigest).where(ChapterDigest.project_id == project_uuid))
|
||
await cleanup.execute(delete(ChapterReview).where(ChapterReview.project_id == project_uuid))
|
||
await cleanup.execute(delete(Outline).where(Outline.project_id == project_uuid))
|
||
await cleanup.execute(delete(Foreshadow).where(Foreshadow.project_id == project_uuid))
|
||
await cleanup.execute(delete(Chapter).where(Chapter.project_id == project_uuid))
|
||
await cleanup.execute(delete(Project).where(Project.id == project_uuid))
|
||
await cleanup.commit()
|
||
|
||
|
||
async def test_m4_learn_style_job_done_and_fingerprint_persisted(
|
||
e2e_sm: async_sessionmaker[AsyncSession],
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""用例 1：学文风闭环 → 202 job → 后台跑完 → job done → 指纹真落 pg（含 update version+1）。"""
|
||
import ww_api.routers.style as style_mod
|
||
from ww_api.main import create_app
|
||
from ww_api.services.project_deps import (
|
||
get_session_factory,
|
||
get_style_extract_gateway,
|
||
)
|
||
|
||
adapter = _FakeMultiTierAdapter()
|
||
override = _gateway_override_factory(adapter)
|
||
|
||
app = create_app()
|
||
# 凭据探测（请求阶段）→ 注假网关，绕过无凭据 503。
|
||
app.dependency_overrides[get_style_extract_gateway] = override
|
||
# BackgroundTask 独立 session 工厂 → 真 sessionmaker（同测试 engine/loop）。
|
||
app.dependency_overrides[get_session_factory] = lambda: e2e_sm
|
||
# BackgroundTask 自造网关 → 替成真 Gateway 包假适配器（写侧 repo + SqlJobRepo 保持真）。
|
||
monkeypatch.setattr(style_mod, "build_gateway_for_tier", _patched_build_gateway(adapter))
|
||
|
||
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": "M4 学文风验证作品", "genre": "玄幻"}
|
||
)
|
||
assert create_resp.status_code == 201
|
||
project_id = create_resp.json()["id"]
|
||
project_uuid = uuid.UUID(project_id)
|
||
|
||
# 1) 学文风（首学）→ 202 {job_id}（ASGITransport 等 background task 跑完）。
|
||
learn_resp = await client.post(
|
||
f"/projects/{project_id}/style",
|
||
json={"samples": ["他来了。她走了。", "他看着远方的山。"], "mode": "create"},
|
||
)
|
||
assert learn_resp.status_code == 202
|
||
job_id = learn_resp.json()["job_id"]
|
||
|
||
# 2) 轮询 job → done，result 含 version/dims_count。
|
||
job_resp = await client.get(f"/jobs/{job_id}")
|
||
assert job_resp.status_code == 200
|
||
job = job_resp.json()
|
||
assert job["status"] == "done"
|
||
assert job["kind"] == "style_learn"
|
||
assert job["result"]["version"] == 1
|
||
assert job["result"]["dims_count"] == 2
|
||
|
||
# 3) GET /style → 最新指纹（完整维度 + 证据 + 版本）。
|
||
style_resp = await client.get(f"/projects/{project_id}/style")
|
||
assert style_resp.status_code == 200
|
||
fp = style_resp.json()
|
||
assert fp["version"] == 1
|
||
# 新契约：dimensions 为 list[DimensionEntry{name,value,evidence}]（不再裸 dict）。
|
||
dims = {d["name"]: d for d in fp["dimensions"]}
|
||
assert dims["句长节奏"]["value"] == "短句为主"
|
||
assert dims["句长节奏"]["evidence"] == ["他来了。", "她走了。"]
|
||
|
||
# 4) mode="update" 再学一次 → version+1。
|
||
update_resp = await client.post(
|
||
f"/projects/{project_id}/style",
|
||
json={"samples": ["新样本正文。"], "mode": "update"},
|
||
)
|
||
assert update_resp.status_code == 202
|
||
update_job = (await client.get(f"/jobs/{update_resp.json()['job_id']}")).json()
|
||
assert update_job["status"] == "done"
|
||
assert update_job["result"]["version"] == 2
|
||
|
||
# DB 真源断言。
|
||
assert project_uuid is not None
|
||
async with e2e_sm() as verify:
|
||
fp_rows = (
|
||
(
|
||
await verify.execute(
|
||
select(StyleFingerprint)
|
||
.where(StyleFingerprint.project_id == project_uuid)
|
||
.order_by(StyleFingerprint.version)
|
||
)
|
||
)
|
||
.scalars()
|
||
.all()
|
||
)
|
||
# 两次学文风 → 两行（版本化 INSERT，不覆盖历史）。
|
||
assert [r.version for r in fp_rows] == [1, 2]
|
||
first = fp_rows[0]
|
||
assert first.dimensions_json == {"句长节奏": "短句为主", "叙事人称": "第三人称"}
|
||
assert first.evidence_json["句长节奏"] == ["他来了。", "她走了。"]
|
||
|
||
# job 真落 pg：两行 done。
|
||
job_rows = (
|
||
(await verify.execute(select(Job).where(Job.project_id == project_uuid)))
|
||
.scalars()
|
||
.all()
|
||
)
|
||
assert len(job_rows) == 2
|
||
assert all(r.status == "done" and r.kind == "style_learn" for r in job_rows)
|
||
finally:
|
||
if project_uuid is not None:
|
||
await _cleanup(e2e_sm, project_uuid)
|
||
|
||
|
||
async def test_m4_review_fourth_style_audit_sse_and_trace(
|
||
e2e_sm: async_sessionmaker[AsyncSession],
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""用例 2：写章第四审（drift）→ SSE section{style} + style{score,segments} + style 列真填。
|
||
|
||
先学文风 → `assemble` 的 stable_core 含指纹 → 第四审能在 review_context 拿指纹对照打分。
|
||
四审齐（continuity/foreshadow/pace/style）无 error。
|
||
"""
|
||
import ww_api.routers.style as style_mod
|
||
from ww_api.main import create_app
|
||
from ww_api.services.project_deps import (
|
||
get_review_gateway,
|
||
get_session_factory,
|
||
get_style_extract_gateway,
|
||
get_writer_gateway,
|
||
)
|
||
|
||
adapter = _FakeMultiTierAdapter()
|
||
override = _gateway_override_factory(adapter)
|
||
|
||
app = create_app()
|
||
app.dependency_overrides[get_writer_gateway] = override
|
||
app.dependency_overrides[get_review_gateway] = override
|
||
app.dependency_overrides[get_style_extract_gateway] = override
|
||
app.dependency_overrides[get_session_factory] = lambda: e2e_sm
|
||
monkeypatch.setattr(style_mod, "build_gateway_for_tier", _patched_build_gateway(adapter))
|
||
|
||
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": "M4 第四审验证作品"})
|
||
assert create_resp.status_code == 201
|
||
project_id = create_resp.json()["id"]
|
||
project_uuid = uuid.UUID(project_id)
|
||
|
||
# 1) 学文风 → 指纹入库（供 assemble 注入第四审做对照）。
|
||
learn_resp = await client.post(
|
||
f"/projects/{project_id}/style",
|
||
json={"samples": ["他来了。她走了。"], "mode": "create"},
|
||
)
|
||
assert learn_resp.status_code == 202
|
||
assert (await client.get(f"/jobs/{learn_resp.json()['job_id']}")).json()[
|
||
"status"
|
||
] == "done"
|
||
|
||
# 2) 写第1章草稿 + 自动保存（供续审）。
|
||
draft_resp = await client.post(f"/projects/{project_id}/chapters/1/draft")
|
||
draft_text = "".join(
|
||
json.loads(d)["text"]
|
||
for (ev, d) in _parse_sse(draft_resp.text)
|
||
if ev == "token"
|
||
)
|
||
await client.put(
|
||
f"/projects/{project_id}/chapters/1/draft",
|
||
json={"text": draft_text},
|
||
)
|
||
|
||
# 3) 续审（SSE）：四审齐 + section{style} + style{score,segments}，无 error。
|
||
review_resp = await client.post(
|
||
f"/projects/{project_id}/chapters/1/review",
|
||
json={"draft": draft_text},
|
||
)
|
||
assert review_resp.status_code == 200
|
||
assert review_resp.headers["content-type"].startswith("text/event-stream")
|
||
rframes = _parse_sse(review_resp.text)
|
||
assert [d for (ev, d) in rframes if ev == "error"] == []
|
||
section_names = {
|
||
json.loads(d).get("name") for (ev, d) in rframes if ev == "section"
|
||
}
|
||
assert {"continuity", "foreshadow", "pace", "style"} <= section_names
|
||
style_events = [json.loads(d) for (ev, d) in rframes if ev == "style"]
|
||
assert len(style_events) == 1
|
||
style_ev = style_events[0]
|
||
assert style_ev["score"] == 72
|
||
assert style_ev["segments"][0]["idx"] == 2
|
||
assert style_ev["segments"][0]["score"] == 40
|
||
assert style_ev["segments"][0]["label"] == "机翻腔"
|
||
|
||
# 4) GET 历史：留痕行 style(dict) 已填。
|
||
hist_resp = await client.get(f"/projects/{project_id}/chapters/1/reviews")
|
||
assert hist_resp.status_code == 200
|
||
latest = hist_resp.json()["reviews"][0]
|
||
assert isinstance(latest["style"], dict)
|
||
assert latest["style"]["score"] == 72
|
||
|
||
# DB 真源断言：chapter_reviews.style 列真填。
|
||
assert project_uuid is not None
|
||
async with e2e_sm() as verify:
|
||
review_rows = (
|
||
(
|
||
await verify.execute(
|
||
select(ChapterReview).where(ChapterReview.project_id == project_uuid)
|
||
)
|
||
)
|
||
.scalars()
|
||
.all()
|
||
)
|
||
assert len(review_rows) == 1
|
||
row = review_rows[0]
|
||
assert isinstance(row.style, dict)
|
||
assert row.style["score"] == 72
|
||
assert row.style["segments"][0]["idx"] == 2
|
||
assert row.style["segments"][0]["label"] == "机翻腔"
|
||
finally:
|
||
if project_uuid is not None:
|
||
await _cleanup(e2e_sm, project_uuid)
|
||
|
||
|
||
async def test_m4_refine_returns_diff_without_persisting(
|
||
e2e_sm: async_sessionmaker[AsyncSession],
|
||
) -> None:
|
||
"""用例 3：回炉 → 200 {original, refined}；refined != original；不写库（不变量 #3）。"""
|
||
from ww_api.main import create_app
|
||
from ww_api.services.project_deps import get_refine_gateway
|
||
|
||
adapter = _FakeMultiTierAdapter()
|
||
override = _gateway_override_factory(adapter)
|
||
|
||
app = create_app()
|
||
app.dependency_overrides[get_refine_gateway] = override
|
||
|
||
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": "M4 回炉验证作品"})
|
||
assert create_resp.status_code == 201
|
||
project_id = create_resp.json()["id"]
|
||
project_uuid = uuid.UUID(project_id)
|
||
|
||
segment = "他緩緩走進房間。"
|
||
refine_resp = await client.post(
|
||
f"/projects/{project_id}/chapters/1/refine",
|
||
json={"segment": segment, "instruction": "去掉机翻腔"},
|
||
)
|
||
assert refine_resp.status_code == 200
|
||
body = refine_resp.json()
|
||
assert body["original"] == segment
|
||
# writer 假适配器返改写串（前缀），保证 refined != original。
|
||
assert body["refined"] != body["original"]
|
||
assert body["refined"].startswith(_REFINE_PREFIX)
|
||
# 指令进了输入文本（writer 据 output_schema is None 走纯文本路径）。
|
||
assert "去掉机翻腔" in body["refined"]
|
||
|
||
# DB 真源断言：回炉不写库 → 无 chapter/chapter_review 行（不变量 #3）。
|
||
assert project_uuid is not None
|
||
async with e2e_sm() as verify:
|
||
chapter_rows = (
|
||
(await verify.execute(select(Chapter).where(Chapter.project_id == project_uuid)))
|
||
.scalars()
|
||
.all()
|
||
)
|
||
assert chapter_rows == []
|
||
review_rows = (
|
||
(
|
||
await verify.execute(
|
||
select(ChapterReview).where(ChapterReview.project_id == project_uuid)
|
||
)
|
||
)
|
||
.scalars()
|
||
.all()
|
||
)
|
||
assert review_rows == []
|
||
# 但回炉记账落库（writer 网关 ledger，端点末尾 commit）。
|
||
ledger_rows = (
|
||
(
|
||
await verify.execute(
|
||
select(UsageLedger).where(UsageLedger.project_id == project_uuid)
|
||
)
|
||
)
|
||
.scalars()
|
||
.all()
|
||
)
|
||
assert len(ledger_rows) == 1
|
||
assert ledger_rows[0].input_tokens == _USAGE["writer"].input_tokens
|
||
finally:
|
||
if project_uuid is not None:
|
||
await _cleanup(e2e_sm, project_uuid)
|