hunjing/backend/tests/test_m7_xingyi.py

"""M7 验收测试（rewrite-plan M7 验收门）。

① 生成内容未定相不成为正式形仪（candidate 不进呈现位；定相才转正；主仪唯一）；
② 克隆授权可撤回（撤回立即生效，合成前置校验被拒；过期同理）；
③ 关闭全部多模态后文字相伴完整可用（ASR 503 降级时镜树/缘契链路照常——回归）。
"""

from collections.abc import AsyncGenerator

import pytest
from httpx import ASGITransport, AsyncClient

from app.main import app


@pytest.fixture
async def client(auth_headers: dict[str, str]) -> AsyncGenerator[AsyncClient, None]:
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test", headers=auth_headers) as c:
        yield c


async def _character(client: AsyncClient) -> dict:
    resp = await client.get("/api/v1/characters/active")
    assert resp.status_code == 200, resp.text
    return resp.json()


# ── ① 未定相不成为正式形仪 ──────────────────────────────


@pytest.mark.asyncio
async def test_candidate_not_primary_until_adopted(client: AsyncClient) -> None:
    character = await _character(client)
    base = f"/api/v1/characters/{character['id']}"

    # 登记两个主仪候选（生成来源）
    c1 = await client.post(
        f"{base}/forms",
        json={"kind": "primary", "source": "image_gen", "description": "立绘候选一"},
    )
    assert c1.status_code == 201, c1.text
    c2 = await client.post(
        f"{base}/forms",
        json={"kind": "primary", "source": "image_gen", "description": "立绘候选二"},
    )
    assert c2.status_code == 201

    # 未定相：呈现位为空（candidate 不进正式形仪）
    primary = await client.get(f"{base}/forms/primary")
    assert primary.status_code == 200
    assert primary.json() is None

    # 定相候选一 → 成为主仪
    adopt1 = await client.post(f"{base}/forms/{c1.json()['id']}/adopt")
    assert adopt1.status_code == 200
    assert adopt1.json()["status"] == "adopted"
    primary = await client.get(f"{base}/forms/primary")
    assert primary.json()["id"] == c1.json()["id"]

    # 定相候选二 → 主仪唯一（旧主仪自动 retired）
    adopt2 = await client.post(f"{base}/forms/{c2.json()['id']}/adopt")
    assert adopt2.status_code == 200
    primary = await client.get(f"{base}/forms/primary")
    assert primary.json()["id"] == c2.json()["id"]
    forms = await client.get(f"{base}/forms")
    by_id = {f["id"]: f for f in forms.json()}
    assert by_id[c1.json()["id"]]["status"] == "retired"
    assert by_id[c2.json()["id"]]["status"] == "adopted"

    # 废弃候选
    c3 = await client.post(
        f"{base}/forms", json={"kind": "costume", "source": "upload", "description": "化仪候选"}
    )
    discard = await client.post(f"{base}/forms/{c3.json()['id']}/discard")
    assert discard.json()["status"] == "discarded"


# ── ② 克隆授权可撤回 ──────────────────────────────


@pytest.mark.asyncio
async def test_voice_clone_grant_revocable(client: AsyncClient) -> None:
    character = await _character(client)
    base = f"/api/v1/characters/{character['id']}"

    # 授权克隆必须声明用途与传播范围
    incomplete = await client.post(
        f"{base}/voices", json={"source": "authorized_clone", "name": "授权声线"}
    )
    assert incomplete.status_code == 422 or "用途" in incomplete.text

    voice = await client.post(
        f"{base}/voices",
        json={
            "source": "authorized_clone",
            "name": "授权声线",
            "params": {"timbre": "warm"},
            "grant_purpose": "人物语音消息",
            "grant_distribution": "仅本镜域私用",
        },
    )
    assert voice.status_code == 201, voice.text
    voice_id = voice.json()["id"]

    # 授权有效：合成前置校验通过
    usable = await client.post(f"{base}/voices/{voice_id}/assert-usable")
    assert usable.status_code == 200

    # 撤回 → 立即生效
    revoked = await client.post(f"{base}/voices/{voice_id}/revoke-grant")
    assert revoked.status_code == 200
    assert revoked.json()["effective"] == "immediate"
    denied = await client.post(f"{base}/voices/{voice_id}/assert-usable")
    assert denied.status_code == 409 or denied.status_code == 403

    # 合成预设不受授权约束
    preset = await client.post(
        f"{base}/voices", json={"source": "synthesis_preset", "name": "预设音色 A"}
    )
    assert preset.status_code == 201
    ok = await client.post(f"{base}/voices/{preset.json()['id']}/assert-usable")
    assert ok.status_code == 200


# ── ③ 关闭全部多模态后文字相伴完整可用（回归） ──────────────────────────────


@pytest.mark.asyncio
async def test_multimodal_off_text_companion_still_complete(
    client: AsyncClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    from app.domains.xingyi import asr as xingyi_asr

    character = await _character(client)
    base = f"/api/v1/characters/{character['id']}"

    # 多模态全关：元炁未配置 → ASR 503（不降级不伪造）
    monkeypatch.setattr(xingyi_asr.settings, "yuanqi_base_url", "")
    muted = await client.post(
        f"{base}/voice/transcriptions",
        files={"audio": ("a.webm", b"fake-bytes", "audio/webm")},
    )
    assert muted.status_code == 503

    # 文字相伴链路照常：镜树问答（文字）+ 缘契（文字）
    tree = await client.post(
        "/api/v1/trees", json={"character_id": character["id"], "title": "纯文字回归"}
    )
    assert tree.status_code in (200, 201)
    cov = await client.post(
        "/api/v1/covenants",
        json={"character_id": character["id"], "dimensions": {"companionship": "text-only"}},
    )
    assert cov.status_code == 201
    caps = await client.get("/api/v1/capabilities")
    assert caps.status_code == 200