91 lines
2.4 KiB
Python
91 lines
2.4 KiB
Python
"""FastAPI 网关测试：v1 legacy 端点保持 + v2 端点（封闭，注入 mock 管线）。"""
|
|
import pytest
|
|
|
|
pytest.importorskip("fastapi")
|
|
pytest.importorskip("httpx")
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
import gateway.api as ga
|
|
from gateway.api import app
|
|
|
|
|
|
@pytest.fixture()
|
|
def client():
|
|
return TestClient(app)
|
|
|
|
|
|
@pytest.fixture()
|
|
def v2_client(client):
|
|
# 用 mock worker 构建真实 v2 管线并注入（无需 API key / 真实模型）
|
|
pipe = ga.build_v2_pipeline(worker_cfg_override={"backend": "mock"})
|
|
ga.set_pipeline(pipe)
|
|
return client
|
|
|
|
|
|
def test_health(client):
|
|
resp = client.get("/health")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["status"] == "ok"
|
|
assert "code" in data["domains"]
|
|
|
|
|
|
def test_chat_legacy(client):
|
|
resp = client.post("/chat/legacy", json={"query": "用 Python 写一个快速排序函数"})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["response"]
|
|
assert data["domain"] == "code"
|
|
assert "route" in data
|
|
|
|
|
|
def test_chat_v2(v2_client):
|
|
resp = v2_client.post("/chat", json={"query": "请介绍快速排序算法"})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["response"]
|
|
assert data["request_id"]
|
|
assert "fast_path" in data
|
|
assert "route" in data
|
|
assert data["status"] in ("fast_path", "done", "escalated", "failed")
|
|
|
|
|
|
def test_chat_empty_query(client):
|
|
resp = client.post("/chat", json={"query": ""})
|
|
assert resp.status_code == 422
|
|
|
|
|
|
def test_metrics(client):
|
|
resp = client.get("/metrics")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "router" in data
|
|
assert "cache" in data
|
|
assert "v2" in data
|
|
assert "review" in data
|
|
|
|
|
|
def test_workspace_not_found(client):
|
|
resp = client.get("/runs/nonexistent/workspace")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_web_ui_served(client):
|
|
resp = client.get("/")
|
|
assert resp.status_code == 200
|
|
assert resp.headers["content-type"].startswith("text/html")
|
|
assert "端云协同" in resp.text
|
|
assert "发送" in resp.text
|
|
|
|
|
|
def test_review_flow(client):
|
|
q = ga.get_review()
|
|
rid = q.enqueue("req-x", "q", "ans", tags=["safety"], reason="test")
|
|
assert q.count() >= 1
|
|
resp = client.get("/review/queue")
|
|
assert resp.status_code == 200
|
|
resp2 = client.post(f"/review/{rid}", params={"verdict": "approve"})
|
|
assert resp2.status_code == 200
|
|
assert resp2.json()["ok"] is True
|