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

"""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.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", 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_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"] == []