tzt
c54ad23c88
- _run_chat 开头缓存分支：cacheable（stop+单轮）-> canonical_hash -> L1/L2 查 -> 命中：try_hold->立即 settle(cost=0, charged=售价口径, status=cached, gateway_cached=1) -> SSE 合成回放或 JSON（X-Cache: HIT） - 未命中流程收尾写缓存（_json_response 尾部，stop 且单轮）； 缓存层任何故障降级直连上游（不影响可用性） - 缓存命中计费 _cached_charge（成本 0，入按字符估收售价——全毛利杠杆 L0） - _get_semcache 按 db_path 分实例（测试隔离）+ reset_semcache_instances - 测试 +1（端到端：首问 miss 打上游/二问 HIT 上游仅 1 次/账目 cached）， 全量 423 passed
325 lines
13 KiB
Python
325 lines
13 KiB
Python
"""代理路由端到端测试（T-P4，M1 验收）：全程计费/错误码/SSE 流式/httpx 合规客户端。"""
|
||
import asyncio
|
||
import json
|
||
|
||
import httpx
|
||
import pytest
|
||
|
||
pytest.importorskip("fastapi")
|
||
|
||
from fastapi import FastAPI
|
||
from fastapi.testclient import TestClient
|
||
|
||
import gateway.proxy.upstream as up
|
||
import gateway.model_pool as mp
|
||
from gateway.model_pool import PoolStore
|
||
from gateway.proxy.auth import issue_key, reset_auth_state
|
||
from gateway.proxy.config import build_proxy_config
|
||
from gateway.proxy.errors import (
|
||
BalanceError,
|
||
BodyTooLargeError,
|
||
ProxyAuthError,
|
||
QuotaError,
|
||
)
|
||
|
||
|
||
def asyncio_run(coro):
|
||
return asyncio.run(coro)
|
||
|
||
|
||
UPSTREAM_SSE = "\n\n".join([
|
||
'data: {"choices":[{"delta":{"role":"assistant","content":"\u4f60\u597d"}}]}',
|
||
'data: {"choices":[{"delta":{"content":"\uff0c\u4e16\u754c"}}]}',
|
||
'data: {"choices":[{"delta":{},"finish_reason":"stop"}],'
|
||
# 智能体任务级用量：in_miss=600k, in_hit=300k, out=80k
|
||
'"usage":{"prompt_tokens":900000,"prompt_cache_hit_tokens":300000,"completion_tokens":80000}}',
|
||
"data: [DONE]",
|
||
]) + "\n\n"
|
||
|
||
|
||
def _make_app(tmp_path, upstream_body=UPSTREAM_SSE, fail_upstream=False):
|
||
"""独立 FastAPI app + 池 + key。返回 (client, ledger, key, upstream_calls)。"""
|
||
mp.reset_pool()
|
||
mp._store = PoolStore(path=tmp_path / "pool.json")
|
||
mp.get_pool().upsert({
|
||
"id": "up1", "name": "云端", "tier": "budget", "backend": "openai",
|
||
"base_url": "http://upstream.test", "model": "deepseek-chat",
|
||
"api_key": "up-key", "provider": "deepseek",
|
||
"price_in": 0.1, "price_out": 0.1, "enabled": True})
|
||
reset_auth_state()
|
||
|
||
calls = {"n": 0}
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
calls["n"] += 1
|
||
if fail_upstream:
|
||
raise httpx.ConnectError("上游不可达")
|
||
return httpx.Response(200, content=upstream_body.encode("utf-8"))
|
||
|
||
import gateway.proxy.upstream as upmod
|
||
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
||
orig = upmod._client
|
||
upmod._client = client
|
||
|
||
cfg = build_proxy_config({"proxy": {
|
||
"enabled": True,
|
||
"db_path": str(tmp_path / "proxy.sqlite3"),
|
||
"buckets": {"default": {"system_template": "你是校园学习助手。",
|
||
"doc_prefix_file": None, "doc_version": 1,
|
||
"ttl_hours": 72}},
|
||
"pricing": {"deepseek-chat": {"in_miss": 3.0, "in_hit": 0.1, "out": 9.0},
|
||
"peak_window": {"start": "00:00", "end": "23:59"},
|
||
"offpeak_factor": 0.5,
|
||
"sale_discount": {"in": 0.5, "out": 0.8}},
|
||
"limits": {"rpm_per_key": 100, "day_req_cap": 1000,
|
||
"concurrent_per_key": 8, "max_body_chars": 6000},
|
||
}})
|
||
from gateway.proxy.routes import build_proxy_router, install_error_handlers
|
||
app = FastAPI()
|
||
app.include_router(build_proxy_router(cfg, mp.get_pool()))
|
||
install_error_handlers(app)
|
||
ledger = app.router.routes # noqa: F841（占位说明：ledger 经 state 可取）
|
||
st_ledger = None
|
||
for r in app.routes:
|
||
st = getattr(r, "state", None)
|
||
# 直接从 routes builder 拿 ledger：router.state 在 APIRouter 上不可用，
|
||
# 用模块内函数重建一次（同 db path 幂等）
|
||
from gateway.proxy.ledger import Ledger
|
||
st_ledger = Ledger.init_db(tmp_path / "proxy.sqlite3")
|
||
|
||
tc = TestClient(app)
|
||
key = issue_key(st_ledger, _mk_student(st_ledger), rpm_cap=100, day_cap_req=1000)
|
||
return tc, st_ledger, key, calls, (upmod, orig, client)
|
||
|
||
|
||
def _mk_student(ledger):
|
||
return ledger.upsert_student("学生A", balance_yuan=10.0, daily_cap_yuan=100)
|
||
|
||
|
||
def _auth(key):
|
||
return {"Authorization": f"Bearer {key['key']}"}
|
||
|
||
|
||
BODY = {"model": "deepseek-chat",
|
||
"messages": [{"role": "user",
|
||
"content": "请详细介绍快速排序算法的原理、复杂度与实现要点，"
|
||
"并给出 Python 示例代码与适用场景分析。"}]}
|
||
|
||
|
||
def test_chat_non_stream_end_to_end(tmp_path):
|
||
"""非流式全程：回答 + OpenAI 形状 + 账本三值一致 + usage 注入。"""
|
||
tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
|
||
try:
|
||
r = tc.post("/proxy/v1/chat/completions", json=BODY, headers=_auth(key))
|
||
assert r.status_code == 200, r.text
|
||
data = r.json()
|
||
assert data["object"] == "chat.completion"
|
||
assert data["model"] == "deepseek-chat"
|
||
assert "你好，世界" == data["choices"][0]["message"]["content"]
|
||
usage = data["usage"]
|
||
assert usage["prompt_tokens"] == 900000 and usage["completion_tokens"] == 80000
|
||
assert calls["n"] == 1
|
||
# 账本三值一致（手算黄金用例：in_miss=600k, in_hit=300k, out=80k）
|
||
rid = r.headers["x-request-id"]
|
||
u = ledger.get_usage(rid)
|
||
assert u["status"] == "ok"
|
||
# cost = 600k×3000/1M + 300k×100/1M + 80k×9000/1M = 1800+30+720 = 2550 毫元
|
||
# charged = 900 + 15 + 576 = 1491 毫元（in 5 折 / out 8 折；分项 round 后求和）
|
||
assert u["upstream_cost_milli"] == 2550
|
||
assert u["charged_milli"] == 1491
|
||
assert u["margin_milli"] == u["charged_milli"] - u["upstream_cost_milli"]
|
||
assert u["in_miss_tok"] == 600000 and u["in_hit_tok"] == 300000
|
||
assert u["out_tok"] == 80000
|
||
finally:
|
||
upmod._client = orig
|
||
try:
|
||
asyncio_run(hclient.aclose())
|
||
except Exception:
|
||
pass
|
||
mp.reset_pool()
|
||
reset_auth_state()
|
||
|
||
|
||
def test_chat_stream_end_to_end(tmp_path):
|
||
"""流式全程：SSE 逐块、usage chunk 被过滤（客户端未要求）、终态 DONE。"""
|
||
tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
|
||
try:
|
||
with tc.stream("POST", "/proxy/v1/chat/completions",
|
||
json={**BODY, "stream": True},
|
||
headers=_auth(key)) as resp:
|
||
assert resp.status_code == 200
|
||
assert resp.headers["content-type"].startswith("text/event-stream")
|
||
lines = [l for l in resp.iter_lines() if l.strip()]
|
||
assert lines[-1] == "data: [DONE]"
|
||
text = "\n".join(lines)
|
||
assert "usage" not in text # 客户端未要求 -> 过滤
|
||
assert "\u4f60\u597d" in text
|
||
rid = resp.headers["x-request-id"]
|
||
u = ledger.get_usage(rid)
|
||
assert u["status"] == "ok" and u["charged_milli"] > 0
|
||
finally:
|
||
upmod._client = orig
|
||
try:
|
||
asyncio_run(hclient.aclose())
|
||
except Exception:
|
||
pass
|
||
mp.reset_pool()
|
||
reset_auth_state()
|
||
|
||
|
||
def test_402_when_balance_insufficient(tmp_path):
|
||
"""余额不足 -> 402（预扣失败）。"""
|
||
tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
|
||
try:
|
||
sid = ledger.list_usage() and None
|
||
# 直接清空该 key 学生余额
|
||
row = ledger.find_key(key["key"] and __import__("hashlib").sha256(
|
||
key["key"].encode()).hexdigest())
|
||
ledger.topup(row["student_id"], -10.0) # 余额归零
|
||
r = tc.post("/proxy/v1/chat/completions", json=BODY, headers=_auth(key))
|
||
assert r.status_code == 402
|
||
assert calls["n"] == 0 # 未打上游
|
||
finally:
|
||
upmod._client = orig
|
||
try:
|
||
asyncio_run(hclient.aclose())
|
||
except Exception:
|
||
pass
|
||
mp.reset_pool()
|
||
reset_auth_state()
|
||
|
||
|
||
def test_401_invalid_key(tmp_path):
|
||
tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
|
||
try:
|
||
r = tc.post("/proxy/v1/chat/completions", json=BODY,
|
||
headers={"Authorization": "Bearer sk-campus-wrong"})
|
||
assert r.status_code == 401
|
||
finally:
|
||
upmod._client = orig
|
||
mp.reset_pool()
|
||
reset_auth_state()
|
||
|
||
|
||
def test_429_day_cap(tmp_path):
|
||
"""日请求上限 -> 429（配置 cap=2）。"""
|
||
tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
|
||
try:
|
||
ok = 0
|
||
for _ in range(3):
|
||
r = tc.post("/proxy/v1/chat/completions", json=BODY, headers=_auth(key))
|
||
if r.status_code == 200:
|
||
ok += 1
|
||
assert ok >= 1
|
||
finally:
|
||
upmod._client = orig
|
||
try:
|
||
asyncio_run(hclient.aclose())
|
||
except Exception:
|
||
pass
|
||
mp.reset_pool()
|
||
reset_auth_state()
|
||
|
||
|
||
def test_413_body_too_large(tmp_path):
|
||
tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
|
||
try:
|
||
big = {"model": "deepseek-chat",
|
||
"messages": [{"role": "user", "content": "x" * 7000}]}
|
||
r = tc.post("/proxy/v1/chat/completions", json=big, headers=_auth(key))
|
||
assert r.status_code == 413
|
||
finally:
|
||
upmod._client = orig
|
||
mp.reset_pool()
|
||
reset_auth_state()
|
||
|
||
|
||
def test_502_upstream_fail_voids_hold(tmp_path):
|
||
"""上游全挂 -> 502 且预扣全额退（余额不变）。"""
|
||
tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path,
|
||
fail_upstream=True)
|
||
try:
|
||
row = ledger.find_key(__import__("hashlib").sha256(
|
||
key["key"].encode()).hexdigest())
|
||
before = ledger.get_student(row["student_id"])["balance_milli"]
|
||
r = tc.post("/proxy/v1/chat/completions", json=BODY, headers=_auth(key))
|
||
assert r.status_code == 502
|
||
after = ledger.get_student(row["student_id"])["balance_milli"]
|
||
assert before == after # void 全退
|
||
finally:
|
||
upmod._client = orig
|
||
try:
|
||
asyncio_run(hclient.aclose())
|
||
except Exception:
|
||
pass
|
||
mp.reset_pool()
|
||
reset_auth_state()
|
||
|
||
|
||
def test_openai_protocol_compliance_via_httpx(tmp_path):
|
||
"""httpx 手写 OpenAI SDK 合规断言：流式可被标准解析器消费、形状正确。"""
|
||
tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
|
||
try:
|
||
# 模拟 OpenAI SDK 的流式解析路径
|
||
with tc.stream("POST", "/proxy/v1/chat/completions",
|
||
json={**BODY, "stream": True,
|
||
"stream_options": {"include_usage": True}},
|
||
headers=_auth(key)) as resp:
|
||
collected = []
|
||
usage_seen = False
|
||
done = False
|
||
for line in resp.iter_lines():
|
||
if not line.startswith("data:"):
|
||
continue
|
||
payload = line[5:].strip()
|
||
if payload == "[DONE]":
|
||
done = True
|
||
continue
|
||
obj = json.loads(payload)
|
||
assert obj.get("object") == "chat.completion.chunk"
|
||
for choice in obj.get("choices", []):
|
||
collected.append(choice.get("delta", {}).get("content") or "")
|
||
if obj.get("usage"):
|
||
usage_seen = True
|
||
assert done and usage_seen
|
||
assert "".join(collected) == "你好，世界"
|
||
finally:
|
||
upmod._client = orig
|
||
try:
|
||
asyncio_run(hclient.aclose())
|
||
except Exception:
|
||
pass
|
||
mp.reset_pool()
|
||
reset_auth_state()
|
||
|
||
|
||
def test_cache_hit_second_request(tmp_path):
|
||
"""T-P6 端到端：首问打上游并写缓存；同问再答 X-Cache: HIT 且上游仅 1 次。"""
|
||
from gateway.proxy.routes import reset_semcache_instances
|
||
reset_semcache_instances()
|
||
tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
|
||
try:
|
||
r1 = tc.post("/proxy/v1/chat/completions", json=BODY, headers=_auth(key))
|
||
assert r1.status_code == 200
|
||
assert r1.headers.get("x-cache") is None # 首问未命中
|
||
assert calls["n"] == 1
|
||
r2 = tc.post("/proxy/v1/chat/completions", json=BODY, headers=_auth(key))
|
||
assert r2.status_code == 200
|
||
assert r2.headers.get("x-cache") == "HIT" # 缓存命中
|
||
assert calls["n"] == 1 # 上游仍只 1 次
|
||
assert r2.json()["choices"][0]["message"]["content"] == "你好，世界"
|
||
rid = r2.headers["x-request-id"]
|
||
u = ledger.get_usage(rid)
|
||
assert u["status"] == "cached" and u["gateway_cached"] == 1
|
||
assert u["upstream_cost_milli"] == 0 # 缓存命中成本 0（全毛利）
|
||
# 小请求售价取整后可为 0（D-P1 毫元整数语义）；大请求 charged>0 由真实流量体现
|
||
finally:
|
||
reset_semcache_instances()
|
||
upmod._client = orig
|
||
try:
|
||
asyncio_run(hclient.aclose())
|
||
except Exception:
|
||
pass
|
||
mp.reset_pool()
|
||
reset_auth_state()
|