mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 14:31:58 +00:00
ThearchyHelios
d14fc97da7
api/apps.py 里这两处直接回 JSONResponse（detail + errors），上一次补错误码时只查了 HTTPException，漏了它们：外部系统发起 实例时表单不对，仍然只能按中文句子认。detail 和 errors 不变。 tests/test_integration_errors.py 的静态检查跟着扩：这几个模块里直接回的 JSONResponse 错误，内容里也要有 code（改之前 这条检查报出了这两处）。
170 lines
7.9 KiB
Python
170 lines
7.9 KiB
Python
"""集成接口的错误码与 /docs（04 细案 A7）。
|
||
|
||
外部系统按 code 认是哪种错，不按中文句子：应用（api/apps.py）、待办（api/tasks.py）、工件（api/artifacts.py）的错误，
|
||
和角色、运行读权限的拒绝（api/deps.py、auth.require）全部带 code；detail 还是原来那一句，老前端照常显示。
|
||
|
||
开了登录，/docs、/redoc、/openapi.json 和 /api 一样要登录、要搭建者：以前中间件只拦 /api/ 开头的路径，接口全貌对
|
||
匿名的人敞开。关登录照旧（tests/test_provenance_contract.py 的断言不受影响）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import ast
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from httpx import ASGITransport, AsyncClient
|
||
|
||
from app.api import coded
|
||
from app.db.base import SessionLocal
|
||
from app.db.models_apps import App
|
||
from app.main import app as main_app
|
||
from tests.fakes import auth_env
|
||
|
||
APP_DIR = Path(__file__).resolve().parents[1] / "app"
|
||
DOCS = ("/docs", "/redoc", "/openapi.json")
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
async def clean():
|
||
await auth_env.reset()
|
||
try:
|
||
yield
|
||
finally:
|
||
await auth_env.reset()
|
||
|
||
|
||
def client() -> AsyncClient:
|
||
return AsyncClient(transport=ASGITransport(app=main_app), base_url="http://test")
|
||
|
||
|
||
async def new_app() -> str:
|
||
async with SessionLocal() as session:
|
||
row = App(name=f"报销-{uuid.uuid4().hex[:6]}")
|
||
session.add(row)
|
||
await session.commit()
|
||
return row.id
|
||
|
||
|
||
def _called(node: ast.AST, name: str) -> bool:
|
||
return isinstance(node, ast.Call) and (getattr(node.func, "id", None) or getattr(node.func, "attr", None)) == name
|
||
|
||
|
||
def test_no_error_without_a_code_left_in_the_integration_modules():
|
||
"""这几个模块里不许再直接抛 HTTPException（没有 code），一律 CodedHTTPException；直接回 JSONResponse 的错误（表单
|
||
没通过校验那几处）内容里也要有 code。"""
|
||
for rel in ("api/apps.py", "api/tasks.py", "api/artifacts.py", "api/deps.py", "auth/__init__.py"):
|
||
tree = ast.parse((APP_DIR / rel).read_text())
|
||
bare = [node.lineno for node in ast.walk(tree) if _called(node, "HTTPException")]
|
||
assert not bare, f"{rel} 第 {bare} 行抛的 HTTPException 没有 code"
|
||
for node in ast.walk(tree):
|
||
if not _called(node, "JSONResponse"):
|
||
continue
|
||
content = next((k.value for k in node.keywords if k.arg == "content"), node.args[0] if node.args else None)
|
||
keys = {k.value for k in getattr(content, "keys", []) if isinstance(k, ast.Constant)}
|
||
assert "code" in keys, f"{rel} 第 {node.lineno} 行回的 JSONResponse 没有 code"
|
||
|
||
|
||
def test_codes_agree_with_the_ones_already_in_use():
|
||
from app.apps import entries
|
||
|
||
assert coded.APP_NOT_FOUND == entries.APP_GONE, "入口发起时的「应用不存在」和这里同一个码"
|
||
|
||
|
||
async def test_app_errors_carry_codes_and_keep_their_detail():
|
||
app_id = await new_app()
|
||
async with client() as c:
|
||
r = await c.get("/api/apps/nope-nope")
|
||
assert (r.status_code, r.json()) == (404, {"detail": "应用不存在，可能已被删除", "code": "app_not_found"})
|
||
r = await c.get(f"/api/apps/{app_id}/versions/9")
|
||
assert r.status_code == 404 and r.json()["code"] == "app_version_not_found" and "没有 v9" in r.json()["detail"]
|
||
r = await c.get(f"/api/apps/{app_id}/pages/home", params={"params": "[1, 2]", "draft": 1})
|
||
assert (r.status_code, r.json()) == (422, {"detail": "页面参数需要是 JSON 对象", "code": "app_params_invalid"})
|
||
r = await c.patch(f"/api/apps/{app_id}", json={"draft_bindings": {"slots": {"财务": "不是列表"}}})
|
||
assert r.status_code == 422 and r.json()["code"] == "app_bindings_invalid", r.text
|
||
r = await c.post(f"/api/apps/{app_id}/hooks/nope/rotate")
|
||
assert r.status_code == 404 and r.json()["code"] == "app_hook_not_declared"
|
||
r = await c.patch(f"/api/apps/{app_id}/hooks/nope", json={"enabled": False})
|
||
assert r.status_code == 404 and r.json()["code"] == "app_hook_not_found", r.text
|
||
r = await c.post("/api/apps/derive", json={"workflow_id": "nope-nope", "version": 1})
|
||
assert (r.status_code, r.json()["code"]) == (404, "workflow_not_found")
|
||
|
||
|
||
async def test_task_errors_carry_codes():
|
||
async with client() as c:
|
||
r = await c.get("/api/tasks/nope-nope")
|
||
assert (r.status_code, r.json()) == (404, {"detail": "待办不存在，所属的运行可能已被删除", "code": "task_not_found"})
|
||
# 服务层没带码的拒绝（这里是待办不存在）按状态补上同一类的码
|
||
r = await c.post("/api/tasks/nope-nope/respond", json={"decision": "approve"})
|
||
assert (r.status_code, r.json()["code"]) == (404, "task_not_found")
|
||
r = await c.post("/api/tasks/nope-nope/remind")
|
||
assert (r.status_code, r.json()["code"]) == (404, "task_not_found")
|
||
|
||
|
||
async def test_artifact_errors_carry_codes():
|
||
async with client() as c:
|
||
r = await c.get(f"/api/artifacts/{'0' * 64}")
|
||
assert (r.status_code, r.json()) == (404, {"detail": "工件不存在，可能已被删除", "code": "artifact_not_found"})
|
||
|
||
|
||
async def test_role_and_run_refusals_carry_codes():
|
||
await auth_env.turn_on_local()
|
||
await auth_env.make_user("uma", "user")
|
||
uma = await auth_env.login("uma")
|
||
r = await uma.get("/api/workflows")
|
||
assert r.status_code == 403 and r.json() == {"detail": "没有权限：这项操作需要搭建者及以上的角色",
|
||
"code": "role_required"}
|
||
r = await uma.get("/api/tasks", params={"assigned": "all"})
|
||
assert r.status_code == 403 and r.json()["code"] == "role_required"
|
||
r = await uma.get("/api/runs/nope-nope/evidence")
|
||
assert r.status_code == 403 and r.json() == {"detail": "没有权限查看这次运行", "code": "run_forbidden"}
|
||
r = await uma.get(f"/api/artifacts/{'0' * 64}")
|
||
assert r.status_code == 403 and r.json() == {"detail": "没有权限查看这件工件", "code": "artifact_forbidden"}
|
||
r = await uma.get("/api/runs")
|
||
assert r.status_code == 403 and r.json()["code"] == "role_required"
|
||
|
||
|
||
async def test_ensure_role_inside_an_endpoint_carries_codes():
|
||
from app.api.deps import Principal, ensure_role, reset_principal, set_principal
|
||
|
||
for principal, status, code in ((Principal(username=None, role="", mode="local"), 401, "auth_required"),
|
||
(Principal(username="bob", role="builder", verified=True, mode="local"), 403,
|
||
"role_required")):
|
||
token = set_principal(principal)
|
||
try:
|
||
with pytest.raises(coded.CodedHTTPException) as e:
|
||
ensure_role("admin")
|
||
finally:
|
||
reset_principal(token)
|
||
assert (e.value.status_code, e.value.code) == (status, code)
|
||
assert "管理员" in e.value.detail
|
||
|
||
|
||
# ---------------------------------------------------------------- /docs
|
||
|
||
|
||
async def test_docs_stay_open_when_login_is_off():
|
||
async with client() as c:
|
||
for path in DOCS:
|
||
assert (await c.get(path)).status_code == 200, path
|
||
|
||
|
||
async def test_docs_need_a_builder_once_login_is_on():
|
||
await auth_env.turn_on_local()
|
||
await auth_env.make_user("uma", "user")
|
||
await auth_env.make_user("bob", "builder")
|
||
async with client() as anonymous:
|
||
for path in DOCS:
|
||
r = await anonymous.get(path)
|
||
assert r.status_code == 401 and r.json()["code"] == "auth_required", path
|
||
# 子路径（Swagger 的 OAuth 回跳页）一样
|
||
assert (await anonymous.get("/docs/oauth2-redirect")).status_code == 401
|
||
uma = await auth_env.login("uma")
|
||
for path in DOCS:
|
||
r = await uma.get(path)
|
||
assert r.status_code == 403 and r.json()["code"] == "role_required", path
|
||
bob = await auth_env.login("bob")
|
||
for path in DOCS:
|
||
assert (await bob.get(path)).status_code == 200, path
|
||
assert "/api/apps/{app_id}" in (await bob.get("/openapi.json")).json()["paths"]
|