"""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"