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")