feat(toolbox): T6 创作工具箱通用生成器框架 — 8 新生成器 + 声明驱动落地页 + P2 收尾
通用执行路径驱动全部生成器（"加生成器=加一份声明"）：
- @llm: ww_agents +7 输出 schema + 7 spec（book-title/blurb/name/golden-finger/
glossary/opening/fine-outline，只声明 tier）+ build_outline_chapter_context
- @backend: ww_skills GeneratorTool 描述符 + TOOLBOX(11) + get_tool；3 通用端点
GET /skills/toolbox · POST .../skills/{tool_key}/generate(预览不写库,仅记账) ·
POST .../ingest(复用 continuity 409 + partition_writes 白名单)；纯 context 派发
- @frontend: 工具箱落地页 RSC + 声明驱动 GeneratorRunner + lib/toolbox 纯函数
+ LeftNav「工具箱」+ ⌘K nav-toolbox/action-gen-*；legacy 3 跳现页
- @qa: tests/test_t6_toolbox_e2e.py 5 用例真 pg + mock 网关零 token，无端点 bug
- P2 收尾: 限流→decisions.md 记延后(单用户原型)；noopener/Committable 早已修
守不变量 #2(只声明 tier)/#3(预览不写库,入库经验收 gate)/#9(缓存前缀)。无 DB 迁移。
门禁绿: 后端 ruff/format/mypy 195/alembic 无漂移/pytest 583；前端 lint/tsc/vitest 279/build。
spec 回写 PRODUCT_SPEC §7 + ARCHITECTURE §7.2 端点表。
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Yaojia Wang
@@ -492,7 +492,7 @@ async def test_m5_capability_degradation_marks_served_by_degraded() -> None:
|
||||
|
||||
# primary 不支持结构化输出，但仍能返回 parsed（自走 instructor JSON 提示）。
|
||||
degrade_adapter = _FakeGenAdapter(provider=_PRIMARY, structured=False)
|
||||
ledger: LedgerSink = _CapturingLedger() # type: ignore[assignment]
|
||||
ledger: LedgerSink = _CapturingLedger()
|
||||
gw = Gateway(
|
||||
adapters={_PRIMARY: degrade_adapter},
|
||||
ledger=ledger,
|
||||
|
||||
515
tests/test_t6_toolbox_e2e.py
Normal file
515
tests/test_t6_toolbox_e2e.py
Normal file
@@ -0,0 +1,515 @@
|
||||
"""T6.4 端到端：创作工具箱通用生成器端点（真实 DB，零 token，MOCK 网关）。
|
||||
|
||||
证明 T6 通用生成器框架闭环（声明驱动、一条执行路径 + 不变量 #3 预览不写库 / 入库必过
|
||||
continuity gate + 白名单）。镜像 `tests/test_m5_e2e.py` 的范式：真实 `Gateway` + 据
|
||||
`req.output_schema` 分支返回 `parsed` 的假适配器（绝不联网）+ 真实 `SqlAlchemyLedgerSink`
|
||||
（记账真落 pg）+ DB 真源断言。无 pg → skip。
|
||||
|
||||
可注入缝（Wave A）：`get_tier_gateway_builder` 返回 `TierGatewayBuilder`
|
||||
（`Callable[[Tier], Awaitable[Gateway]]`）。toolbox 端点 generate 调 `build_gateway(spec.tier)`、
|
||||
ingest 预检调 `build_gateway("analyst")`——故 override **这一个 builder**，让它返回单个真
|
||||
`Gateway`（假适配器据 output_schema 分支：生成产物 vs ContinuityReview）即可覆盖两路。
|
||||
|
||||
用例：
|
||||
1. `GET /skills/toolbox` 返回全 11 key（legacy 3 带 legacy_route+is_legacy=true；新 8）；
|
||||
brainstorm/glossary/golden-finger 的 ingestable 标志符合预期。
|
||||
2. brainstorm `generate` → 200 ToolGeneratePreviewResponse（output_kind + preview.ideas）；
|
||||
断言 **ledger 真落行** 且 **业务表零新增**（world_entities 计数不变）——不变量 #3。
|
||||
3. glossary `ingest` → world_entities：先**不带 acknowledge** + 预检有冲突 → **409
|
||||
CONFLICT_UNRESOLVED**，断言 world_entities **此刻无新行**；再带 acknowledge_conflicts=true
|
||||
→ 201，world_entities 行**真落 pg**（rules list→JSONB dict 形变保留）；rejected_tables 空。
|
||||
4. 未知 tool_key `generate` → 404 NOT_FOUND envelope。
|
||||
5. 无凭据（该 tier 无配置）→ 503 LLM_UNAVAILABLE envelope（不 override builder，走真依赖）。
|
||||
负向：响应体绝不泄露 token/secret；预览路径只 commit ledger。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from collections.abc import AsyncIterator, Awaitable, Callable
|
||||
from typing import Annotated
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from asgi_lifespan import LifespanManager
|
||||
from fastapi import Depends
|
||||
from sqlalchemy import delete, func, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||
from ww_agents import (
|
||||
Conflict,
|
||||
ContinuityReview,
|
||||
GlossaryResult,
|
||||
GlossaryTerm,
|
||||
GoldenFingerResult,
|
||||
IdeaListResult,
|
||||
)
|
||||
from ww_agents.schemas import GoldenFinger, Idea
|
||||
from ww_db import get_session, get_sessionmaker
|
||||
from ww_db.models import Project, UsageLedger, WorldEntity
|
||||
from ww_llm_gateway import Gateway, SqlAlchemyLedgerSink, resolve_route
|
||||
from ww_llm_gateway.adapters.base import (
|
||||
Capabilities,
|
||||
ProviderResult,
|
||||
ProviderUsage,
|
||||
StreamChunk,
|
||||
)
|
||||
from ww_llm_gateway.types import LlmRequest, Tier
|
||||
|
||||
# light/analyst/writer 默认都路由到 deepseek（config.tier_defaults）。
|
||||
_PROVIDER = "deepseek"
|
||||
|
||||
# 各档位假用量（喂记账；证明记账落库）。
|
||||
_USAGE = {
|
||||
"light": ProviderUsage(input_tokens=19, output_tokens=11),
|
||||
"analyst": ProviderUsage(input_tokens=23, output_tokens=7),
|
||||
"writer": ProviderUsage(input_tokens=31, output_tokens=17),
|
||||
}
|
||||
|
||||
# brainstorm 假产物：3 条脑洞（纯预览，不入库）。
|
||||
_IDEAS = IdeaListResult(
|
||||
ideas=[
|
||||
Idea(premise="天降系统但只在深夜生效", hook="主角被迫昼伏夜出", genre_fit="都市异能"),
|
||||
Idea(premise="转生成反派的家养灵宠", hook="一边卖萌一边算计", genre_fit="搞笑修真"),
|
||||
Idea(premise="末世里唯一能种田的人", hook="种子即权力", genre_fit="末世种田"),
|
||||
]
|
||||
)
|
||||
|
||||
# glossary 假产物（生成预览用；ingest 由前端映射成 world_entities，故端点不强依赖此形）。
|
||||
_GLOSSARY = GlossaryResult(
|
||||
terms=[
|
||||
GlossaryTerm(
|
||||
name="玄晶", type="货币", definition="修真界硬通货", rules=["不可伪造", "遇火则碎"]
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
# golden-finger 假产物（生成预览用）。
|
||||
_GOLDEN = GoldenFingerResult(
|
||||
systems=[
|
||||
GoldenFinger(
|
||||
name="九转炼体诀",
|
||||
mechanism="吞噬天地灵气淬体",
|
||||
growth="每转突破一境",
|
||||
limits="每转需一枚玄晶，不可逆修",
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
# 待入库的 world_entities（贴 WorldEntityCardView：type/name/rules）。
|
||||
_INGEST_ENTITIES = [
|
||||
{"type": "货币", "name": "玄晶", "rules": ["不可伪造", "遇火则碎"]},
|
||||
{"type": "力量体系", "name": "九转炼体诀", "rules": ["每转需一枚玄晶", "不可逆修"]},
|
||||
]
|
||||
|
||||
# precheck 假冲突（让入库 gate 触发 409）。
|
||||
_PRECHECK_CONFLICT = ContinuityReview(
|
||||
conflicts=[
|
||||
Conflict(
|
||||
type="设定违例",
|
||||
where="玄晶",
|
||||
refs=["既有货币体系"],
|
||||
suggestion="玄晶『遇火则碎』与既有设定冲突，请裁决",
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
class _FakeToolboxAdapter:
|
||||
"""实现 `ProviderAdapter` Protocol：按 `req.output_schema` 分支返回 parsed，绝不联网。
|
||||
|
||||
- `IdeaListResult` → 3 条脑洞（light）。
|
||||
- `GlossaryResult` → 术语表（analyst）。
|
||||
- `GoldenFingerResult` → 金手指（writer）。
|
||||
- `ContinuityReview` → 带冲突的预检结果（analyst）——触发 ingest 409 gate。
|
||||
每路带 `ProviderUsage` → 网关据此落 usage_ledger。
|
||||
"""
|
||||
|
||||
def __init__(self, provider: str = _PROVIDER) -> None:
|
||||
self.provider = provider
|
||||
|
||||
def capabilities(self) -> Capabilities:
|
||||
return Capabilities(structured_output=True)
|
||||
|
||||
async def complete(self, req: LlmRequest, model: str) -> ProviderResult:
|
||||
schema = req.output_schema
|
||||
if schema is IdeaListResult:
|
||||
return ProviderResult(
|
||||
text=_IDEAS.model_dump_json(), parsed=_IDEAS, usage=_USAGE["light"]
|
||||
)
|
||||
if schema is GlossaryResult:
|
||||
return ProviderResult(
|
||||
text=_GLOSSARY.model_dump_json(), parsed=_GLOSSARY, usage=_USAGE["analyst"]
|
||||
)
|
||||
if schema is GoldenFingerResult:
|
||||
return ProviderResult(
|
||||
text=_GOLDEN.model_dump_json(), parsed=_GOLDEN, usage=_USAGE["writer"]
|
||||
)
|
||||
if schema is ContinuityReview:
|
||||
return ProviderResult(
|
||||
text=_PRECHECK_CONFLICT.model_dump_json(),
|
||||
parsed=_PRECHECK_CONFLICT,
|
||||
usage=_USAGE["analyst"],
|
||||
)
|
||||
raise AssertionError(f"unexpected output_schema in T6 fake adapter: {schema!r}")
|
||||
|
||||
async def stream(self, req: LlmRequest, model: str) -> AsyncIterator[StreamChunk]:
|
||||
# 工具箱生成走 run()（结构化），不走 stream——但 Protocol 要求实现。
|
||||
yield StreamChunk(usage=_USAGE["light"])
|
||||
raise AssertionError("T6 toolbox generation must not stream")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def e2e_sm() -> AsyncIterator[async_sessionmaker[AsyncSession]]:
|
||||
"""真实 DB session 工厂；无 pg 时跳过（每测试清缓存重建 engine、结束 dispose）。"""
|
||||
get_sessionmaker.cache_clear()
|
||||
maker = get_sessionmaker()
|
||||
try:
|
||||
async with maker() as probe:
|
||||
await probe.execute(select(1))
|
||||
except Exception:
|
||||
pytest.skip("postgres not reachable")
|
||||
yield maker
|
||||
await maker.kw["bind"].dispose()
|
||||
get_sessionmaker.cache_clear()
|
||||
|
||||
|
||||
def _tier_builder_override(
|
||||
adapter: _FakeToolboxAdapter,
|
||||
) -> Callable[[AsyncSession], Callable[[Tier], Awaitable[Gateway]]]:
|
||||
"""依赖覆盖：`get_tier_gateway_builder` → 返回「按 tier 建网关」的 builder。
|
||||
|
||||
builder 忽略 tier 返回同一个真 `Gateway`（单 provider 假适配器据 output_schema 分支 +
|
||||
真实 ledger，请求 session）——同时覆盖 generate（按 spec.tier）与 ingest 预检（analyst）。
|
||||
"""
|
||||
|
||||
def _override(
|
||||
session: Annotated[AsyncSession, Depends(get_session)],
|
||||
) -> Callable[[Tier], Awaitable[Gateway]]:
|
||||
async def _build(_tier: Tier) -> Gateway:
|
||||
return Gateway(
|
||||
adapters={adapter.provider: adapter},
|
||||
ledger=SqlAlchemyLedgerSink(session),
|
||||
resolver=resolve_route,
|
||||
)
|
||||
|
||||
return _build
|
||||
|
||||
return _override
|
||||
|
||||
|
||||
async def _ledger_count(sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> int:
|
||||
async with sm() as s:
|
||||
return int(
|
||||
(
|
||||
await s.execute(
|
||||
select(func.count())
|
||||
.select_from(UsageLedger)
|
||||
.where(UsageLedger.project_id == project_uuid)
|
||||
)
|
||||
).scalar_one()
|
||||
)
|
||||
|
||||
|
||||
async def _world_count(sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> int:
|
||||
async with sm() as s:
|
||||
return int(
|
||||
(
|
||||
await s.execute(
|
||||
select(func.count())
|
||||
.select_from(WorldEntity)
|
||||
.where(WorldEntity.project_id == project_uuid)
|
||||
)
|
||||
).scalar_one()
|
||||
)
|
||||
|
||||
|
||||
async def _cleanup(sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> None:
|
||||
"""按 FK 顺序清理。"""
|
||||
async with sm() as cleanup:
|
||||
await cleanup.execute(delete(UsageLedger).where(UsageLedger.project_id == project_uuid))
|
||||
await cleanup.execute(delete(WorldEntity).where(WorldEntity.project_id == project_uuid))
|
||||
await cleanup.execute(delete(Project).where(Project.id == project_uuid))
|
||||
await cleanup.commit()
|
||||
|
||||
|
||||
def _assert_no_secret_leak(payload: object) -> None:
|
||||
"""负向：响应体序列化后绝不含 token/secret/api_key 等敏感字样。"""
|
||||
import json
|
||||
|
||||
blob = json.dumps(payload, ensure_ascii=False).lower()
|
||||
for needle in ("api_key", "secret", "token", "bearer", "sk-", "credential"):
|
||||
assert needle not in blob, f"sensitive marker leaked into response: {needle!r}"
|
||||
|
||||
|
||||
async def test_t6_list_toolbox_returns_all_eleven_tools(
|
||||
e2e_sm: async_sessionmaker[AsyncSession],
|
||||
) -> None:
|
||||
"""用例 1：GET /skills/toolbox 返回全 11 key；legacy 3 带 legacy_route，新 8 走通用执行器。"""
|
||||
from ww_api.main import create_app
|
||||
|
||||
app = create_app()
|
||||
transport = httpx.ASGITransport(app=app)
|
||||
async with LifespanManager(app):
|
||||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||||
resp = await client.get("/skills/toolbox")
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
|
||||
tools = {t["key"]: t for t in body["tools"]}
|
||||
# 全 11 key（legacy 3 + 新 8）。
|
||||
assert set(tools) == {
|
||||
"worldbuilding",
|
||||
"character",
|
||||
"outline",
|
||||
"brainstorm",
|
||||
"book-title",
|
||||
"blurb",
|
||||
"name",
|
||||
"golden-finger",
|
||||
"glossary",
|
||||
"opening",
|
||||
"fine-outline",
|
||||
}
|
||||
# legacy（spec=None）：is_legacy=true + 带 legacy_route，不可经通用执行入库。
|
||||
legacy_keys = {"worldbuilding", "character", "outline"}
|
||||
for key in legacy_keys:
|
||||
assert tools[key]["is_legacy"] is True
|
||||
assert tools[key]["legacy_route"]
|
||||
# 新工具：is_legacy=false。
|
||||
for key in set(tools) - legacy_keys:
|
||||
assert tools[key]["is_legacy"] is False
|
||||
# ingestable 标志：glossary/golden-finger/fine-outline 可入库；brainstorm 纯预览。
|
||||
assert tools["glossary"]["ingestable"] is True
|
||||
assert tools["golden-finger"]["ingestable"] is True
|
||||
assert tools["fine-outline"]["ingestable"] is True
|
||||
assert tools["brainstorm"]["ingestable"] is False
|
||||
# brainstorm 描述符含一句话需求字段。
|
||||
assert any(f["name"] == "brief" for f in tools["brainstorm"]["input_fields"])
|
||||
_assert_no_secret_leak(body)
|
||||
|
||||
|
||||
async def test_t6_brainstorm_generate_previews_without_writing_business_tables(
|
||||
e2e_sm: async_sessionmaker[AsyncSession],
|
||||
) -> None:
|
||||
"""用例 2：brainstorm generate → 200 预览；ledger 真落行 + 业务表零新增（不变量 #3）。"""
|
||||
from ww_api.main import create_app
|
||||
from ww_api.services.project_deps import get_tier_gateway_builder
|
||||
|
||||
app = create_app()
|
||||
app.dependency_overrides[get_tier_gateway_builder] = _tier_builder_override(
|
||||
_FakeToolboxAdapter()
|
||||
)
|
||||
|
||||
transport = httpx.ASGITransport(app=app)
|
||||
project_uuid: uuid.UUID | None = None
|
||||
try:
|
||||
async with LifespanManager(app):
|
||||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||||
create_resp = await client.post(
|
||||
"/projects", json={"title": "T6 脑洞验证作品", "genre": "玄幻"}
|
||||
)
|
||||
assert create_resp.status_code == 201
|
||||
project_id = create_resp.json()["id"]
|
||||
project_uuid = uuid.UUID(project_id)
|
||||
|
||||
world_before = await _world_count(e2e_sm, project_uuid)
|
||||
|
||||
gen_resp = await client.post(
|
||||
f"/projects/{project_id}/skills/brainstorm/generate",
|
||||
json={"brief": "想要三个突破常规的脑洞"},
|
||||
)
|
||||
assert gen_resp.status_code == 200
|
||||
body = gen_resp.json()
|
||||
assert body["tool_key"] == "brainstorm"
|
||||
assert body["output_kind"] == "IdeaListResult"
|
||||
ideas = body["preview"]["ideas"]
|
||||
assert [i["premise"] for i in ideas] == [
|
||||
"天降系统但只在深夜生效",
|
||||
"转生成反派的家养灵宠",
|
||||
"末世里唯一能种田的人",
|
||||
]
|
||||
_assert_no_secret_leak(body)
|
||||
|
||||
# DB 真源：ledger 落 1 行（light）；业务表（world_entities）零新增（预览不写库）。
|
||||
assert project_uuid is not None
|
||||
assert await _ledger_count(e2e_sm, project_uuid) == 1
|
||||
assert await _world_count(e2e_sm, project_uuid) == world_before
|
||||
async with e2e_sm() as verify:
|
||||
row = (
|
||||
(
|
||||
await verify.execute(
|
||||
select(UsageLedger).where(UsageLedger.project_id == project_uuid)
|
||||
)
|
||||
)
|
||||
.scalars()
|
||||
.one()
|
||||
)
|
||||
assert row.provider == _PROVIDER
|
||||
assert row.input_tokens == _USAGE["light"].input_tokens
|
||||
finally:
|
||||
if project_uuid is not None:
|
||||
await _cleanup(e2e_sm, project_uuid)
|
||||
|
||||
|
||||
async def test_t6_glossary_ingest_gate_then_acknowledge_persists(
|
||||
e2e_sm: async_sessionmaker[AsyncSession],
|
||||
) -> None:
|
||||
"""用例 3：glossary ingest → 409 gate（无 ack + 有冲突）→ ack 放行 → world_entities 真落 pg。"""
|
||||
from ww_api.main import create_app
|
||||
from ww_api.services.project_deps import get_tier_gateway_builder
|
||||
|
||||
app = create_app()
|
||||
app.dependency_overrides[get_tier_gateway_builder] = _tier_builder_override(
|
||||
_FakeToolboxAdapter()
|
||||
)
|
||||
|
||||
transport = httpx.ASGITransport(app=app)
|
||||
project_uuid: uuid.UUID | None = None
|
||||
try:
|
||||
async with LifespanManager(app):
|
||||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||||
create_resp = await client.post(
|
||||
"/projects", json={"title": "T6 词条入库作品", "genre": "玄幻"}
|
||||
)
|
||||
assert create_resp.status_code == 201
|
||||
project_id = create_resp.json()["id"]
|
||||
project_uuid = uuid.UUID(project_id)
|
||||
|
||||
# 3a) 入库（未 acknowledge）→ precheck 有冲突 → 409 CONFLICT_UNRESOLVED。
|
||||
blocked = await client.post(
|
||||
f"/projects/{project_id}/skills/glossary/ingest",
|
||||
json={"world_entities": _INGEST_ENTITIES},
|
||||
)
|
||||
assert blocked.status_code == 409
|
||||
err = blocked.json()["error"]
|
||||
assert err["code"] == "CONFLICT_UNRESOLVED"
|
||||
details = err["details"]
|
||||
assert details["conflict_count"] == 1
|
||||
assert details["conflicts"][0]["type"] == "设定违例"
|
||||
assert details["conflicts"][0]["where"] == "玄晶"
|
||||
_assert_no_secret_leak(blocked.json())
|
||||
|
||||
# 入库被拦 → 此刻 world_entities 表应无行（不静默入库，守不变量 #3）。
|
||||
assert await _world_count(e2e_sm, project_uuid) == 0
|
||||
|
||||
# 3b) 作者裁决后带 acknowledge_conflicts=true 重发 → 201 入库。
|
||||
ok = await client.post(
|
||||
f"/projects/{project_id}/skills/glossary/ingest",
|
||||
json={"world_entities": _INGEST_ENTITIES, "acknowledge_conflicts": True},
|
||||
)
|
||||
assert ok.status_code == 201
|
||||
ok_body = ok.json()
|
||||
assert ok_body["table"] == "world_entities"
|
||||
assert ok_body["created"] == ["玄晶", "九转炼体诀"]
|
||||
# 合法路径（glossary writes=["world_entities"]）无越权 → rejected_tables 空。
|
||||
assert ok_body["rejected_tables"] == []
|
||||
_assert_no_secret_leak(ok_body)
|
||||
|
||||
# DB 真源断言：world_entities 行真落 pg + rules list→JSONB dict 形变保留。
|
||||
assert project_uuid is not None
|
||||
async with e2e_sm() as verify:
|
||||
rows = (
|
||||
(
|
||||
await verify.execute(
|
||||
select(WorldEntity)
|
||||
.where(WorldEntity.project_id == project_uuid)
|
||||
.order_by(WorldEntity.name)
|
||||
)
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
assert {r.name for r in rows} == {"玄晶", "九转炼体诀"}
|
||||
xuanjing = next(r for r in rows if r.name == "玄晶")
|
||||
assert xuanjing.type == "货币"
|
||||
# 形变：rules list → JSONB dict {"rules":[...]}（rules 列表保留）。
|
||||
assert xuanjing.rules == {"rules": ["不可伪造", "遇火则碎"]}
|
||||
|
||||
# 记账真落 pg：precheck(409 路径) + precheck(入库路径) = 2 条（入库本体不调网关）。
|
||||
ledger_rows = (
|
||||
(
|
||||
await verify.execute(
|
||||
select(UsageLedger).where(UsageLedger.project_id == project_uuid)
|
||||
)
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
assert len(ledger_rows) == 2
|
||||
assert all(r.provider == _PROVIDER for r in ledger_rows)
|
||||
finally:
|
||||
if project_uuid is not None:
|
||||
await _cleanup(e2e_sm, project_uuid)
|
||||
|
||||
|
||||
async def test_t6_unknown_tool_key_generate_returns_404(
|
||||
e2e_sm: async_sessionmaker[AsyncSession],
|
||||
) -> None:
|
||||
"""用例 4：未知 tool_key generate → 404 NOT_FOUND envelope。"""
|
||||
from ww_api.main import create_app
|
||||
from ww_api.services.project_deps import get_tier_gateway_builder
|
||||
|
||||
app = create_app()
|
||||
# builder 即使被覆盖，也应在工具解析阶段 404（不触达网关）。
|
||||
app.dependency_overrides[get_tier_gateway_builder] = _tier_builder_override(
|
||||
_FakeToolboxAdapter()
|
||||
)
|
||||
|
||||
transport = httpx.ASGITransport(app=app)
|
||||
project_uuid: uuid.UUID | None = None
|
||||
try:
|
||||
async with LifespanManager(app):
|
||||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||||
create_resp = await client.post("/projects", json={"title": "T6 未知工具作品"})
|
||||
assert create_resp.status_code == 201
|
||||
project_id = create_resp.json()["id"]
|
||||
project_uuid = uuid.UUID(project_id)
|
||||
|
||||
resp = await client.post(
|
||||
f"/projects/{project_id}/skills/does-not-exist/generate",
|
||||
json={"brief": "x"},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
err = resp.json()["error"]
|
||||
assert err["code"] == "NOT_FOUND"
|
||||
assert err["details"]["tool_key"] == "does-not-exist"
|
||||
_assert_no_secret_leak(resp.json())
|
||||
finally:
|
||||
if project_uuid is not None:
|
||||
await _cleanup(e2e_sm, project_uuid)
|
||||
|
||||
|
||||
async def test_t6_no_provider_returns_503(
|
||||
e2e_sm: async_sessionmaker[AsyncSession],
|
||||
) -> None:
|
||||
"""用例 5：无凭据（不 override builder，走真 build_gateway_for_tier）→ 503 LLM_UNAVAILABLE。"""
|
||||
from ww_api.main import create_app
|
||||
|
||||
app = create_app()
|
||||
# 不 override get_tier_gateway_builder → 真依赖在无凭据时抛 LLM_UNAVAILABLE。
|
||||
|
||||
transport = httpx.ASGITransport(app=app)
|
||||
project_uuid: uuid.UUID | None = None
|
||||
try:
|
||||
async with LifespanManager(app):
|
||||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||||
create_resp = await client.post("/projects", json={"title": "T6 无凭据作品"})
|
||||
assert create_resp.status_code == 201
|
||||
project_id = create_resp.json()["id"]
|
||||
project_uuid = uuid.UUID(project_id)
|
||||
|
||||
resp = await client.post(
|
||||
f"/projects/{project_id}/skills/brainstorm/generate",
|
||||
json={"brief": "脑洞"},
|
||||
)
|
||||
assert resp.status_code == 503
|
||||
err = resp.json()["error"]
|
||||
assert err["code"] == "LLM_UNAVAILABLE"
|
||||
_assert_no_secret_leak(resp.json())
|
||||
|
||||
# 503 后业务表与 ledger 均无新增（凭据探测在生成前拦下）。
|
||||
assert await _world_count(e2e_sm, project_uuid) == 0
|
||||
assert await _ledger_count(e2e_sm, project_uuid) == 0
|
||||
finally:
|
||||
if project_uuid is not None:
|
||||
await _cleanup(e2e_sm, project_uuid)
|
||||
Reference in New Issue
Block a user