From 38625771435e6a92be5c31af06ad5b77395c1310 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Mon, 19 Oct 2026 17:51:46 +0800 Subject: [PATCH] =?UTF-8?q?test:=20=E8=B7=AF=E5=BE=84=E6=A8=A1=E5=BC=8F?= =?UTF-8?q?=E5=85=A8=E9=93=BE=E4=B8=8E=E9=99=90=E6=B5=81=E9=9A=94=E7=A6=BB?= =?UTF-8?q?=E9=92=89=E4=BD=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- tests/test_host_routing.py | 89 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 89 insertions(+) diff --git a/tests/test_host_routing.py b/tests/test_host_routing.py index 97136a5..0c568e6 100644 --- a/tests/test_host_routing.py +++ b/tests/test_host_routing.py @@ -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")