writer-work-flow/apps/api/tests/test_injection.py

"""B0 注入透明读端点测试：GET /projects/:id/chapters/:no/injection。

读取 assemble 的确定性 SelectionTrace（无 LLM、无 commit）：
- 命中实体 + 入选理由徽标；项目不存在 → 404；无大纲 → selected: []（不报错）。
"""

from __future__ import annotations

import uuid

import httpx
import pytest
from fakes_projects import FakeProjectRepo
from ww_api.services.credentials import STUB_OWNER_ID
from ww_core.domain.project_repo import ProjectCreate
from ww_core.domain.repositories import (
    CharacterView,
    DigestView,
    ForeshadowView,
    MemoryRepos,
    OutlineView,
    ProjectSpecView,
    RuleView,
    StyleView,
    WorldEntityView,
)
from ww_shared import ErrorCode

# ---- 内存 fake memory repos（无 DB；只够断言 SelectionTrace 形状）----


class _OutlineRepo:
    def __init__(self, rows: dict[int, OutlineView]) -> None:
        self._rows = rows

    async def get(self, project_id: uuid.UUID, chapter_no: int) -> OutlineView | None:
        return self._rows.get(chapter_no)

    async def list_for_project(self, project_id: uuid.UUID) -> list[OutlineView]:
        return [self._rows[k] for k in sorted(self._rows)]


class _CharRepo:
    def __init__(self, rows: list[CharacterView]) -> None:
        self._rows = rows

    async def list_for_project(self, project_id: uuid.UUID) -> list[CharacterView]:
        return list(self._rows)


class _WorldRepo:
    def __init__(self, rows: list[WorldEntityView]) -> None:
        self._rows = rows

    async def list_for_project(self, project_id: uuid.UUID) -> list[WorldEntityView]:
        return list(self._rows)


class _DigestRepo:
    async def recent(self, project_id: uuid.UUID, k: int) -> list[DigestView]:
        return []


class _ForeshadowRepo:
    async def list_for_codes(self, project_id: uuid.UUID, codes: list[str]) -> list[ForeshadowView]:
        return []


class _StyleRepo:
    async def latest(self, project_id: uuid.UUID) -> StyleView | None:
        return None


class _RulesRepo:
    async def all_for_project(self, project_id: uuid.UUID) -> list[RuleView]:
        return []


class _SpecRepo:
    async def spec(self, project_id: uuid.UUID) -> ProjectSpecView | None:
        return ProjectSpecView(title="测试作品", premise="测试前提")


def _memory_repos(
    *,
    outline: dict[int, OutlineView] | None = None,
    characters: list[CharacterView] | None = None,
    world: list[WorldEntityView] | None = None,
) -> MemoryRepos:
    return MemoryRepos(
        outline=_OutlineRepo(outline or {}),
        character=_CharRepo(characters or []),
        world_entity=_WorldRepo(world or []),
        digest=_DigestRepo(),
        foreshadow=_ForeshadowRepo(),
        style=_StyleRepo(),
        rules=_RulesRepo(),
        project=_SpecRepo(),
    )


def _make_client(
    *,
    project_repo: FakeProjectRepo,
    memory_repos: MemoryRepos,
) -> httpx.AsyncClient:
    import os

    os.environ.setdefault("CREDENTIAL_ENC_KEY", "x" * 44)
    from ww_api.main import create_app
    from ww_api.services.project_deps import get_memory_repos, get_project_repo

    app = create_app()
    app.dependency_overrides[get_project_repo] = lambda: project_repo
    app.dependency_overrides[get_memory_repos] = lambda: memory_repos
    transport = httpx.ASGITransport(app=app)
    return httpx.AsyncClient(transport=transport, base_url="http://test")


async def _create_project(repo: FakeProjectRepo, title: str = "注入") -> uuid.UUID:
    view = await repo.create(STUB_OWNER_ID, ProjectCreate(title=title))
    return uuid.UUID(str(view.id))


@pytest.mark.asyncio
async def test_injection_lists_selected_entities_with_reasons() -> None:
    repo = FakeProjectRepo()
    pid = await _create_project(repo)
    memory = _memory_repos(
        outline={3: OutlineView(volume=1, chapter_no=3, beats={"entities": ["青檀"]})},
        characters=[
            CharacterView(name="林动", role="主角"),
            CharacterView(name="青檀", role="配角"),
            CharacterView(name="路人甲", role="龙套"),
        ],
    )
    client = _make_client(project_repo=repo, memory_repos=memory)
    async with client:
        resp = await client.get(f"/projects/{pid}/chapters/3/injection")
    assert resp.status_code == 200
    body = resp.json()
    assert body["project_id"] == str(pid)
    assert body["chapter_no"] == 3
    by_name = {e["name"]: e for e in body["selected"]}
    assert "林动" in by_name and "main_character" in by_name["林动"]["reasons"]
    assert "青檀" in by_name and "explicit_beat" in by_name["青檀"]["reasons"]
    assert "路人甲" not in by_name  # 未命中任何理由 → 不注入
    assert body["recent_n"] >= 1


@pytest.mark.asyncio
async def test_injection_unknown_project_404() -> None:
    repo = FakeProjectRepo()
    client = _make_client(project_repo=repo, memory_repos=_memory_repos())
    async with client:
        resp = await client.get(f"/projects/{uuid.uuid4()}/chapters/1/injection")
    assert resp.status_code == 404
    assert resp.json()["error"]["code"] == ErrorCode.NOT_FOUND


@pytest.mark.asyncio
async def test_injection_no_outline_returns_empty() -> None:
    repo = FakeProjectRepo()
    pid = await _create_project(repo)
    client = _make_client(project_repo=repo, memory_repos=_memory_repos())
    async with client:
        resp = await client.get(f"/projects/{pid}/chapters/9/injection")
    assert resp.status_code == 200
    assert resp.json()["selected"] == []