test: 路径模式全链与限流隔离钉位

Task 3:TestPathPrefixRouting 追加 5 用例——
- chat REST 全链(建会话/resume/公开历史,全程 /t/ 前缀 + 未知 Host)
- 限流隔离:同 IP 同设备双租户独立计数(打满 jiancai 30 上限 429,
  auto 消息通路仍 200——Task 2 桶键含 slug 的验收钉位)
- _PREFIX_RE.groupindex == {}(TENANT_SLUG_RE 派生捕获组漂移守护)
- WS 停用租户路径模式 1013(Host 未知,1013 只能出自路径分支)
- 双层前缀 /t/a/t/b 仅剥一层(剩余路径无路由 → 路由层 JSON 404)

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-10-19 17:51:46 +08:00
parent 9b97eafa1c
commit 3862577143
+89
View File
@@ -12,6 +12,7 @@ from types import SimpleNamespace
import pytest
from starlette.websockets import WebSocketDisconnect
from app.engine.services import conversation as conv_mod
from app.main import app
from app.platform.host_middleware import TenantHostMiddleware
from app.platform.registry import TenantRegistry
@@ -320,3 +321,91 @@ class TestPathPrefixRouting:
headers={"host": AUTO_HOST})
assert r.status_code == 200
assert r.json()["slug"] == "jiancai"
def test_chat_REST路径模式_建会话与消息(self, app_client):
# REST 最小真实链路(照抄 test_chat_rest_session_history 手法,
# 全程 /t/ 前缀 + 未知 Host):建会话 → resume 命中同 sid →
# 公开历史端点 200 空表(app-source 会话)
r = app_client.post("/t/jiancai/api/chat/session", json={},
headers={"host": NOPE_HOST})
assert r.status_code == 200
sid = r.json()["session_id"]
assert r.json()["is_new"] is True
r2 = app_client.post("/t/jiancai/api/chat/session",
json={"session_id": sid},
headers={"host": NOPE_HOST})
assert r2.status_code == 200
assert r2.json()["is_new"] is False
assert r2.json()["session_id"] == sid
r3 = app_client.get(f"/t/jiancai/api/chat/{sid}/messages",
headers={"host": NOPE_HOST})
assert r3.status_code == 200
assert r3.json() == []
def test_限流隔离_同IP双租户独立计数(self, app_client, monkeypatch):
# Task 2 验收钉位:chat_rest 限流桶键含 runtime.config.slug——同
# IP(TestClient 恒 "testclient")同设备双租户独立计数。手法照抄
# test_chat_rest_message 的 429 用例(monkeypatch stream + 打满
# 默认上限 30,seed 未覆盖 rate_limit_per_hour);桶键唯一差异即
# slug,打满 jiancai 后 auto 消息通路必须仍 200。
async def fake_stream(self, sid, content):
yield {"type": "stream_chunk", "content": "ok"}
yield {"type": "evaluated", "ended": False,
"intent_level": "pending"}
monkeypatch.setattr(
conv_mod.ConversationManager, "stream_process_message",
fake_stream)
dev = {"host": NOPE_HOST, "X-Device-Id": "shared-device"}
sid_jc = app_client.post("/t/jiancai/api/chat/session", json={},
headers={"host": NOPE_HOST}
).json()["session_id"]
for _ in range(30):
r = app_client.post(
"/t/jiancai/api/chat/message",
json={"session_id": sid_jc, "content": "x"}, headers=dev)
assert r.status_code == 200
over = app_client.post(
"/t/jiancai/api/chat/message",
json={"session_id": sid_jc, "content": "x"}, headers=dev)
assert over.status_code == 429
# 另一家(auto)同 IP 同设备:桶独立,消息通路仍可用
sid_auto = app_client.post("/t/auto/api/chat/session", json={},
headers={"host": NOPE_HOST}
).json()["session_id"]
r_auto = app_client.post(
"/t/auto/api/chat/message",
json={"session_id": sid_auto, "content": "x"}, headers=dev)
assert r_auto.status_code == 200
assert r_auto.json()["reply"] == "ok"
def test_PREFIX_RE_无捕获组_派生钉位(self):
# _PREFIX_RE 从 TENANT_SLUG_RE.pattern 派生(掐锚拼接),中间件按
# m.group(1)=slug / m.group(2)=剩余路径 取值——若源模式将来加
# 命名捕获组,组号静默漂移、group(2) 语义移位。groupindex 只含
# 命名组,== {} 把"偶发不漂移"变有意守护:源模式一加命名组即红。
assert TenantHostMiddleware._PREFIX_RE.groupindex == {}
def test_WS停用租户路径模式_1013(self, disabled_client):
# Host 用未知值:1013 只可能出自路径分支 resolve_by_slug 的
# TenantDisabledError——若路径模式没接停用语义,未知 Host 该是
# 1008(断言形态照抄 test_disabled_tenant_ws_rejected_with_1013)
with pytest.raises(WebSocketDisconnect) as ei:
with disabled_client.websocket_connect(
"/t/jiancai/ws/chat", headers={"host": NOPE_HOST}):
pass # pragma: no cover — 未 accept 即 close,进不来
assert ei.value.code == 1013
def test_双层前缀_t_a_t_b_仅剥一层(self, app_client):
# 剥一层后剩余 /t/auto/api/public/brand 不再二次剥离、也无路由
# → 404;JSON body 证明到了路由层(中间件 404 是 text/plain,
# test_health_exemption_is_exact_match 的判别手法)——钉住
# 「只剥一层」语义
r = app_client.get("/t/jiancai/t/auto/api/public/brand",
headers={"host": NOPE_HOST})
assert r.status_code == 404
assert r.headers["content-type"].startswith("application/json")