c596a8d342
GET /outline?volume=N 原在端点 Python 侧过滤全部章节；现把 volume 条件下推到 OutlineRepo.list_for_project 的 DB WHERE，只查该卷行（少读多余卷 + 语义正确）。 Protocol/SqlOutlineRepo/各 OutlineRepo fake 同步加 keyword-only volume；补 repo 级 下推断言测试。向后兼容：不传 volume 仍返回全部（默认 None）。
344 lines
14 KiB
Python
344 lines
14 KiB
Python
"""T3.5 端点测试：大纲生成 + 持久化 + 伏笔看板（内存替身，无 DB/无网络）。
|
||
|
||
覆盖：
|
||
- POST /outline：mock 网关产 `OutlineResult` → 逐章 upsert（含 foreshadow_windows）→ 端点 commit；
|
||
- 幂等：同 (project_id, chapter_no) 再生成覆盖；
|
||
- 项目不存在 → 404；无凭据 → LLM_UNAVAILABLE（503）；
|
||
- GET /foreshadow?status=：按状态过滤；非法 status → 422 VALIDATION。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import uuid
|
||
|
||
import httpx
|
||
import pytest
|
||
from cryptography.fernet import Fernet
|
||
from fakes_projects import (
|
||
FakeForeshadowRepo,
|
||
FakeOutlineReadRepo,
|
||
FakeOutlineWriteRepo,
|
||
FakeProjectRepo,
|
||
FakeReviewGateway,
|
||
FakeSession,
|
||
)
|
||
from test_projects import _empty_memory_repos
|
||
from ww_agents import ForeshadowWindow, OutlineChapter, OutlineResult
|
||
from ww_core.domain.project_repo import ProjectCreate
|
||
from ww_shared import AppError, ErrorCode
|
||
|
||
|
||
def _outline_result() -> OutlineResult:
|
||
return OutlineResult(
|
||
chapters=[
|
||
OutlineChapter(
|
||
no=1,
|
||
beats=["开篇引入主角", "埋下信物伏笔"],
|
||
foreshadow_windows=[
|
||
ForeshadowWindow(code="F1", plant_chapter=1, expected_close_to=10)
|
||
],
|
||
),
|
||
OutlineChapter(no=2, beats=["冲突升级"], foreshadow_windows=[]),
|
||
]
|
||
)
|
||
|
||
|
||
def _make_client(
|
||
*,
|
||
project_repo: FakeProjectRepo | None = None,
|
||
foreshadow_repo: FakeForeshadowRepo | None = None,
|
||
outline_repo: FakeOutlineWriteRepo | None = None,
|
||
gateway: FakeReviewGateway | None = None,
|
||
session: FakeSession | None = None,
|
||
) -> tuple[httpx.AsyncClient, FakeOutlineWriteRepo, FakeSession]:
|
||
os.environ.setdefault("CREDENTIAL_ENC_KEY", Fernet.generate_key().decode())
|
||
from ww_api.main import create_app
|
||
from ww_api.services.project_deps import (
|
||
get_foreshadow_repo,
|
||
get_memory_repos,
|
||
get_outline_gateway,
|
||
get_outline_write_repo,
|
||
get_project_repo,
|
||
)
|
||
from ww_db import get_session
|
||
|
||
project_repo = project_repo or FakeProjectRepo()
|
||
foreshadow_repo = foreshadow_repo or FakeForeshadowRepo()
|
||
outline_repo = outline_repo or FakeOutlineWriteRepo()
|
||
gateway = gateway or FakeReviewGateway(parsed=_outline_result())
|
||
session = session or FakeSession()
|
||
|
||
app = create_app()
|
||
app.dependency_overrides[get_project_repo] = lambda: project_repo
|
||
app.dependency_overrides[get_foreshadow_repo] = lambda: foreshadow_repo
|
||
app.dependency_overrides[get_memory_repos] = _empty_memory_repos
|
||
app.dependency_overrides[get_outline_write_repo] = lambda: outline_repo
|
||
app.dependency_overrides[get_session] = lambda: session
|
||
app.dependency_overrides[get_outline_gateway] = lambda: gateway
|
||
|
||
transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)
|
||
client = httpx.AsyncClient(transport=transport, base_url="http://test")
|
||
return client, outline_repo, session
|
||
|
||
|
||
async def _seed_project(repo: FakeProjectRepo) -> uuid.UUID:
|
||
view = await repo.create(uuid.UUID(int=1), ProjectCreate(title="测试作品", genre="玄幻"))
|
||
return uuid.UUID(str(view.id))
|
||
|
||
|
||
# ---- 大纲生成 ----
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_outline_upserts_chapters_and_commits() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
client, outline_repo, session = _make_client(project_repo=project_repo)
|
||
|
||
async with client:
|
||
resp = await client.post(f"/projects/{pid}/outline", json={"volume": 1})
|
||
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert [c["no"] for c in body["chapters"]] == [1, 2]
|
||
ch1 = body["chapters"][0]
|
||
assert ch1["beats"] == ["开篇引入主角", "埋下信物伏笔"]
|
||
assert ch1["foreshadow_windows"][0]["code"] == "F1"
|
||
assert ch1["foreshadow_windows"][0]["expected_close_to"] == 10
|
||
# 逐章 upsert（2 章）+ 端点末尾一次 commit。
|
||
assert outline_repo.upsert_calls == 2
|
||
assert (pid, 1) in outline_repo.rows
|
||
assert outline_repo.rows[(pid, 1)][1] == ["开篇引入主角", "埋下信物伏笔"]
|
||
assert session.commits == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_outline_is_idempotent_overwrites_same_chapter() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
outline_repo = FakeOutlineWriteRepo()
|
||
client, _, _ = _make_client(project_repo=project_repo, outline_repo=outline_repo)
|
||
|
||
async with client:
|
||
await client.post(f"/projects/{pid}/outline", json={"volume": 1})
|
||
await client.post(f"/projects/{pid}/outline", json={"volume": 1})
|
||
|
||
# 同 (project_id, chapter_no) 覆盖，不堆叠 → 仍只 2 行。
|
||
assert len(outline_repo.rows) == 2
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_outline_unknown_project_404() -> None:
|
||
client, _, _ = _make_client()
|
||
async with client:
|
||
resp = await client.post(f"/projects/{uuid.uuid4()}/outline", json={})
|
||
assert resp.status_code == 404
|
||
assert resp.json()["error"]["code"] == ErrorCode.NOT_FOUND
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_outline_without_credentials_maps_to_llm_unavailable() -> None:
|
||
# 无凭据时 `build_gateway_for_tier` 抛 LLM_UNAVAILABLE（503），在网关 dep 解析阶段发生。
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
session = FakeSession()
|
||
os.environ.setdefault("CREDENTIAL_ENC_KEY", Fernet.generate_key().decode())
|
||
from ww_api.main import create_app
|
||
from ww_api.services.project_deps import (
|
||
get_foreshadow_repo,
|
||
get_memory_repos,
|
||
get_outline_gateway,
|
||
get_outline_write_repo,
|
||
get_project_repo,
|
||
)
|
||
from ww_db import get_session
|
||
|
||
async def _no_creds() -> object:
|
||
raise AppError(ErrorCode.LLM_UNAVAILABLE, "未配置凭据", {"provider": "deepseek"})
|
||
|
||
app = create_app()
|
||
app.dependency_overrides[get_project_repo] = lambda: project_repo
|
||
app.dependency_overrides[get_foreshadow_repo] = lambda: FakeForeshadowRepo()
|
||
app.dependency_overrides[get_memory_repos] = _empty_memory_repos
|
||
app.dependency_overrides[get_outline_write_repo] = lambda: FakeOutlineWriteRepo()
|
||
app.dependency_overrides[get_session] = lambda: session
|
||
app.dependency_overrides[get_outline_gateway] = _no_creds
|
||
transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||
resp = await client.post(f"/projects/{pid}/outline", json={})
|
||
|
||
assert resp.status_code == 503
|
||
assert resp.json()["error"]["code"] == ErrorCode.LLM_UNAVAILABLE
|
||
# 网关阶段失败 → 未生成、未提交。
|
||
assert session.commits == 0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_outline_upper_error_propagates() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
gateway = FakeReviewGateway(error=AppError(ErrorCode.LLM_UNAVAILABLE, "boom"))
|
||
client, _, session = _make_client(project_repo=project_repo, gateway=gateway)
|
||
|
||
async with client:
|
||
resp = await client.post(f"/projects/{pid}/outline", json={})
|
||
|
||
assert resp.status_code == 503
|
||
assert session.commits == 0
|
||
|
||
|
||
# ---- 大纲读取（GET /outline）----
|
||
|
||
|
||
def _make_read_client(
|
||
*,
|
||
project_repo: FakeProjectRepo,
|
||
outline_read_repo: FakeOutlineReadRepo,
|
||
) -> httpx.AsyncClient:
|
||
"""装配 GET /outline 测试客户端（注入 fake 项目 repo + 读侧 outline repo）。"""
|
||
os.environ.setdefault("CREDENTIAL_ENC_KEY", Fernet.generate_key().decode())
|
||
from ww_api.main import create_app
|
||
from ww_api.services.project_deps import get_outline_read_repo, get_project_repo
|
||
|
||
app = create_app()
|
||
app.dependency_overrides[get_project_repo] = lambda: project_repo
|
||
app.dependency_overrides[get_outline_read_repo] = lambda: outline_read_repo
|
||
|
||
transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)
|
||
return httpx.AsyncClient(transport=transport, base_url="http://test")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_outline_returns_persisted_chapters_in_order_with_unpacked_beats() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
read_repo = FakeOutlineReadRepo()
|
||
# 乱序入种，断言端点按 chapter_no 升序返回。
|
||
read_repo.add_chapter(pid, volume=1, chapter_no=2, beats=["冲突升级"])
|
||
read_repo.add_chapter(
|
||
pid,
|
||
volume=1,
|
||
chapter_no=1,
|
||
beats=["开篇引入主角", "埋下信物伏笔"],
|
||
foreshadow_windows=[{"code": "F1", "plant_chapter": 1, "expected_close_to": 10}],
|
||
)
|
||
client = _make_read_client(project_repo=project_repo, outline_read_repo=read_repo)
|
||
|
||
async with client:
|
||
resp = await client.get(f"/projects/{pid}/outline")
|
||
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert [c["no"] for c in body["chapters"]] == [1, 2]
|
||
ch1 = body["chapters"][0]
|
||
# beats 由 DB dict `{"beats": [...]}` 解包成裸 list（与 POST 响应同形）。
|
||
assert ch1["beats"] == ["开篇引入主角", "埋下信物伏笔"]
|
||
assert ch1["foreshadow_windows"][0]["code"] == "F1"
|
||
assert ch1["foreshadow_windows"][0]["expected_close_to"] == 10
|
||
assert body["chapters"][1]["beats"] == ["冲突升级"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_outline_with_volume_filters_to_that_volume() -> None:
|
||
# ?volume=N 只返回该卷章节；无该参数返回全部（向后兼容）。
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
read_repo = FakeOutlineReadRepo()
|
||
read_repo.add_chapter(pid, volume=1, chapter_no=1, beats=["卷一·开篇"])
|
||
read_repo.add_chapter(pid, volume=1, chapter_no=2, beats=["卷一·冲突"])
|
||
read_repo.add_chapter(pid, volume=2, chapter_no=3, beats=["卷二·新篇"])
|
||
client = _make_read_client(project_repo=project_repo, outline_read_repo=read_repo)
|
||
|
||
async with client:
|
||
all_resp = await client.get(f"/projects/{pid}/outline")
|
||
vol2_resp = await client.get(f"/projects/{pid}/outline?volume=2")
|
||
|
||
# 不带 volume → 全部三章。
|
||
assert all_resp.status_code == 200
|
||
assert [c["no"] for c in all_resp.json()["chapters"]] == [1, 2, 3]
|
||
# ?volume=2 → 仅卷二的第 3 章。
|
||
assert vol2_resp.status_code == 200
|
||
vol2_chapters = vol2_resp.json()["chapters"]
|
||
assert [c["no"] for c in vol2_chapters] == [3]
|
||
assert all(c["volume"] == 2 for c in vol2_chapters)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_outline_repo_volume_filter_pushed_into_query() -> None:
|
||
# 卷过滤下推到 repo（DB WHERE），而非端点 Python 侧过滤：直接调 repo 断言只回该卷。
|
||
repo = FakeOutlineReadRepo()
|
||
pid = uuid.uuid4()
|
||
repo.add_chapter(pid, volume=1, chapter_no=1, beats=["卷一·1"])
|
||
repo.add_chapter(pid, volume=1, chapter_no=2, beats=["卷一·2"])
|
||
repo.add_chapter(pid, volume=2, chapter_no=3, beats=["卷二·3"])
|
||
|
||
vol1 = await repo.list_for_project(pid, volume=1)
|
||
vol2 = await repo.list_for_project(pid, volume=2)
|
||
all_rows = await repo.list_for_project(pid)
|
||
|
||
assert [v.chapter_no for v in vol1] == [1, 2]
|
||
assert all(v.volume == 1 for v in vol1)
|
||
assert [v.chapter_no for v in vol2] == [3]
|
||
assert [v.chapter_no for v in all_rows] == [1, 2, 3]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_outline_returns_empty_list_when_no_outline() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
client = _make_read_client(project_repo=project_repo, outline_read_repo=FakeOutlineReadRepo())
|
||
|
||
async with client:
|
||
resp = await client.get(f"/projects/{pid}/outline")
|
||
|
||
# 项目存在但无大纲 → 200 空列表（非 404）。
|
||
assert resp.status_code == 200
|
||
assert resp.json()["chapters"] == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_outline_unknown_project_404() -> None:
|
||
client = _make_read_client(
|
||
project_repo=FakeProjectRepo(), outline_read_repo=FakeOutlineReadRepo()
|
||
)
|
||
|
||
async with client:
|
||
resp = await client.get(f"/projects/{uuid.uuid4()}/outline")
|
||
|
||
assert resp.status_code == 404
|
||
assert resp.json()["error"]["code"] == ErrorCode.NOT_FOUND
|
||
|
||
|
||
# ---- 伏笔看板 ----
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_foreshadow_board_filters_by_status() -> None:
|
||
repo = FakeForeshadowRepo()
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
await repo.register(pid, code="F1", title="开放伏笔")
|
||
await repo.register(pid, code="F2", title="部分回收")
|
||
await repo.transition(pid, "F2", to_status="PARTIAL")
|
||
from ww_api.main import create_app
|
||
from ww_api.services.project_deps import get_foreshadow_repo, get_project_repo
|
||
|
||
os.environ.setdefault("CREDENTIAL_ENC_KEY", Fernet.generate_key().decode())
|
||
app = create_app()
|
||
app.dependency_overrides[get_foreshadow_repo] = lambda: repo
|
||
# 读端点现做 project 存在性 404 校验（P2）——注入有该 project 的 FakeProjectRepo。
|
||
app.dependency_overrides[get_project_repo] = lambda: project_repo
|
||
transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||
all_resp = await client.get(f"/projects/{pid}/foreshadow")
|
||
open_resp = await client.get(f"/projects/{pid}/foreshadow", params={"status": "OPEN"})
|
||
bad_resp = await client.get(f"/projects/{pid}/foreshadow", params={"status": "NOPE"})
|
||
|
||
assert {f["code"] for f in all_resp.json()["foreshadow"]} == {"F1", "F2"}
|
||
open_codes = [f["code"] for f in open_resp.json()["foreshadow"]]
|
||
assert open_codes == ["F1"]
|
||
assert bad_resp.status_code == 422
|
||
assert bad_resp.json()["error"]["code"] == ErrorCode.VALIDATION
|
||
assert bad_resp.json()["error"]["details"]["reason"] == "invalid_status"
|