765dbdfbd4
M4（文风）: style-auditor 双轨（提取指纹/漂移第四审）+ jobs 长任务框架(zombie reaper) + 回炉 refine + GET /style read-back。 M5（生成+扩展）: worldbuilder/character-gen（入库 continuity 409 gate + partition_writes 白名单 + schema→JSONB 形变）； 网关多 provider 回退链/熔断/能力降级（Anthropic/Gemini 适配器）；Skill registry + 表权限沙箱 + 规则； 前端 角色生成器/世界观/Codex/规则页/技能库/⌘K 命令面板。 K1（Kimi Code 订阅接入）: OAuth device-flow（kimi-code）+ 静态 Console key（kimi-code-key）两路径； coding 端点 KimiCLI 伪造头（实测 UA allow-list 门禁，缺则 403）+ JSON 模式结构化（thinking ⊥ tool_choice）。 本地联调修复: CORS 中间件；assemble 注入 premise+「写第N章」指令（修空 prompt 400）； GET /outline·/draft read-back + 大纲/工作台/审稿页重载；写页 client/server 常量边界 + notFound 健壮化； 字数 toLocaleString locale 水合；审稿页终稿从已存草稿 seed（修 accept 422）。 门禁: backend ruff/mypy(157)/alembic 无漂移/pytest 451 · frontend lint/tsc/vitest/build。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
507 lines
17 KiB
Python
507 lines
17 KiB
Python
"""T5.2 生成/入库 + 读端点测试（内存替身，无 DB/无网络）。
|
||
|
||
覆盖：
|
||
- POST /world/generate：mock 网关产 WorldGenResult → 预览 + commit（落 ledger）；项目不存在 404。
|
||
- POST /characters/generate：mock 网关产 CharacterGenResult → 预览（含已有角色注入防雷同）。
|
||
- POST /characters（入库）：precheck 无冲突 → 写库 + 201；有冲突且未确认 → 409；确认后放行写库。
|
||
- 无凭据 → 503（override 网关 dep 抛 LLM_UNAVAILABLE）。
|
||
- GET /rules：规则列表；GET /skills：技能库列表。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import uuid
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
from fakes_projects import FakeProjectRepo, FakeSession
|
||
from test_projects import _empty_memory_repos
|
||
from ww_agents import (
|
||
AgentSpec,
|
||
CharacterCard,
|
||
CharacterGenResult,
|
||
Conflict,
|
||
ContinuityReview,
|
||
WorldEntityCard,
|
||
WorldGenResult,
|
||
)
|
||
from ww_core.domain.character_repo import CharacterWriteView
|
||
from ww_core.domain.project_repo import ProjectCreate
|
||
from ww_core.domain.repositories import RuleView
|
||
from ww_llm_gateway.types import LlmRequest, LlmResponse, ServedBy, Usage
|
||
from ww_shared import AppError, ErrorCode
|
||
from ww_skills import SkillRegistry
|
||
|
||
STUB_OWNER = uuid.UUID(int=1)
|
||
|
||
|
||
class _SchemaRoutingGateway:
|
||
"""按 `req.output_schema` 返对应 parsed（world/character/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 _FakeCharacterWriteRepo:
|
||
def __init__(self) -> None:
|
||
self.rows: list[dict[str, Any]] = []
|
||
|
||
async def create(
|
||
self,
|
||
project_id: uuid.UUID,
|
||
*,
|
||
name: str,
|
||
role: str,
|
||
traits: list[str],
|
||
backstory: str,
|
||
arc: str,
|
||
speech_tics: list[str],
|
||
tags: list[Any],
|
||
relations: list[dict[str, Any]],
|
||
) -> CharacterWriteView:
|
||
self.rows.append(
|
||
{
|
||
"project_id": project_id,
|
||
"name": name,
|
||
"role": role,
|
||
"traits": list(traits),
|
||
"arc": arc,
|
||
"speech_tics": list(speech_tics),
|
||
"tags": list(tags),
|
||
"relations": [dict(r) for r in relations],
|
||
}
|
||
)
|
||
return CharacterWriteView(id=uuid.uuid4(), name=name, role=role)
|
||
|
||
|
||
class _FakeRulesReadRepo:
|
||
def __init__(self, rules: list[RuleView] | None = None) -> None:
|
||
self._rules = rules or []
|
||
|
||
async def all_for_project(self, project_id: uuid.UUID) -> list[RuleView]:
|
||
return list(self._rules)
|
||
|
||
|
||
def _world_result() -> WorldGenResult:
|
||
return WorldGenResult(
|
||
entities=[WorldEntityCard(type="力量体系", name="灵脉", rules=["灵力守恒", "禁止跨界"])]
|
||
)
|
||
|
||
|
||
def _character_result() -> CharacterGenResult:
|
||
return CharacterGenResult(
|
||
cards=[
|
||
CharacterCard(
|
||
name="叶寒",
|
||
role="主角",
|
||
traits=["腹黑", "护短"],
|
||
backstory="孤儿出身",
|
||
arc="从孤儿到帝王",
|
||
speech_tics=["哼"],
|
||
tags=["天才"],
|
||
relations=[],
|
||
)
|
||
]
|
||
)
|
||
|
||
|
||
def _no_conflicts() -> ContinuityReview:
|
||
return ContinuityReview(conflicts=[])
|
||
|
||
|
||
def _with_conflicts() -> ContinuityReview:
|
||
return ContinuityReview(
|
||
conflicts=[Conflict(type="设定违例", where="叶寒卡", refs=["灵脉"], suggestion="改设定")]
|
||
)
|
||
|
||
|
||
def _skill_registry() -> SkillRegistry:
|
||
spec = AgentSpec(
|
||
name="custom-namer",
|
||
tier="light",
|
||
system_prompt="取名",
|
||
input_schema=None,
|
||
output_schema=None,
|
||
reads=["characters"],
|
||
writes=["characters"],
|
||
scope="custom",
|
||
)
|
||
return SkillRegistry({spec.name: spec})
|
||
|
||
|
||
def _make_app(
|
||
*,
|
||
project_repo: FakeProjectRepo,
|
||
gateway: Any,
|
||
char_repo: _FakeCharacterWriteRepo | None = None,
|
||
session: FakeSession | None = None,
|
||
rules_repo: _FakeRulesReadRepo | None = None,
|
||
registry: SkillRegistry | None = None,
|
||
memory: Any = None,
|
||
no_creds: bool = False,
|
||
) -> tuple[Any, FakeSession]:
|
||
os.environ.setdefault("CREDENTIAL_ENC_KEY", "x" * 44)
|
||
from ww_api.main import create_app
|
||
from ww_api.services.project_deps import (
|
||
get_character_gen_gateway,
|
||
get_character_write_repo,
|
||
get_memory_repos,
|
||
get_precheck_gateway,
|
||
get_project_repo,
|
||
get_rules_read_repo,
|
||
get_skill_registry,
|
||
get_worldbuilder_gateway,
|
||
)
|
||
from ww_db import get_session
|
||
|
||
session = session or FakeSession()
|
||
char_repo = char_repo or _FakeCharacterWriteRepo()
|
||
rules_repo = rules_repo or _FakeRulesReadRepo()
|
||
registry = registry or _skill_registry()
|
||
|
||
async def _raise_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_memory_repos] = (
|
||
(lambda: memory) if memory is not None else _empty_memory_repos
|
||
)
|
||
app.dependency_overrides[get_character_write_repo] = lambda: char_repo
|
||
app.dependency_overrides[get_rules_read_repo] = lambda: rules_repo
|
||
app.dependency_overrides[get_skill_registry] = lambda: registry
|
||
app.dependency_overrides[get_session] = lambda: session
|
||
gw = _raise_no_creds if no_creds else (lambda: gateway)
|
||
app.dependency_overrides[get_worldbuilder_gateway] = gw
|
||
app.dependency_overrides[get_character_gen_gateway] = gw
|
||
app.dependency_overrides[get_precheck_gateway] = gw
|
||
return app, session
|
||
|
||
|
||
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")
|
||
|
||
|
||
# ---- 世界观生成 ----
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_world_returns_preview_and_commits_ledger() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
gateway = _SchemaRoutingGateway({WorldGenResult: _world_result()})
|
||
app, session = _make_app(project_repo=repo, gateway=gateway)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{pid}/world/generate", json={"brief": "东方玄幻"})
|
||
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["entities"][0]["name"] == "灵脉"
|
||
assert body["entities"][0]["rules"] == ["灵力守恒", "禁止跨界"]
|
||
assert session.commits == 1 # 预览不写业务表，但落 ledger
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_world_unknown_project_404() -> None:
|
||
repo = FakeProjectRepo()
|
||
gateway = _SchemaRoutingGateway({WorldGenResult: _world_result()})
|
||
app, _ = _make_app(project_repo=repo, gateway=gateway)
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{uuid.uuid4()}/world/generate", json={"brief": "x"})
|
||
assert resp.status_code == 404
|
||
assert resp.json()["error"]["code"] == ErrorCode.NOT_FOUND
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_world_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}/world/generate", json={"brief": "x"})
|
||
assert resp.status_code == 503
|
||
assert resp.json()["error"]["code"] == ErrorCode.LLM_UNAVAILABLE
|
||
assert session.commits == 0
|
||
|
||
|
||
# ---- 角色生成 ----
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_characters_returns_preview() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
gateway = _SchemaRoutingGateway({CharacterGenResult: _character_result()})
|
||
app, session = _make_app(project_repo=repo, gateway=gateway)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/characters/generate", json={"brief": "天才主角", "count": 1}
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
card = resp.json()["cards"][0]
|
||
assert card["name"] == "叶寒"
|
||
assert card["traits"] == ["腹黑", "护短"]
|
||
assert card["arc"] == "从孤儿到帝王"
|
||
assert session.commits == 1
|
||
|
||
|
||
# ---- 角色入库（gate）----
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ingest_characters_no_conflict_writes_and_201() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
gateway = _SchemaRoutingGateway({ContinuityReview: _no_conflicts()})
|
||
char_repo = _FakeCharacterWriteRepo()
|
||
app, session = _make_app(project_repo=repo, gateway=gateway, char_repo=char_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/characters",
|
||
json={
|
||
"cards": [
|
||
{
|
||
"name": "叶寒",
|
||
"role": "主角",
|
||
"traits": ["腹黑"],
|
||
"backstory": "孤儿",
|
||
"arc": "成长",
|
||
"speech_tics": ["哼"],
|
||
"tags": ["天才"],
|
||
"relations": [],
|
||
}
|
||
]
|
||
},
|
||
)
|
||
|
||
assert resp.status_code == 201
|
||
body = resp.json()
|
||
assert body["created"] == ["叶寒"]
|
||
assert body["rejected_tables"] == []
|
||
assert len(char_repo.rows) == 1
|
||
assert char_repo.rows[0]["traits"] == ["腹黑"]
|
||
assert session.commits == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ingest_characters_with_conflict_blocks_409() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
gateway = _SchemaRoutingGateway({ContinuityReview: _with_conflicts()})
|
||
char_repo = _FakeCharacterWriteRepo()
|
||
app, session = _make_app(project_repo=repo, gateway=gateway, char_repo=char_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/characters",
|
||
json={"cards": [{"name": "叶寒", "role": "主角", "backstory": "孤儿", "arc": "成长"}]},
|
||
)
|
||
|
||
assert resp.status_code == 409
|
||
err = resp.json()["error"]
|
||
assert err["code"] == ErrorCode.CONFLICT_UNRESOLVED
|
||
assert err["details"]["conflict_count"] == 1
|
||
assert err["details"]["conflicts"][0]["type"] == "设定违例"
|
||
# 未写库（gate 拦下），但落 precheck ledger。
|
||
assert len(char_repo.rows) == 0
|
||
assert session.commits == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ingest_characters_acknowledged_conflict_writes() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
gateway = _SchemaRoutingGateway({ContinuityReview: _with_conflicts()})
|
||
char_repo = _FakeCharacterWriteRepo()
|
||
app, _ = _make_app(project_repo=repo, gateway=gateway, char_repo=char_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/characters",
|
||
json={
|
||
"cards": [{"name": "叶寒", "role": "主角", "backstory": "孤儿", "arc": "成长"}],
|
||
"acknowledge_conflicts": True,
|
||
},
|
||
)
|
||
|
||
assert resp.status_code == 201
|
||
assert len(char_repo.rows) == 1
|
||
|
||
|
||
# ---- 读端点 ----
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_rules_returns_rules() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
rules_repo = _FakeRulesReadRepo(
|
||
[
|
||
RuleView(level="project", content="主角不复活"),
|
||
RuleView(level="global", content="无脏话"),
|
||
]
|
||
)
|
||
gateway = _SchemaRoutingGateway({})
|
||
app, _ = _make_app(project_repo=repo, gateway=gateway, rules_repo=rules_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.get(f"/projects/{pid}/rules")
|
||
|
||
assert resp.status_code == 200
|
||
rules = resp.json()["rules"]
|
||
assert [r["content"] for r in rules] == ["主角不复活", "无脏话"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_skills_returns_registry() -> None:
|
||
repo = FakeProjectRepo()
|
||
gateway = _SchemaRoutingGateway({})
|
||
app, _ = _make_app(project_repo=repo, gateway=gateway)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.get("/skills")
|
||
|
||
assert resp.status_code == 200
|
||
skills = resp.json()["skills"]
|
||
assert skills[0]["name"] == "custom-namer"
|
||
assert skills[0]["scope"] == "custom"
|
||
assert skills[0]["writes"] == ["characters"]
|
||
|
||
|
||
# ---- 设定库 Codex 读端点（GET characters / world_entities）----
|
||
|
||
|
||
def _codex_memory() -> Any:
|
||
"""MemoryRepos：角色/世界观各一行（DB JSONB dict 形，验反向解包到 API list/str）。"""
|
||
from test_projects import (
|
||
_EmptyDigestRepo,
|
||
_EmptyForeshadowRepo,
|
||
_EmptyOutlineRepo,
|
||
_EmptyRulesRepo,
|
||
_EmptyStyleRepo,
|
||
_StubProjectSpecRepo,
|
||
)
|
||
from ww_core.domain.repositories import (
|
||
CharacterView,
|
||
MemoryRepos,
|
||
WorldEntityView,
|
||
)
|
||
|
||
class _CharRepo:
|
||
async def list_for_project(self, project_id: uuid.UUID) -> list[CharacterView]:
|
||
return [
|
||
CharacterView(
|
||
name="叶寒",
|
||
role="主角",
|
||
traits={"items": ["腹黑", "护短"]},
|
||
backstory="孤儿出身",
|
||
arc={"text": "从孤儿到帝王"},
|
||
speech_tics={"items": ["哼"]},
|
||
tags=["天才"],
|
||
relations=[],
|
||
)
|
||
]
|
||
|
||
class _WorldRepo:
|
||
async def list_for_project(self, project_id: uuid.UUID) -> list[WorldEntityView]:
|
||
return [
|
||
WorldEntityView(
|
||
type="力量体系",
|
||
name="灵脉",
|
||
rules={"rules": ["灵力守恒", "禁止跨界"]},
|
||
)
|
||
]
|
||
|
||
return MemoryRepos(
|
||
outline=_EmptyOutlineRepo(),
|
||
character=_CharRepo(),
|
||
world_entity=_WorldRepo(),
|
||
digest=_EmptyDigestRepo(),
|
||
foreshadow=_EmptyForeshadowRepo(),
|
||
style=_EmptyStyleRepo(),
|
||
rules=_EmptyRulesRepo(),
|
||
project=_StubProjectSpecRepo(),
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_characters_unpacks_jsonb_to_api_shape() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
app, _ = _make_app(project_repo=repo, gateway=_SchemaRoutingGateway({}), memory=_codex_memory())
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.get(f"/projects/{pid}/characters")
|
||
|
||
assert resp.status_code == 200
|
||
chars = resp.json()["characters"]
|
||
assert len(chars) == 1
|
||
card = chars[0]
|
||
# DB JSONB dict → API list/str（反向解包，入库形变的逆向）。
|
||
assert card["name"] == "叶寒"
|
||
assert card["traits"] == ["腹黑", "护短"]
|
||
assert card["speech_tics"] == ["哼"]
|
||
assert card["arc"] == "从孤儿到帝王"
|
||
assert card["tags"] == ["天才"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_world_entities_unpacks_rules_to_list() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
app, _ = _make_app(project_repo=repo, gateway=_SchemaRoutingGateway({}), memory=_codex_memory())
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.get(f"/projects/{pid}/world_entities")
|
||
|
||
assert resp.status_code == 200
|
||
entities = resp.json()["world_entities"]
|
||
assert len(entities) == 1
|
||
assert entities[0]["type"] == "力量体系"
|
||
assert entities[0]["name"] == "灵脉"
|
||
assert entities[0]["rules"] == ["灵力守恒", "禁止跨界"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_characters_empty_when_none_ingested() -> 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.get(f"/projects/{pid}/characters")
|
||
|
||
assert resp.status_code == 200
|
||
assert resp.json()["characters"] == []
|