"""GET /projects/:id/chapters 列章端点 + 纯聚合逻辑测试（审稿选章枚举）。 分两层，与本仓「真 PG 行为交 tests/ E2E、单测保持确定性」纪律一致： - **纯聚合** `build_chapter_list`：喂造好的 chapter/review 行，断言 has_draft/accepted/ reviewed_at、去重、按章号升序、空正文草稿不计（这是端点真正的业务逻辑，脱库直测）。 - **端点集成**：FastAPI client + 覆盖 `get_chapter_lister` 为 Fake（同 test_projects.py 风格）， 断言 404、空项目 []、字段透传、升序；网关无关（本端点不触 LLM）。 """ from __future__ import annotations import uuid from datetime import UTC, datetime import httpx import pytest from cryptography.fernet import Fernet from fakes_projects import FakeProjectRepo from ww_api.services.chapter_list import ( ChapterRow, ReviewRow, build_chapter_list, ) from ww_api.services.credentials import STUB_OWNER_ID from ww_core.domain.project_repo import ProjectView from ww_shared import ErrorCode # ---- 纯聚合逻辑（build_chapter_list）---- def _dt(day: int) -> datetime: return datetime(2026, 7, day, tzinfo=UTC) def test_build_chapter_list_marks_draft_accepted_and_reviewed() -> None: # 三态各出现：草稿章(1) / 已验收章(2) / 已审章(3, 有 review 时间)。 chapter_rows = [ ChapterRow(chapter_no=1, status="draft", content="草稿正文"), ChapterRow(chapter_no=2, status="draft", content="待验收"), ChapterRow(chapter_no=2, status="accepted", content="终稿"), ChapterRow(chapter_no=3, status="draft", content="被审的草稿"), ] review_rows = [ReviewRow(chapter_no=3, created_at=_dt(5))] items = build_chapter_list(chapter_rows, review_rows) by_no = {it.chapter_no: it for it in items} assert by_no[1].has_draft is True assert by_no[1].accepted is False assert by_no[1].reviewed_at is None assert by_no[2].accepted is True assert by_no[2].has_draft is True assert by_no[3].reviewed_at == _dt(5) def test_build_chapter_list_empty_returns_empty() -> None: assert build_chapter_list([], []) == [] def test_build_chapter_list_sorted_ascending_and_deduped() -> None: # 乱序 + 同章多版本行 → 去重 + 章号升序。 chapter_rows = [ ChapterRow(chapter_no=3, status="draft", content="c3"), ChapterRow(chapter_no=1, status="draft", content="c1"), ChapterRow(chapter_no=1, status="accepted", content="c1a"), ChapterRow(chapter_no=2, status="accepted", content="c2"), ] items = build_chapter_list(chapter_rows, []) assert [it.chapter_no for it in items] == [1, 2, 3] def test_build_chapter_list_reviewed_at_is_latest() -> None: # 同章多次审稿 → reviewed_at 取最近一次。 chapter_rows = [ChapterRow(chapter_no=1, status="draft", content="x")] review_rows = [ ReviewRow(chapter_no=1, created_at=_dt(3)), ReviewRow(chapter_no=1, created_at=_dt(9)), ReviewRow(chapter_no=1, created_at=_dt(6)), ] items = build_chapter_list(chapter_rows, review_rows) assert items[0].reviewed_at == _dt(9) def test_build_chapter_list_empty_content_draft_not_marked() -> None: # 空/纯空白正文的草稿行不算「有草稿正文」（与 GET .../draft 语义一致）。 chapter_rows = [ ChapterRow(chapter_no=1, status="draft", content=" "), ChapterRow(chapter_no=1, status="draft", content=None), ] items = build_chapter_list(chapter_rows, []) assert items[0].chapter_no == 1 assert items[0].has_draft is False def test_build_chapter_list_review_only_chapter_appears() -> None: # 有审稿但无 chapters 行（作者以 body.draft 送审未存草稿）→ 该章仍枚举、标已审。 items = build_chapter_list([], [ReviewRow(chapter_no=7, created_at=_dt(2))]) assert len(items) == 1 assert items[0].chapter_no == 7 assert items[0].has_draft is False assert items[0].accepted is False assert items[0].reviewed_at == _dt(2) # ---- 端点集成（FastAPI client + Fake lister）---- class _FakeChapterLister: """实现 ChapterLister Protocol 的内存版：回放预置的列章结果。""" def __init__(self, items_by_project: dict[uuid.UUID, list[object]] | None = None) -> None: self.items_by_project = items_by_project or {} async def list_chapters(self, project_id: uuid.UUID) -> list[object]: return self.items_by_project.get(project_id, []) def _make_client( *, project_repo: FakeProjectRepo, lister: _FakeChapterLister, ) -> httpx.AsyncClient: import os 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_chapter_lister, get_project_repo app = create_app() app.dependency_overrides[get_project_repo] = lambda: project_repo app.dependency_overrides[get_chapter_lister] = lambda: lister transport = httpx.ASGITransport(app=app) return httpx.AsyncClient(transport=transport, base_url="http://test") def _seed_project(project_repo: FakeProjectRepo) -> uuid.UUID: pid = uuid.uuid4() project_repo.rows[pid] = (STUB_OWNER_ID, ProjectView(id=pid, title="测试")) return pid @pytest.mark.asyncio async def test_list_chapters_returns_items_with_fields() -> None: from ww_api.schemas.projects import ChapterListItem project_repo = FakeProjectRepo() pid = _seed_project(project_repo) lister = _FakeChapterLister( { pid: [ ChapterListItem(chapter_no=1, has_draft=True, accepted=False, reviewed_at=None), ChapterListItem(chapter_no=2, has_draft=True, accepted=True, reviewed_at=_dt(4)), ] } ) client = _make_client(project_repo=project_repo, lister=lister) async with client: resp = await client.get(f"/projects/{pid}/chapters") assert resp.status_code == 200 body = resp.json() assert isinstance(body, list) assert body[0] == { "chapter_no": 1, "has_draft": True, "accepted": False, "reviewed_at": None, } assert body[1]["chapter_no"] == 2 assert body[1]["accepted"] is True assert body[1]["reviewed_at"].startswith("2026-07-04") @pytest.mark.asyncio async def test_list_chapters_empty_project_returns_empty_list() -> None: project_repo = FakeProjectRepo() pid = _seed_project(project_repo) client = _make_client(project_repo=project_repo, lister=_FakeChapterLister()) async with client: resp = await client.get(f"/projects/{pid}/chapters") assert resp.status_code == 200 assert resp.json() == [] @pytest.mark.asyncio async def test_list_chapters_unknown_project_404() -> None: client = _make_client(project_repo=FakeProjectRepo(), lister=_FakeChapterLister()) async with client: resp = await client.get(f"/projects/{uuid.uuid4()}/chapters") assert resp.status_code == 404 assert resp.json()["error"]["code"] == ErrorCode.NOT_FOUND @pytest.mark.asyncio async def test_list_chapters_ascending_order() -> None: from ww_api.schemas.projects import ChapterListItem project_repo = FakeProjectRepo() pid = _seed_project(project_repo) lister = _FakeChapterLister( { pid: [ ChapterListItem(chapter_no=1, has_draft=True, accepted=False, reviewed_at=None), ChapterListItem(chapter_no=2, has_draft=False, accepted=True, reviewed_at=None), ChapterListItem(chapter_no=5, has_draft=True, accepted=False, reviewed_at=None), ] } ) client = _make_client(project_repo=project_repo, lister=lister) async with client: resp = await client.get(f"/projects/{pid}/chapters") assert resp.status_code == 200 assert [it["chapter_no"] for it in resp.json()] == [1, 2, 5]