0d473e726e
- selection.select_relevant_entities 加 pinned/excluded frozenset 入参；新理由 author_pin（末位） - assemble 加 override 关键字参（recent_n 覆盖回看章数 + pin/排除）；GET/PUT/draft 同读同一覆盖（不变量 #6 作者兜底） - 新 domain/injection_repo.py（InjectionOverride/EntityRef/SqlInjectionOverrideRepo，upsert 只 flush） - 新表 chapter_injection（迁移 ad2c4c663daf，唯一 project_id+chapter_no；复用 outline 行已否决） - PUT/GET injection 端点 + InjectionOverrideRequest/回显 pinned/excluded；recent_n 1..20 校验 422 - 单测：selection pin/排除/exclude>pin + assemble override + 端点 PUT/404/422；regen TS 客户端 - 门禁绿：ruff/format · mypy 163 · alembic 无漂移 · pytest 476 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
261 lines
9.0 KiB
Python
261 lines
9.0 KiB
Python
"""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.injection_repo import InjectionOverride
|
||
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="测试前提")
|
||
|
||
|
||
class _FakeInjectionRepo:
|
||
"""内存注入覆盖 repo（按 (project_id, chapter_no) 存最后一次 upsert）。"""
|
||
|
||
def __init__(self) -> None:
|
||
self._store: dict[tuple[str, int], InjectionOverride] = {}
|
||
|
||
async def get(self, project_id: uuid.UUID, chapter_no: int) -> InjectionOverride | None:
|
||
return self._store.get((str(project_id), chapter_no))
|
||
|
||
async def upsert(
|
||
self, project_id: uuid.UUID, chapter_no: int, override: InjectionOverride
|
||
) -> InjectionOverride:
|
||
self._store[(str(project_id), chapter_no)] = override
|
||
return override
|
||
|
||
|
||
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,
|
||
injection_repo: _FakeInjectionRepo | None = None,
|
||
) -> httpx.AsyncClient:
|
||
import os
|
||
|
||
os.environ.setdefault("CREDENTIAL_ENC_KEY", "x" * 44)
|
||
from fakes_projects import FakeSession
|
||
from ww_api.main import create_app
|
||
from ww_api.services.project_deps import (
|
||
get_injection_repo,
|
||
get_memory_repos,
|
||
get_project_repo,
|
||
)
|
||
from ww_db import get_session
|
||
|
||
app = create_app()
|
||
app.dependency_overrides[get_project_repo] = lambda: project_repo
|
||
app.dependency_overrides[get_memory_repos] = lambda: memory_repos
|
||
app.dependency_overrides[get_injection_repo] = lambda: injection_repo or _FakeInjectionRepo()
|
||
app.dependency_overrides[get_session] = lambda: FakeSession()
|
||
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"] == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_injection_put_pin_excludes_and_reflects() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _create_project(repo)
|
||
memory = _memory_repos(
|
||
outline={3: OutlineView(volume=1, chapter_no=3, beats={})},
|
||
characters=[
|
||
CharacterView(name="林动", role="主角"),
|
||
CharacterView(name="路人乙", role="龙套"),
|
||
],
|
||
)
|
||
injection = _FakeInjectionRepo()
|
||
client = _make_client(project_repo=repo, memory_repos=memory, injection_repo=injection)
|
||
async with client:
|
||
# pin 龙套路人乙（本无理由）、排除主角林动、近况回看设为 2。
|
||
put = await client.put(
|
||
f"/projects/{pid}/chapters/3/injection",
|
||
json={
|
||
"pinned": [{"kind": "character", "name": "路人乙"}],
|
||
"excluded": [{"kind": "character", "name": "林动"}],
|
||
"recent_n": 2,
|
||
},
|
||
)
|
||
assert put.status_code == 200
|
||
body = put.json()
|
||
by_name = {e["name"]: e for e in body["selected"]}
|
||
assert "路人乙" in by_name and "author_pin" in by_name["路人乙"]["reasons"]
|
||
assert "林动" not in by_name # 排除生效
|
||
assert body["recent_n"] == 2
|
||
assert body["pinned"] == [{"kind": "character", "name": "路人乙"}]
|
||
assert body["excluded"] == [{"kind": "character", "name": "林动"}]
|
||
|
||
# 覆盖已落 fake repo → GET 回放同一结果（看到的=写章用的）。
|
||
got = await client.get(f"/projects/{pid}/chapters/3/injection")
|
||
assert got.status_code == 200
|
||
got_names = {e["name"] for e in got.json()["selected"]}
|
||
assert "路人乙" in got_names and "林动" not in got_names
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_injection_put_unknown_project_404() -> None:
|
||
repo = FakeProjectRepo()
|
||
client = _make_client(project_repo=repo, memory_repos=_memory_repos())
|
||
async with client:
|
||
resp = await client.put(
|
||
f"/projects/{uuid.uuid4()}/chapters/1/injection",
|
||
json={"pinned": [], "excluded": [], "recent_n": None},
|
||
)
|
||
assert resp.status_code == 404
|
||
assert resp.json()["error"]["code"] == ErrorCode.NOT_FOUND
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_injection_put_rejects_out_of_range_recent_n() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _create_project(repo)
|
||
client = _make_client(project_repo=repo, memory_repos=_memory_repos())
|
||
async with client:
|
||
resp = await client.put(
|
||
f"/projects/{pid}/chapters/1/injection",
|
||
json={"pinned": [], "excluded": [], "recent_n": 999},
|
||
)
|
||
assert resp.status_code == 422
|