bf39f50b2f
#5 规则 DELETE + id：暴露 RuleView.id（PK）；新增 DELETE /projects/{id}/rules/{rule_id} （项目/规则不存在→404，成功→204，按 (id,project_id) 限定）；rule_repo 加 list_for_project/delete；RulesPage 每条加删除（乐观删+回滚+toast）。assemble 侧 RuleView（缓存前缀）不动，列表另立 RuleListItemView。 #7 codex 角色 relations：写侧本已持久化、读端点 _existing_characters 硬编码 []。 加 _relations_from_jsonb 解析 {name,kind,note}，CodexPage 渲染关系 chip。 #8 角色入库幂等：SqlCharacterWriteRepo.create 改 (project_id,name) app 层 upsert—— 重复入库改更新而非插入；不加 UNIQUE/迁移（线上已有重复行会让约束迁移失败）。 #1 大纲卷过滤：GET /outline 支持可选 ?volume（无参=全部，向后兼容）；OutlineEditor 加「查看：全部/卷N」筛选，与生成目标卷解耦。 H3/#9 文风回炉锚点：StyleDriftSegment 加 text（逐字命中段），style.md 指示审稿输出； 前端按内容锚点定位回炉目标（idx 仅排序），命中失败 → 提示「无法定位该段」而非 静默 no-op。style golden fixture 已重生成。 契约变更已 pnpm gen:api（RuleView.id / DELETE rules / outline ?volume）。无迁移 （alembic 无漂移）。门禁绿：ruff/mypy(210)/alembic/pytest 760 · 前端 tsc/lint/vitest 329。
701 lines
24 KiB
Python
701 lines
24 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 cryptography.fernet import Fernet
|
||
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.rule_repo import RuleListItemView
|
||
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:
|
||
# 幂等：按 (project_id, name) upsert——同名更新既有行而非插重复（镜像 Sql 实现）。
|
||
existing = next(
|
||
(r for r in self.rows if r["project_id"] == project_id and r["name"] == name),
|
||
None,
|
||
)
|
||
if existing is not None:
|
||
existing.update(
|
||
{
|
||
"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=existing["id"], name=name, role=role)
|
||
row_id = uuid.uuid4()
|
||
self.rows.append(
|
||
{
|
||
"id": row_id,
|
||
"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=row_id, name=name, role=role)
|
||
|
||
|
||
class _FakeRulesReadRepo:
|
||
"""实现规则 repo 的读侧 `list_for_project`（带 id，供 list_rules 端点）。"""
|
||
|
||
def __init__(self, rules: list[RuleListItemView] | None = None) -> None:
|
||
self._rules = rules or []
|
||
|
||
async def list_for_project(self, project_id: uuid.UUID) -> list[RuleListItemView]:
|
||
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", Fernet.generate_key().decode())
|
||
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_rule_write_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_rule_write_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_list_rules_unknown_project_returns_404() -> None:
|
||
# QA MEDIUM 回归：GET 规则列表对不存在 project 返 404（原返误导性空 200，掩盖坏 id）。
|
||
app, _ = _make_app(project_repo=FakeProjectRepo(), gateway=object())
|
||
async with _client(app) as client:
|
||
resp = await client.get(f"/projects/{uuid.uuid4()}/rules")
|
||
assert resp.status_code == 404
|
||
assert resp.json()["error"]["code"] == ErrorCode.NOT_FOUND
|
||
|
||
|
||
# ---- 世界观生成 ----
|
||
|
||
|
||
@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_ingest_same_name_twice_updates_not_duplicates() -> None:
|
||
# Arrange：同一项目、同名角色入库两次，第二次字段不同。
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
gateway = _SchemaRoutingGateway({ContinuityReview: _no_conflicts()})
|
||
char_repo = _FakeCharacterWriteRepo()
|
||
app, _ = _make_app(project_repo=repo, gateway=gateway, char_repo=char_repo)
|
||
|
||
def _payload(role: str, trait: str) -> dict[str, Any]:
|
||
return {
|
||
"cards": [
|
||
{
|
||
"name": "叶寒",
|
||
"role": role,
|
||
"traits": [trait],
|
||
"backstory": "孤儿",
|
||
"arc": "成长",
|
||
"speech_tics": [],
|
||
"tags": [],
|
||
"relations": [],
|
||
}
|
||
]
|
||
}
|
||
|
||
async with _client(app) as client:
|
||
# Act：先入主角/腹黑，再以同名入对手/隐忍。
|
||
first = await client.post(f"/projects/{pid}/characters", json=_payload("主角", "腹黑"))
|
||
second = await client.post(f"/projects/{pid}/characters", json=_payload("对手", "隐忍"))
|
||
|
||
# Assert：仍是一行（按 name upsert），字段被更新而非新增重复。
|
||
assert first.status_code == 201
|
||
assert second.status_code == 201
|
||
assert len(char_repo.rows) == 1
|
||
assert char_repo.rows[0]["role"] == "对手"
|
||
assert char_repo.rows[0]["traits"] == ["隐忍"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ingest_relations_persist_on_write() -> None:
|
||
# Arrange：入库带关系网的角色卡。
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
gateway = _SchemaRoutingGateway({ContinuityReview: _no_conflicts()})
|
||
char_repo = _FakeCharacterWriteRepo()
|
||
app, _ = _make_app(project_repo=repo, gateway=gateway, char_repo=char_repo)
|
||
|
||
relations = [{"name": "苏离", "kind": "宿敌", "note": "灭门之仇"}]
|
||
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": relations,
|
||
}
|
||
]
|
||
},
|
||
)
|
||
|
||
# Assert：relations 持久化（写侧未丢）。
|
||
assert resp.status_code == 201
|
||
assert char_repo.rows[0]["relations"] == relations
|
||
|
||
|
||
def test_relations_from_jsonb_parses_and_skips_dirty() -> None:
|
||
# Arrange：混入脏条目（非 dict / 缺 name / 缺 kind）。
|
||
from ww_api.routers.generation import _relations_from_jsonb
|
||
|
||
raw = [
|
||
{"name": "苏离", "kind": "宿敌", "note": "灭门之仇"},
|
||
{"name": "无类型"}, # 缺 kind → 跳过
|
||
{"kind": "无名"}, # 缺 name → 跳过
|
||
"not-a-dict", # 非 dict → 跳过
|
||
{"name": "墨白", "kind": "师徒"}, # note 可缺
|
||
]
|
||
# Act
|
||
out = _relations_from_jsonb(raw)
|
||
# Assert：只保留有效两条，note 缺为 None。
|
||
assert [(r.name, r.kind, r.note) for r in out] == [
|
||
("苏离", "宿敌", "灭门之仇"),
|
||
("墨白", "师徒", None),
|
||
]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_characters_includes_relations() -> None:
|
||
# Arrange：读侧 memory view 带 relations（JSONB list）。
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
app, _ = _make_app(
|
||
project_repo=repo,
|
||
gateway=_SchemaRoutingGateway({}),
|
||
memory=_codex_memory_with_relations(),
|
||
)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.get(f"/projects/{pid}/characters")
|
||
|
||
# Assert：读端点不再丢 relations（修 #7）。
|
||
assert resp.status_code == 200
|
||
card = resp.json()["characters"][0]
|
||
assert card["relations"] == [{"name": "苏离", "kind": "宿敌", "note": "灭门之仇"}]
|
||
|
||
|
||
# ---- 读端点 ----
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_rules_returns_rules() -> None:
|
||
repo = FakeProjectRepo()
|
||
pid = await _seed_project(repo)
|
||
rid1, rid2 = uuid.uuid4(), uuid.uuid4()
|
||
rules_repo = _FakeRulesReadRepo(
|
||
[
|
||
RuleListItemView(id=rid1, level="project", content="主角不复活"),
|
||
RuleListItemView(id=rid2, 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] == ["主角不复活", "无脏话"]
|
||
# 列表带稳定 id（前端删除 handle）。
|
||
assert [r["id"] for r in rules] == [str(rid1), str(rid2)]
|
||
|
||
|
||
@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(),
|
||
)
|
||
|
||
|
||
def _codex_memory_with_relations() -> Any:
|
||
"""MemoryRepos：角色行携 relations（JSONB list），验读端点还原关系网（修 #7）。"""
|
||
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=[{"name": "苏离", "kind": "宿敌", "note": "灭门之仇"}],
|
||
)
|
||
]
|
||
|
||
class _WorldRepo:
|
||
async def list_for_project(self, project_id: uuid.UUID) -> list[WorldEntityView]:
|
||
return []
|
||
|
||
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"] == []
|