61398a1452
- teardown_spec.writes=["rules"]（只声明 tier，入库仍经白名单 gate，不变量 #3） - toolbox_registry teardown entry 加 ingest=IngestSpec(table="rules") - routers/toolbox.py 通用 ingest dispatcher 加 _ingest_rules： BookTeardownResult(themes/archetypes/structure/hooks) 拍平为可读 rules 条目， 落项目级（无 continuity 预检，仿 _ingest_outline 无 409） - schemas/toolbox.py 加 TeardownIngestView + ToolIngestRequest.teardown - 单测：teardown ingest→rules 真落行 + 预览(generate)不写库 后端门禁绿：ruff/format/mypy clean · alembic no drift · pytest 644 passed
713 lines
26 KiB
Python
713 lines
26 KiB
Python
"""T6.2/T6.3 创作工具箱通用端点测试（内存替身，无 DB/无网络）。
|
||
|
||
覆盖：
|
||
- GET /skills/toolbox：列出全部工具（legacy + 新）+ ingestable/legacy_route 形。
|
||
- POST .../generate：按 key 解析 spec → 结构化预览；未知 key 404；legacy key 404；无凭据 503。
|
||
- POST .../ingest：world_entities continuity 409 gate + acknowledge 放行 + partition_writes 丢越权；
|
||
不可入库工具 422；outline 细纲入库 upsert。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import uuid
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
from cryptography.fernet import Fernet
|
||
from fakes_projects import FakeProjectRepo, FakeSession
|
||
from test_projects import _empty_memory_repos
|
||
from ww_agents import Conflict, ContinuityReview
|
||
from ww_agents.schemas import (
|
||
BookTeardownResult,
|
||
ContinuationResult,
|
||
DeAiResult,
|
||
Idea,
|
||
IdeaListResult,
|
||
PolishResult,
|
||
)
|
||
from ww_core.domain.chapter_repo import ChapterDraftView, ChapterView
|
||
from ww_core.domain.outline_write_repo import OutlineWriteView
|
||
from ww_core.domain.project_repo import ProjectCreate
|
||
from ww_core.domain.repositories import OutlineView
|
||
from ww_core.domain.rule_repo import RuleWriteView
|
||
from ww_core.domain.world_entity_repo import WorldEntityWriteView
|
||
from ww_llm_gateway.types import LlmRequest, LlmResponse, ServedBy, Usage
|
||
from ww_shared import AppError, ErrorCode
|
||
|
||
STUB_OWNER = uuid.UUID(int=1)
|
||
|
||
|
||
class _SchemaRoutingGateway:
|
||
"""按 `req.output_schema` 返对应 parsed（generate / precheck 各拿自己的产物）。"""
|
||
|
||
def __init__(self, by_schema: dict[type[Any] | None, Any]) -> None:
|
||
self._by_schema = by_schema
|
||
self.calls: list[type[Any] | None] = []
|
||
|
||
async def run(self, req: LlmRequest) -> LlmResponse:
|
||
schema = req.output_schema
|
||
self.calls.append(schema)
|
||
parsed = self._by_schema.get(schema)
|
||
return LlmResponse(
|
||
text=parsed.model_dump_json() if parsed is not None else "{}",
|
||
parsed=parsed,
|
||
usage=Usage(
|
||
provider="fake",
|
||
model="fake",
|
||
input_tokens=1,
|
||
output_tokens=1,
|
||
cost_minor=0,
|
||
currency="USD",
|
||
),
|
||
served_by=ServedBy(provider="fake", model="fake"),
|
||
)
|
||
|
||
|
||
class _FakeWorldWriteRepo:
|
||
def __init__(self) -> None:
|
||
self.rows: list[dict[str, Any]] = []
|
||
|
||
async def create(
|
||
self, project_id: uuid.UUID, *, type: str, name: str, rules: list[str]
|
||
) -> WorldEntityWriteView:
|
||
self.rows.append({"type": type, "name": name, "rules": list(rules)})
|
||
return WorldEntityWriteView(id=uuid.uuid4(), type=type, name=name)
|
||
|
||
|
||
class _FakeOutlineWriteRepo:
|
||
def __init__(self) -> None:
|
||
self.rows: list[dict[str, Any]] = []
|
||
|
||
async def upsert_chapter(
|
||
self,
|
||
project_id: uuid.UUID,
|
||
*,
|
||
volume: int,
|
||
chapter_no: int,
|
||
beats: list[str],
|
||
foreshadow_windows: list[dict[str, Any]],
|
||
) -> OutlineWriteView:
|
||
self.rows.append({"chapter_no": chapter_no, "beats": list(beats)})
|
||
return OutlineWriteView(
|
||
chapter_no=chapter_no, volume=volume, beats=list(beats), foreshadow_windows=[]
|
||
)
|
||
|
||
|
||
class _FakeRuleWriteRepo:
|
||
"""拆书入库 rules 用的内存替身：按 (level, content) 收行（只 flush 语义）。"""
|
||
|
||
def __init__(self) -> None:
|
||
self.rows: list[dict[str, Any]] = []
|
||
|
||
async def create(self, project_id: uuid.UUID, *, level: str, content: str) -> RuleWriteView:
|
||
self.rows.append({"project_id": project_id, "level": level, "content": content})
|
||
return RuleWriteView(project_id=project_id, level=level, content=content)
|
||
|
||
|
||
class _FakeOutlineReadRepo:
|
||
"""续写/按章工具读节拍用的内存替身：缺章 → None（端点降级到空节拍）。"""
|
||
|
||
def __init__(self, beats_by_chapter: dict[int, list[str]] | None = None) -> None:
|
||
self._beats = beats_by_chapter or {}
|
||
|
||
async def get(self, project_id: uuid.UUID, chapter_no: int) -> Any:
|
||
beats = self._beats.get(chapter_no)
|
||
if beats is None:
|
||
return None
|
||
return OutlineView(
|
||
chapter_no=chapter_no, volume=1, beats={"beats": beats}, foreshadow_windows=[]
|
||
)
|
||
|
||
async def list_for_project(self, project_id: uuid.UUID) -> list[Any]:
|
||
return []
|
||
|
||
|
||
class _FakeChapterRepo:
|
||
"""续写读前文用的内存章节替身：按 chapter_no 存 accepted/draft 正文。"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
accepted: dict[int, str] | None = None,
|
||
drafts: dict[int, str] | None = None,
|
||
) -> None:
|
||
self._accepted = accepted or {}
|
||
self._drafts = drafts or {}
|
||
|
||
async def latest_accepted(self, project_id: uuid.UUID, chapter_no: int) -> ChapterView | None:
|
||
content = self._accepted.get(chapter_no)
|
||
if content is None:
|
||
return None
|
||
return ChapterView(
|
||
project_id=project_id,
|
||
chapter_no=chapter_no,
|
||
volume=1,
|
||
content=content,
|
||
status="accepted",
|
||
version=1,
|
||
)
|
||
|
||
async def get_draft(self, project_id: uuid.UUID, chapter_no: int) -> ChapterDraftView | None:
|
||
content = self._drafts.get(chapter_no)
|
||
if content is None:
|
||
return None
|
||
return ChapterDraftView(
|
||
project_id=project_id,
|
||
chapter_no=chapter_no,
|
||
volume=1,
|
||
content=content,
|
||
status="draft",
|
||
version=1,
|
||
)
|
||
|
||
|
||
class _CaptureGateway:
|
||
"""记录最近一次 req.input（验证续写读前文/原文路由进了 context），返指定 parsed。"""
|
||
|
||
def __init__(self, parsed: Any) -> None:
|
||
self._parsed = parsed
|
||
self.last_input: str | None = None
|
||
|
||
async def run(self, req: LlmRequest) -> LlmResponse:
|
||
# 工具箱端点恒传序列化文本 context（str）；存为 str 供断言。
|
||
self.last_input = req.input if isinstance(req.input, str) else str(req.input)
|
||
return LlmResponse(
|
||
text=self._parsed.model_dump_json(),
|
||
parsed=self._parsed,
|
||
usage=Usage(
|
||
provider="fake",
|
||
model="fake",
|
||
input_tokens=1,
|
||
output_tokens=1,
|
||
cost_minor=0,
|
||
currency="USD",
|
||
),
|
||
served_by=ServedBy(provider="fake", model="fake"),
|
||
)
|
||
|
||
|
||
def _ideas() -> IdeaListResult:
|
||
return IdeaListResult(ideas=[Idea(premise="废柴觉醒系统", hook="开局即巅峰", genre_fit="玄幻")])
|
||
|
||
|
||
def _no_conflicts() -> ContinuityReview:
|
||
return ContinuityReview(conflicts=[])
|
||
|
||
|
||
def _with_conflicts() -> ContinuityReview:
|
||
return ContinuityReview(
|
||
conflicts=[Conflict(type="设定违例", where="吞噬系统", refs=["灵脉"], suggestion="改设定")]
|
||
)
|
||
|
||
|
||
def _make_app(
|
||
*,
|
||
project_repo: FakeProjectRepo,
|
||
gateway: Any,
|
||
session: FakeSession | None = None,
|
||
world_repo: _FakeWorldWriteRepo | None = None,
|
||
outline_write_repo: _FakeOutlineWriteRepo | None = None,
|
||
chapter_repo: _FakeChapterRepo | None = None,
|
||
outline_read_repo: _FakeOutlineReadRepo | None = None,
|
||
rule_write_repo: _FakeRuleWriteRepo | None = None,
|
||
memory: Any = None,
|
||
no_creds: bool = False,
|
||
) -> tuple[Any, FakeSession, _FakeWorldWriteRepo, _FakeOutlineWriteRepo]:
|
||
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_repo,
|
||
get_memory_repos,
|
||
get_outline_read_repo,
|
||
get_outline_write_repo,
|
||
get_project_repo,
|
||
get_rule_write_repo,
|
||
get_tier_gateway_builder,
|
||
get_world_entity_write_repo,
|
||
)
|
||
from ww_db import get_session
|
||
|
||
session = session or FakeSession()
|
||
world_repo = world_repo or _FakeWorldWriteRepo()
|
||
outline_write_repo = outline_write_repo or _FakeOutlineWriteRepo()
|
||
chapter_repo = chapter_repo or _FakeChapterRepo()
|
||
outline_read_repo = outline_read_repo or _FakeOutlineReadRepo()
|
||
rule_write_repo = rule_write_repo or _FakeRuleWriteRepo()
|
||
|
||
async def _build_ok(_tier: str) -> Any:
|
||
return gateway
|
||
|
||
async def _build_no_creds(_tier: str) -> Any:
|
||
raise AppError(ErrorCode.LLM_UNAVAILABLE, "未配置凭据", {"provider": "deepseek"})
|
||
|
||
app = create_app()
|
||
app.dependency_overrides[get_project_repo] = lambda: project_repo
|
||
app.dependency_overrides[get_memory_repos] = (
|
||
(lambda: memory) if memory is not None else _empty_memory_repos
|
||
)
|
||
app.dependency_overrides[get_world_entity_write_repo] = lambda: world_repo
|
||
app.dependency_overrides[get_outline_write_repo] = lambda: outline_write_repo
|
||
app.dependency_overrides[get_chapter_repo] = lambda: chapter_repo
|
||
app.dependency_overrides[get_outline_read_repo] = lambda: outline_read_repo
|
||
app.dependency_overrides[get_rule_write_repo] = lambda: rule_write_repo
|
||
app.dependency_overrides[get_session] = lambda: session
|
||
app.dependency_overrides[get_tier_gateway_builder] = lambda: (
|
||
_build_no_creds if no_creds else _build_ok
|
||
)
|
||
return app, session, world_repo, outline_write_repo
|
||
|
||
|
||
async def _seed_project(repo: FakeProjectRepo) -> uuid.UUID:
|
||
view = await repo.create(STUB_OWNER, ProjectCreate(title="测试作品", genre="玄幻"))
|
||
return uuid.UUID(str(view.id))
|
||
|
||
|
||
def _client(app: Any) -> httpx.AsyncClient:
|
||
transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)
|
||
return httpx.AsyncClient(transport=transport, base_url="http://test")
|
||
|
||
|
||
# ---- GET /skills/toolbox ----
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_toolbox_contains_all_keys() -> None:
|
||
repo = FakeProjectRepo()
|
||
app, *_ = _make_app(project_repo=repo, gateway=_SchemaRoutingGateway({}))
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.get("/skills/toolbox")
|
||
|
||
assert resp.status_code == 200
|
||
tools = resp.json()["tools"]
|
||
keys = {t["key"] for t in tools}
|
||
assert keys >= {
|
||
"worldbuilding",
|
||
"character",
|
||
"outline",
|
||
"brainstorm",
|
||
"book-title",
|
||
"blurb",
|
||
"name",
|
||
"golden-finger",
|
||
"glossary",
|
||
"opening",
|
||
"fine-outline",
|
||
"continue",
|
||
"expand",
|
||
"de-ai",
|
||
"teardown",
|
||
}
|
||
by_key = {t["key"]: t for t in tools}
|
||
# 续写/扩写/降AI：非 legacy、纯预览（不可入库）。
|
||
for key in ("continue", "expand", "de-ai"):
|
||
assert by_key[key]["is_legacy"] is False
|
||
assert by_key[key]["ingestable"] is False
|
||
# F1：拆书非 legacy、可入库（落 rules）。
|
||
assert by_key["teardown"]["is_legacy"] is False
|
||
assert by_key["teardown"]["ingestable"] is True
|
||
# legacy 携 legacy_route + is_legacy；新工具 ingestable 标记。
|
||
assert by_key["worldbuilding"]["is_legacy"] is True
|
||
assert by_key["worldbuilding"]["legacy_route"]
|
||
assert by_key["brainstorm"]["is_legacy"] is False
|
||
assert by_key["golden-finger"]["ingestable"] is True
|
||
assert by_key["brainstorm"]["ingestable"] is False
|
||
# input_fields 形（brief textarea）。
|
||
assert any(f["name"] == "brief" for f in by_key["brainstorm"]["input_fields"])
|
||
|
||
|
||
# ---- POST .../generate ----
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_resolves_spec_and_returns_preview() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
gateway = _SchemaRoutingGateway({IdeaListResult: _ideas()})
|
||
app, session, *_ = _make_app(project_repo=repo, gateway=gateway)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/skills/brainstorm/generate", json={"brief": "废柴流"}
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["tool_key"] == "brainstorm"
|
||
assert body["output_kind"] == "IdeaListResult"
|
||
assert body["preview"]["ideas"][0]["premise"] == "废柴觉醒系统"
|
||
assert session.commits == 1 # 预览不写业务表，但落 ledger
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_unknown_tool_404() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
app, *_ = _make_app(project_repo=repo, gateway=_SchemaRoutingGateway({}))
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{pid}/skills/nope/generate", json={"brief": "x"})
|
||
|
||
assert resp.status_code == 404
|
||
assert resp.json()["error"]["code"] == ErrorCode.NOT_FOUND
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_legacy_tool_404() -> None:
|
||
# legacy 工具无通用执行路径（前端走 legacy_route）。
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
app, *_ = _make_app(project_repo=repo, gateway=_SchemaRoutingGateway({}))
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/skills/worldbuilding/generate", json={"brief": "x"}
|
||
)
|
||
|
||
assert resp.status_code == 404
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_unknown_project_404() -> None:
|
||
repo = FakeProjectRepo()
|
||
app, *_ = _make_app(
|
||
project_repo=repo, gateway=_SchemaRoutingGateway({IdeaListResult: _ideas()})
|
||
)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{uuid.uuid4()}/skills/brainstorm/generate", json={"brief": "x"}
|
||
)
|
||
|
||
assert resp.status_code == 404
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_no_credentials_503() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
app, session, *_ = _make_app(project_repo=repo, gateway=object(), no_creds=True)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{pid}/skills/brainstorm/generate", json={"brief": "x"})
|
||
|
||
assert resp.status_code == 503
|
||
assert resp.json()["error"]["code"] == ErrorCode.LLM_UNAVAILABLE
|
||
assert session.commits == 0
|
||
|
||
|
||
# ---- POST .../generate（Scope B 竞品快赢：续写/扩写/降AI/拆书）----
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_continue_reads_prior_chapter() -> None:
|
||
# 续写：端点读该章最新 accepted 正文 → 注入 context（with_prior_chapter 路由）。
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
chapter_repo = _FakeChapterRepo(accepted={2: "他握紧了那柄剑，缓缓转身。"})
|
||
gateway = _CaptureGateway(ContinuationResult(text="剑光一闪，敌人应声倒地。"))
|
||
app, session, *_ = _make_app(project_repo=repo, gateway=gateway, chapter_repo=chapter_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/skills/continue/generate", json={"chapter_no": 2, "brief": "续写战斗"}
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["tool_key"] == "continue"
|
||
assert body["output_kind"] == "ContinuationResult"
|
||
assert body["preview"]["text"] == "剑光一闪，敌人应声倒地。"
|
||
# 前文正文确实进了喂网关的 context（读章路由生效）。
|
||
assert gateway.last_input is not None
|
||
assert "他握紧了那柄剑" in gateway.last_input
|
||
assert session.commits == 1 # 预览不写业务表，落 ledger
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_continue_falls_back_to_draft() -> None:
|
||
# 无 accepted → 退回草稿正文。
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
chapter_repo = _FakeChapterRepo(drafts={1: "草稿里的开头一段。"})
|
||
gateway = _CaptureGateway(ContinuationResult(text="续写。"))
|
||
app, *_ = _make_app(project_repo=repo, gateway=gateway, chapter_repo=chapter_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/skills/continue/generate", json={"chapter_no": 1}
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
assert gateway.last_input is not None
|
||
assert "草稿里的开头一段" in gateway.last_input
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_expand_routes_text_input() -> None:
|
||
# 扩写：原文经 text 字段 → 注入 context（text_input 路由）。
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
gateway = _CaptureGateway(PolishResult(text="扩写后的丰富正文。"))
|
||
app, session, *_ = _make_app(project_repo=repo, gateway=gateway)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/skills/expand/generate",
|
||
json={"text": "原始简短的一句话。", "brief": "加强描写"},
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["output_kind"] == "PolishResult"
|
||
assert body["preview"]["text"] == "扩写后的丰富正文。"
|
||
assert gateway.last_input is not None
|
||
assert "原始简短的一句话" in gateway.last_input
|
||
assert session.commits == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_de_ai_routes_text_input() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
gateway = _CaptureGateway(DeAiResult(text="更自然的人写质感正文。"))
|
||
app, *_ = _make_app(project_repo=repo, gateway=gateway)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/skills/de-ai/generate",
|
||
json={"text": "充斥着 AI 腔的排比句。"},
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
assert resp.json()["output_kind"] == "DeAiResult"
|
||
assert gateway.last_input is not None
|
||
assert "充斥着 AI 腔的排比句" in gateway.last_input
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_teardown_returns_structured() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
parsed = BookTeardownResult(
|
||
themes=["逆袭"],
|
||
archetypes=["废柴主角"],
|
||
structure="黄金三章立钩",
|
||
hooks=["开局即巅峰"],
|
||
)
|
||
gateway = _CaptureGateway(parsed)
|
||
app, *_ = _make_app(project_repo=repo, gateway=gateway)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/skills/teardown/generate",
|
||
json={"kind": "某爆款", "text": "样本章节正文……"},
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["output_kind"] == "BookTeardownResult"
|
||
assert body["preview"]["themes"] == ["逆袭"]
|
||
assert body["preview"]["structure"] == "黄金三章立钩"
|
||
# 书名（body.kind）前置入原文 → 进了喂网关的 context（拆书 prompt 期待书名）。
|
||
assert gateway.last_input is not None
|
||
assert "书名：某爆款" in gateway.last_input
|
||
assert "样本章节正文" in gateway.last_input
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_text_input_empty_text_422() -> None:
|
||
# text_input 工具（扩写/降AI/拆书）原文为空/空白 → 422，不触达网关（不让 LLM 空跑）。
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
gateway = _CaptureGateway(PolishResult(text="不应被调用。"))
|
||
app, session, *_ = _make_app(project_repo=repo, gateway=gateway)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/skills/expand/generate",
|
||
json={"text": " ", "brief": "加强描写"},
|
||
)
|
||
|
||
assert resp.status_code == 422
|
||
assert resp.json()["error"]["code"] == ErrorCode.VALIDATION
|
||
# 校验在生成前拦下：网关未被调用、未 commit。
|
||
assert gateway.last_input is None
|
||
assert session.commits == 0
|
||
|
||
|
||
# ---- POST .../ingest（world_entities gate）----
|
||
|
||
|
||
def _gf_payload() -> dict[str, Any]:
|
||
return {
|
||
"world_entities": [
|
||
{"type": "力量体系", "name": "吞噬系统", "rules": ["每日上限", "不可越级"]}
|
||
]
|
||
}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ingest_world_entities_no_conflict_writes_201() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
gateway = _SchemaRoutingGateway({ContinuityReview: _no_conflicts()})
|
||
app, session, world_repo, _ = _make_app(project_repo=repo, gateway=gateway)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{pid}/skills/golden-finger/ingest", json=_gf_payload())
|
||
|
||
assert resp.status_code == 201
|
||
body = resp.json()
|
||
assert body["table"] == "world_entities"
|
||
assert body["created"] == ["吞噬系统"]
|
||
assert body["rejected_tables"] == []
|
||
assert len(world_repo.rows) == 1
|
||
assert world_repo.rows[0]["rules"] == ["每日上限", "不可越级"]
|
||
assert session.commits == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ingest_world_entities_conflict_blocks_409() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
gateway = _SchemaRoutingGateway({ContinuityReview: _with_conflicts()})
|
||
app, session, world_repo, _ = _make_app(project_repo=repo, gateway=gateway)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{pid}/skills/golden-finger/ingest", json=_gf_payload())
|
||
|
||
assert resp.status_code == 409
|
||
err = resp.json()["error"]
|
||
assert err["code"] == ErrorCode.CONFLICT_UNRESOLVED
|
||
assert err["details"]["conflict_count"] == 1
|
||
assert len(world_repo.rows) == 0
|
||
assert session.commits == 1 # 落 precheck ledger，不写业务表
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ingest_world_entities_acknowledged_writes() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
gateway = _SchemaRoutingGateway({ContinuityReview: _with_conflicts()})
|
||
app, _, world_repo, _ = _make_app(project_repo=repo, gateway=gateway)
|
||
|
||
payload = {**_gf_payload(), "acknowledge_conflicts": True}
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{pid}/skills/golden-finger/ingest", json=payload)
|
||
|
||
assert resp.status_code == 201
|
||
assert len(world_repo.rows) == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ingest_non_ingestable_tool_422() -> None:
|
||
# brainstorm 纯预览 → 不可入库。
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
app, *_ = _make_app(project_repo=repo, gateway=_SchemaRoutingGateway({}))
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{pid}/skills/brainstorm/ingest", json={})
|
||
|
||
assert resp.status_code == 422
|
||
assert resp.json()["error"]["code"] == ErrorCode.VALIDATION
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ingest_outline_upserts_scenes() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
gateway = _SchemaRoutingGateway({})
|
||
app, session, _, outline_repo = _make_app(project_repo=repo, gateway=gateway)
|
||
|
||
payload = {
|
||
"chapter_no": 3,
|
||
"scenes": [
|
||
{"idx": 1, "beat": "初遇反派", "purpose": "推进", "conflict": "对峙", "hook": "悬念"},
|
||
{"idx": 0, "beat": "主角觉醒", "purpose": "塑造", "conflict": "内心", "hook": "钩子"},
|
||
],
|
||
}
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{pid}/skills/fine-outline/ingest", json=payload)
|
||
|
||
assert resp.status_code == 201
|
||
body = resp.json()
|
||
assert body["table"] == "outline"
|
||
assert len(outline_repo.rows) == 1
|
||
# 场景按 idx 升序拼装成 beats（确定性）。
|
||
assert outline_repo.rows[0]["beats"] == ["主角觉醒", "初遇反派"]
|
||
assert outline_repo.rows[0]["chapter_no"] == 3
|
||
assert session.commits == 1
|
||
|
||
|
||
# ---- POST .../ingest（F1：拆书落库成 rules）----
|
||
|
||
|
||
def _teardown_payload() -> dict[str, Any]:
|
||
return {
|
||
"teardown": {
|
||
"themes": ["逆袭", "复仇"],
|
||
"archetypes": ["废柴主角", "腹黑反派"],
|
||
"structure": "黄金三章立钩→铺垫→爆发",
|
||
"hooks": ["开局即巅峰", "扮猪吃虎"],
|
||
}
|
||
}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ingest_teardown_writes_rules_201() -> None:
|
||
# F1：拆书结论拍平为可读 rules 条目落库（无 continuity 预检，仿 outline 无 409）。
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
rule_repo = _FakeRuleWriteRepo()
|
||
gateway = _SchemaRoutingGateway({})
|
||
app, session, *_ = _make_app(project_repo=repo, gateway=gateway, rule_write_repo=rule_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/skills/teardown/ingest", json=_teardown_payload()
|
||
)
|
||
|
||
assert resp.status_code == 201
|
||
body = resp.json()
|
||
assert body["table"] == "rules"
|
||
assert body["rejected_tables"] == []
|
||
# 真落行：themes/archetypes/structure/hooks 各拼成可读规则条目。
|
||
assert len(rule_repo.rows) >= 1
|
||
contents = "\n".join(r["content"] for r in rule_repo.rows)
|
||
assert "逆袭" in contents
|
||
assert "废柴主角" in contents
|
||
assert "黄金三章立钩" in contents
|
||
assert "开局即巅峰" in contents
|
||
# 全部落到 project 级 rules（owner stub 单用户）。
|
||
for r in rule_repo.rows:
|
||
assert r["level"] == "project"
|
||
assert r["project_id"] == pid
|
||
assert session.commits == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_teardown_generate_preview_does_not_write_rules() -> None:
|
||
# 负向：预览（generate）只产结构化产物，绝不写 rules（不变量 #3：入库经 ingest gate）。
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
rule_repo = _FakeRuleWriteRepo()
|
||
parsed = BookTeardownResult(
|
||
themes=["逆袭"], archetypes=["废柴主角"], structure="立钩", hooks=["开局巅峰"]
|
||
)
|
||
gateway = _CaptureGateway(parsed)
|
||
app, _, *_ = _make_app(project_repo=repo, gateway=gateway, rule_write_repo=rule_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/skills/teardown/generate",
|
||
json={"kind": "某爆款", "text": "样本章节正文……"},
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
assert resp.json()["output_kind"] == "BookTeardownResult"
|
||
assert rule_repo.rows == [] # 预览不写库
|