1652ad9d20
润色「再沟通」意见含糊时先反问给选项（路线A：问题走独立非流式 JSON 预检端点，正文仍走
既有 refine 一字不改）。新增：
- ClarifyDecision/ClarifyQuestion/ClarifyOption 结构化 schema（既有 output-schema 处，供
producer 与端点共用）；clarify_refine.md 教条（含糊→need_clarification+≤1问+2–4锚定选项+
自由输入；明确→verification 放行；防循环）；注册 clarify_refine_spec(analyst 档,#24)+
SCHEMA_CATALOG + 重生成金标准。
- clarify_node：build_clarify_request 纯函数(缓存前缀不含易变) + run_clarify（gateway.run
结构化，判别/校验失败确定性回退 need_clarification=false，只读不写库）。
- POST /projects/{id}/chapters/{no}/refine/clarify → ClarifyDecision（analyst 网关，404/503，
末尾 commit 记账）。**既有 refine 端点/RefineRequest/Response 完全未改**。
门禁绿：ruff/mypy 227/pytest 900(+test_clarify_spec/_node/style clarify)/alembic 无漂移。
532 lines
20 KiB
Python
532 lines
20 KiB
Python
"""T4.3 端点测试：学文风（jobs 异步）+ 回炉 + 最新指纹（内存替身，无 DB/无网络）。
|
||
|
||
覆盖：
|
||
- POST /style：写一行 job 返 202 `{job_id}`；BackgroundTask 跑完写 style_fingerprint
|
||
+ 置 job done（ASGITransport 下 background task 在请求返回前跑完，见 gotcha）；
|
||
- mode="update" → 第二次 append version+1；
|
||
- 项目不存在 → 404；无凭据 → 503（dep 解析阶段拦下，未写 job）；
|
||
- GET /style：返最新指纹（dimensions/evidence/version）；无指纹 → 404；
|
||
- POST /refine：返 {original, refined}（refiner 纯文本）+ 末尾 commit（记账落库）；无凭据 → 503。
|
||
|
||
学文风 work 在 `run_job` 自建独立 session 上自造网关 + repo（请求 session 已关闭）——
|
||
测试 monkeypatch `build_gateway_for_tier`（产 StyleFingerprintResult）+ 写侧 repo
|
||
（指到共享 fake repo），并 override `get_session_factory` 为 FakeSessionFactory，断言后台落库。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import uuid
|
||
|
||
import httpx
|
||
import pytest
|
||
from cryptography.fernet import Fernet
|
||
from fakes_projects import (
|
||
FakeJobRepo,
|
||
FakeProjectRepo,
|
||
FakeReviewGateway,
|
||
FakeSession,
|
||
FakeSessionFactory,
|
||
FakeStyleWriteRepo,
|
||
)
|
||
from fastapi import FastAPI
|
||
from ww_agents import StyleDimension, StyleFingerprintResult
|
||
from ww_core.domain.job_repo import STATUS_DONE
|
||
from ww_core.domain.project_repo import ProjectCreate
|
||
from ww_shared import AppError, ErrorCode
|
||
|
||
|
||
def _fingerprint() -> StyleFingerprintResult:
|
||
return StyleFingerprintResult(
|
||
dimensions=[
|
||
StyleDimension(name="句长节奏", value="短句为主", evidence=["他来了。", "她走了。"]),
|
||
StyleDimension(name="叙事人称", value="第三人称", evidence=["他看着远方。"]),
|
||
]
|
||
)
|
||
|
||
|
||
async def _seed_project(repo: FakeProjectRepo) -> uuid.UUID:
|
||
view = await repo.create(uuid.UUID(int=1), ProjectCreate(title="测试作品", genre="玄幻"))
|
||
return uuid.UUID(str(view.id))
|
||
|
||
|
||
def _app_with_overrides(
|
||
*,
|
||
project_repo: FakeProjectRepo,
|
||
job_repo: FakeJobRepo | None = None,
|
||
style_repo: FakeStyleWriteRepo | None = None,
|
||
session: FakeSession | None = None,
|
||
session_factory: FakeSessionFactory | None = None,
|
||
extract_gateway: object | None = None,
|
||
refine_gateway: object | None = None,
|
||
clarify_gateway: object | None = None,
|
||
) -> FastAPI:
|
||
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_clarify_gateway,
|
||
get_job_repo,
|
||
get_project_repo,
|
||
get_refine_gateway,
|
||
get_session_factory,
|
||
get_style_extract_gateway,
|
||
get_style_write_repo,
|
||
)
|
||
from ww_db import get_session
|
||
|
||
app = create_app()
|
||
app.dependency_overrides[get_project_repo] = lambda: project_repo
|
||
app.dependency_overrides[get_session] = lambda: session or FakeSession()
|
||
if job_repo is not None:
|
||
app.dependency_overrides[get_job_repo] = lambda: job_repo
|
||
if style_repo is not None:
|
||
app.dependency_overrides[get_style_write_repo] = lambda: style_repo
|
||
if session_factory is not None:
|
||
app.dependency_overrides[get_session_factory] = lambda: session_factory
|
||
if extract_gateway is not None:
|
||
app.dependency_overrides[get_style_extract_gateway] = lambda: extract_gateway
|
||
if refine_gateway is not None:
|
||
app.dependency_overrides[get_refine_gateway] = lambda: refine_gateway
|
||
if clarify_gateway is not None:
|
||
app.dependency_overrides[get_clarify_gateway] = lambda: clarify_gateway
|
||
return app
|
||
|
||
|
||
def _client(app: FastAPI) -> httpx.AsyncClient:
|
||
transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)
|
||
return httpx.AsyncClient(transport=transport, base_url="http://test")
|
||
|
||
|
||
# ---- POST /style（学文风 jobs 异步）----
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_learn_style_returns_202_and_writes_job_row(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
job_repo = FakeJobRepo()
|
||
session = FakeSession()
|
||
app = _app_with_overrides(
|
||
project_repo=project_repo,
|
||
job_repo=job_repo,
|
||
session=session,
|
||
session_factory=FakeSessionFactory(),
|
||
extract_gateway=FakeReviewGateway(parsed=_fingerprint()),
|
||
)
|
||
|
||
# background task work 触网关/repo → monkeypatch 成 no-op fake（本用例不验落库）；
|
||
# run_job 默认 repo_factory 用 SqlJobRepo(FakeSession) 会触 .execute → 改用 fake job repo。
|
||
import ww_api.routers.style as style_mod
|
||
import ww_api.services.job_runner as runner_mod
|
||
|
||
async def _noop_work(_session: object) -> dict[str, int]:
|
||
return {"version": 1, "dims_count": 2}
|
||
|
||
monkeypatch.setattr(style_mod, "_make_style_learn_work", lambda *_a, **_k: _noop_work)
|
||
monkeypatch.setattr(runner_mod, "SqlJobRepo", lambda _s: job_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{pid}/style", json={"samples": ["样本正文一"]})
|
||
|
||
assert resp.status_code == 202
|
||
body = resp.json()
|
||
job_id = uuid.UUID(body["job_id"])
|
||
# job 行已创建 + 请求阶段 commit 一次（供前端立即轮询）。
|
||
assert job_id in job_repo.rows
|
||
assert job_repo.rows[job_id].kind == "style_learn"
|
||
assert session.commits == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_learn_style_background_task_writes_fingerprint_and_completes_job(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
job_repo = FakeJobRepo()
|
||
style_repo = FakeStyleWriteRepo()
|
||
session_factory = FakeSessionFactory()
|
||
|
||
import ww_api.routers.style as style_mod
|
||
|
||
# work 内 build_gateway_for_tier → fake 网关产指纹；写侧 repo → 共享 fake repo。
|
||
async def _fake_build_gateway(_session, _store, _tier): # type: ignore[no-untyped-def]
|
||
return FakeReviewGateway(parsed=_fingerprint())
|
||
|
||
monkeypatch.setattr(style_mod, "build_gateway_for_tier", _fake_build_gateway)
|
||
monkeypatch.setattr(style_mod, "SqlStyleFingerprintWriteRepo", lambda _s: style_repo)
|
||
# `run_job` 用 SqlJobRepo（默认 repo_factory）会触 FakeSession.execute → 改用 fake job repo。
|
||
import ww_api.services.job_runner as runner_mod
|
||
|
||
monkeypatch.setattr(runner_mod, "SqlJobRepo", lambda _s: job_repo)
|
||
|
||
app = _app_with_overrides(
|
||
project_repo=project_repo,
|
||
job_repo=job_repo,
|
||
session=FakeSession(),
|
||
session_factory=session_factory,
|
||
extract_gateway=FakeReviewGateway(parsed=_fingerprint()),
|
||
)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/style", json={"samples": ["样本一", "样本二"], "mode": "create"}
|
||
)
|
||
|
||
assert resp.status_code == 202
|
||
job_id = uuid.UUID(resp.json()["job_id"])
|
||
# BackgroundTask 已跑完（ASGITransport 等其完成）：style_fingerprint 落库 + job done。
|
||
assert style_repo.append_calls == 1
|
||
assert style_repo.rows[1][0] == {"句长节奏": "短句为主", "叙事人称": "第三人称"}
|
||
assert style_repo.rows[1][1]["句长节奏"] == ["他来了。", "她走了。"]
|
||
assert job_repo.rows[job_id].status == STATUS_DONE
|
||
assert job_repo.rows[job_id].result == {"version": 1, "dims_count": 2}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_learn_style_update_mode_bumps_version(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
job_repo = FakeJobRepo()
|
||
style_repo = FakeStyleWriteRepo()
|
||
# 预置 version 1（首学已存在）。
|
||
await style_repo.append(pid, dimensions_json={"x": "y"}, evidence_json={"x": []})
|
||
|
||
import ww_api.routers.style as style_mod
|
||
import ww_api.services.job_runner as runner_mod
|
||
|
||
async def _fake_build_gateway(_session, _store, _tier): # type: ignore[no-untyped-def]
|
||
return FakeReviewGateway(parsed=_fingerprint())
|
||
|
||
monkeypatch.setattr(style_mod, "build_gateway_for_tier", _fake_build_gateway)
|
||
monkeypatch.setattr(style_mod, "SqlStyleFingerprintWriteRepo", lambda _s: style_repo)
|
||
monkeypatch.setattr(runner_mod, "SqlJobRepo", lambda _s: job_repo)
|
||
|
||
app = _app_with_overrides(
|
||
project_repo=project_repo,
|
||
job_repo=job_repo,
|
||
session=FakeSession(),
|
||
session_factory=FakeSessionFactory(),
|
||
extract_gateway=FakeReviewGateway(parsed=_fingerprint()),
|
||
)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/style", json={"samples": ["新样本"], "mode": "update"}
|
||
)
|
||
|
||
assert resp.status_code == 202
|
||
job_id = uuid.UUID(resp.json()["job_id"])
|
||
# 第二次 append → version 2（version+1）。
|
||
assert max(style_repo.rows) == 2
|
||
assert job_repo.rows[job_id].result == {"version": 2, "dims_count": 2}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_learn_style_unknown_project_404() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
app = _app_with_overrides(
|
||
project_repo=project_repo,
|
||
job_repo=FakeJobRepo(),
|
||
session_factory=FakeSessionFactory(),
|
||
extract_gateway=FakeReviewGateway(parsed=_fingerprint()),
|
||
)
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{uuid.uuid4()}/style", json={"samples": ["x"]})
|
||
assert resp.status_code == 404
|
||
assert resp.json()["error"]["code"] == ErrorCode.NOT_FOUND
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_learn_style_without_credentials_503_and_no_job() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
job_repo = FakeJobRepo()
|
||
session = FakeSession()
|
||
|
||
async def _no_creds() -> object:
|
||
raise AppError(ErrorCode.LLM_UNAVAILABLE, "未配置凭据", {"provider": "deepseek"})
|
||
|
||
app = _app_with_overrides(
|
||
project_repo=project_repo,
|
||
job_repo=job_repo,
|
||
session=session,
|
||
session_factory=FakeSessionFactory(),
|
||
)
|
||
from ww_api.services.project_deps import get_style_extract_gateway
|
||
|
||
app.dependency_overrides[get_style_extract_gateway] = _no_creds
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{pid}/style", json={"samples": ["x"]})
|
||
|
||
assert resp.status_code == 503
|
||
assert resp.json()["error"]["code"] == ErrorCode.LLM_UNAVAILABLE
|
||
# 凭据探测在调度 job 之前 → 未写 job、未 commit。
|
||
assert job_repo.rows == {}
|
||
assert session.commits == 0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_learn_style_empty_samples_422() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
app = _app_with_overrides(
|
||
project_repo=project_repo,
|
||
job_repo=FakeJobRepo(),
|
||
session_factory=FakeSessionFactory(),
|
||
extract_gateway=FakeReviewGateway(parsed=_fingerprint()),
|
||
)
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{pid}/style", json={"samples": []})
|
||
assert resp.status_code == 422
|
||
|
||
|
||
# ---- GET /style（最新指纹）----
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_style_returns_latest_fingerprint() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
style_repo = FakeStyleWriteRepo()
|
||
await style_repo.append(
|
||
pid, dimensions_json={"句长": "短"}, evidence_json={"句长": ["他来了。"]}
|
||
)
|
||
app = _app_with_overrides(project_repo=project_repo, style_repo=style_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.get(f"/projects/{pid}/style")
|
||
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["version"] == 1
|
||
# P2 codegen：dimensions 改为 list[DimensionEntry]（name/value/evidence 合并），
|
||
# 替代此前两列并行裸 dict（强类型给 TS 客户端）。
|
||
assert body["dimensions"] == [{"name": "句长", "value": "短", "evidence": ["他来了。"]}]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_style_no_fingerprint_404() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
app = _app_with_overrides(project_repo=project_repo, style_repo=FakeStyleWriteRepo())
|
||
async with _client(app) as client:
|
||
resp = await client.get(f"/projects/{pid}/style")
|
||
assert resp.status_code == 404
|
||
assert resp.json()["error"]["code"] == ErrorCode.NOT_FOUND
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_style_unknown_project_404() -> None:
|
||
app = _app_with_overrides(project_repo=FakeProjectRepo(), style_repo=FakeStyleWriteRepo())
|
||
async with _client(app) as client:
|
||
resp = await client.get(f"/projects/{uuid.uuid4()}/style")
|
||
assert resp.status_code == 404
|
||
|
||
|
||
# ---- POST /refine（同步回炉）----
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_refine_returns_original_and_refined_and_commits() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
session = FakeSession()
|
||
gateway = FakeReviewGateway(text="重写后的段落。")
|
||
app = _app_with_overrides(project_repo=project_repo, session=session, refine_gateway=gateway)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/chapters/1/refine",
|
||
json={"segment": "原始段落。", "instruction": "去掉机翻腔"},
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["original"] == "原始段落。"
|
||
assert body["refined"] == "重写后的段落。"
|
||
# writer 档 + 末尾 commit（记账落库）。
|
||
assert gateway.requests[0].tier == "writer"
|
||
assert session.commits == 1
|
||
# 指令进入了输入文本。
|
||
assert "去掉机翻腔" in str(gateway.requests[0].input)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_refine_unknown_project_404() -> None:
|
||
app = _app_with_overrides(
|
||
project_repo=FakeProjectRepo(), refine_gateway=FakeReviewGateway(text="x")
|
||
)
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{uuid.uuid4()}/chapters/1/refine", json={"segment": "原段"}
|
||
)
|
||
assert resp.status_code == 404
|
||
assert resp.json()["error"]["code"] == ErrorCode.NOT_FOUND
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_refine_empty_segment_422() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
app = _app_with_overrides(project_repo=project_repo, refine_gateway=FakeReviewGateway(text="x"))
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{pid}/chapters/1/refine", json={"segment": ""})
|
||
assert resp.status_code == 422
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_refine_without_credentials_503() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
session = FakeSession()
|
||
|
||
async def _no_creds() -> object:
|
||
raise AppError(ErrorCode.LLM_UNAVAILABLE, "未配置凭据", {"provider": "deepseek"})
|
||
|
||
app = _app_with_overrides(project_repo=project_repo, session=session)
|
||
from ww_api.services.project_deps import get_refine_gateway
|
||
|
||
app.dependency_overrides[get_refine_gateway] = _no_creds
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{pid}/chapters/1/refine", json={"segment": "原段"})
|
||
|
||
assert resp.status_code == 503
|
||
assert resp.json()["error"]["code"] == ErrorCode.LLM_UNAVAILABLE
|
||
assert session.commits == 0
|
||
|
||
|
||
# ---- POST /refine/clarify（润色预检澄清）----
|
||
|
||
|
||
def _need_clarification() -> object:
|
||
from ww_agents import ClarifyDecision, ClarifyOption, ClarifyQuestion
|
||
|
||
return ClarifyDecision(
|
||
need_clarification=True,
|
||
questions=[
|
||
ClarifyQuestion(
|
||
question="你想让这段更有张力，是指哪种方向？",
|
||
options=[
|
||
ClarifyOption(label="加快节奏", value="拆短句、压缩铺陈"),
|
||
ClarifyOption(label="加重冲突", value="强化人物对立"),
|
||
],
|
||
allow_free_text=True,
|
||
)
|
||
],
|
||
)
|
||
|
||
|
||
def _clear_decision() -> object:
|
||
from ww_agents import ClarifyDecision
|
||
|
||
return ClarifyDecision(
|
||
need_clarification=False,
|
||
verification="我会把被字句改成主动句，其余不动，对吗？",
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_clarify_returns_need_clarification_and_commits() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
session = FakeSession()
|
||
gateway = FakeReviewGateway(parsed=_need_clarification())
|
||
app = _app_with_overrides(project_repo=project_repo, session=session, clarify_gateway=gateway)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/chapters/1/refine/clarify",
|
||
json={"segment": "原始段落。", "instruction": "更有张力"},
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["need_clarification"] is True
|
||
assert len(body["questions"]) == 1
|
||
assert body["questions"][0]["options"][0]["label"] == "加快节奏"
|
||
assert body["questions"][0]["allow_free_text"] is True
|
||
# analyst 档 + 末尾 commit（记账落库）；指令折进输入文本。
|
||
assert gateway.requests[0].tier == "analyst"
|
||
assert session.commits == 1
|
||
assert "更有张力" in str(gateway.requests[0].input)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_clarify_returns_clear_decision_read_only() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
session = FakeSession()
|
||
gateway = FakeReviewGateway(parsed=_clear_decision())
|
||
app = _app_with_overrides(project_repo=project_repo, session=session, clarify_gateway=gateway)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/chapters/2/refine/clarify",
|
||
json={"segment": "把这句的被字句改成主动句。", "instruction": ""},
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["need_clarification"] is False
|
||
assert body["questions"] == []
|
||
assert body["verification"] is not None
|
||
# 只读不写库：无写侧 repo 参与，仅一次 commit 落网关 usage（不变量 #3）。
|
||
assert session.commits == 1
|
||
assert session.rollbacks == 0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_clarify_unknown_project_404() -> None:
|
||
app = _app_with_overrides(
|
||
project_repo=FakeProjectRepo(),
|
||
clarify_gateway=FakeReviewGateway(parsed=_clear_decision()),
|
||
)
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{uuid.uuid4()}/chapters/1/refine/clarify",
|
||
json={"segment": "原段", "instruction": "x"},
|
||
)
|
||
assert resp.status_code == 404
|
||
assert resp.json()["error"]["code"] == ErrorCode.NOT_FOUND
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_clarify_empty_segment_422() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
app = _app_with_overrides(
|
||
project_repo=project_repo, clarify_gateway=FakeReviewGateway(parsed=_clear_decision())
|
||
)
|
||
async with _client(app) as client:
|
||
resp = await client.post(f"/projects/{pid}/chapters/1/refine/clarify", json={"segment": ""})
|
||
assert resp.status_code == 422
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_clarify_without_credentials_503() -> None:
|
||
project_repo = FakeProjectRepo()
|
||
pid = await _seed_project(project_repo)
|
||
session = FakeSession()
|
||
|
||
async def _no_creds() -> object:
|
||
raise AppError(ErrorCode.LLM_UNAVAILABLE, "未配置凭据", {"provider": "deepseek"})
|
||
|
||
app = _app_with_overrides(project_repo=project_repo, session=session)
|
||
from ww_api.services.project_deps import get_clarify_gateway
|
||
|
||
app.dependency_overrides[get_clarify_gateway] = _no_creds
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/chapters/1/refine/clarify", json={"segment": "原段"}
|
||
)
|
||
|
||
assert resp.status_code == 503
|
||
assert resp.json()["error"]["code"] == ErrorCode.LLM_UNAVAILABLE
|
||
assert session.commits == 0
|