# Task 7: Host 中间件租户路由——纯 ASGI(http+websocket 双覆盖)。 # # 本文件的存在意义(前序审查实证过的坑): 若中间件走了 BaseHTTPMiddleware # (@app.middleware("http")),WS 请求会整个绕过租户解析落 get_runtime 兜底, # auto 域名的 WS 流量静默写进 jiancai 库(串库不崩溃,极难发现)。 # test_ws_routes_to_correct_tenant_db 用双库断言钉死这条红线。 import json import shutil 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 from tests.conftest import ( AUTO_HOST, JIANCAI_HOST, NOPE_HOST, SEED_DIR, app_test_client, build_app, ) # ---- 中间件挂载顺序守护 ------------------------------------------------------- def test_tenant_middleware_is_outermost_user_middleware(): # add_middleware 后加者居 user_middleware[0] = 最外层。租户解析必须 # 先于一切用户中间件(里层中间件/路由才读得到 state.tenant)——未来 # 新中间件若更晚 add_middleware 插到租户中间件外层,此测试即红。 assert app.user_middleware[0].cls is TenantHostMiddleware # ---- 平台级豁免 ------------------------------------------------------------- def test_health_served_for_any_host(app_client): # /health 在任何 Host 下可达(部署健康检查 curl 127.0.0.1:8018, # Host 是 IP:端口,对 registry 必然未知) assert app_client.get("/health").status_code == 200 r = app_client.get("/health", headers={"host": NOPE_HOST}) assert r.status_code == 200 def test_health_exemption_is_exact_match(app_client): # 精确 == "/health",/healthxxx 不豁免:未知 Host 由中间件(而非路由) # 拒绝——text/plain 的 PlainTextResponse 区别于路由 404 的 JSON body r = app_client.get("/healthxxx", headers={"host": NOPE_HOST}) assert r.status_code == 404 assert r.headers["content-type"].startswith("text/plain") # ---- 未知 Host -------------------------------------------------------------- def test_unknown_host_gets_404(app_client): r = app_client.get("/api/chat/session", headers={"host": NOPE_HOST}) assert r.status_code == 404 def test_unknown_host_ws_rejected_with_1008(app_client): # 探针实证: 服务器未 accept 直接 close 时,TestClient 在 __enter__ 即抛 # WebSocketDisconnect,code/reason 保留——这就是断言方式 with pytest.raises(WebSocketDisconnect) as ei: with app_client.websocket_connect("/ws/chat", headers={"host": NOPE_HOST}): pass # pragma: no cover — 未 accept 即 close,进不来 assert ei.value.code == 1008 # ---- 停用租户 --------------------------------------------------------------- @pytest.fixture def disabled_client(tmp_path): """jiancai 停用版 app_client: 拷 seed 到 tmp、改 status 再加载。 加载前改盘(而非 registry_reload): 无需跨事件循环碰 registry 锁, 最简;reload 语义本身已被 Task 5a 的测试钉过,此处不重复走。""" seed = tmp_path / "seed" shutil.copytree(SEED_DIR, seed) tenant_json = seed / "jiancai" / "tenant.json" data = json.loads(tenant_json.read_text(encoding="utf-8")) data["status"] = "disabled" tenant_json.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") a = build_app(seed, tmp_path / "data") with app_test_client(a, f"http://{JIANCAI_HOST}", tmp_path) as c: yield c def test_disabled_tenant_gets_pause_page(disabled_client): r = disabled_client.get("/api/chat/session") assert r.status_code == 503 assert "暂停" in r.text # 中性页: 不暴露租户 slug/域名 assert "jiancai" not in r.text.lower() assert JIANCAI_HOST not in r.text def test_disabled_tenant_ws_rejected_with_1013(disabled_client): with pytest.raises(WebSocketDisconnect) as ei: with disabled_client.websocket_connect("/ws/chat"): pass # pragma: no cover assert ei.value.code == 1013 # ---- 租户正确性(防串库红线) -------------------------------------------------- def test_default_host_routes_to_jiancai(app_client): # app_client 不带 host 头 → base_url 的 jiancai 域(既有全量测试的语义) r = app_client.post("/api/chat/session", json={}) assert r.status_code == 200 def test_ws_routes_to_correct_tenant_db(app_client, tmp_path): """WS 经 auto 域名连接 → 会话必须落在 auto 租户库。 三层断言: 1. 文件级——此刻只碰过 auto(WS 用显式 host;lifespan 清扫因无 db 文件而跳过),auto.db 已懒建而 jiancai.db 从未创建; 2. REST resume 经 auto 域名命中同一 sid(is_new=False 且回原 sid); 3. 同一 sid 经 jiancai 域名(base_url 默认)不可见——resume miss 落 is_new=True(新建了另一个会话),证明 jiancai 库里没有它。 (经 /api/chat/session 的 resume 探测而非 /messages: 后者只放行 app-source 会话,WS 造的 standalone 会话两库都会 404,区分不了; resume 探测不触发 AI 调用。)""" with app_client.websocket_connect( "/ws/chat", headers={"host": AUTO_HOST}) as ws: ws.send_text(json.dumps({"source": "standalone"})) init = ws.receive_json() assert init["type"] == "init" sid = init["session_id"] # 1. 文件级(jiancai 还没被任何请求 resolve 过) assert (tmp_path / "auto.db").exists() assert not (tmp_path / "jiancai.db").exists() # 2. auto 库命中 r_auto = app_client.post( "/api/chat/session", json={"session_id": sid}, headers={"host": AUTO_HOST}) assert r_auto.status_code == 200 assert r_auto.json()["is_new"] is False assert r_auto.json()["session_id"] == sid # 3. jiancai 库没有(miss → 新建,is_new=True、sid 不同) r_jc = app_client.post("/api/chat/session", json={"session_id": sid}) assert r_jc.status_code == 200 assert r_jc.json()["is_new"] is True assert r_jc.json()["session_id"] != sid # ---- 路径前缀模式(/t//,spec 2026-10-19 §3) ---------------------------- class TestPathPrefixRouting: """路径前缀与 Host 模式共存:优先级 平台豁免 → 路径前缀 → Host 解析。 Host 用未知值(NOPE_HOST)——路径命中即证明解析先于 Host;反向用已知 AUTO_HOST(Host 说是 auto 租户)+ /t/jiancai/ 断言路径优先(spec §2)。""" def test_聊天页_t_slug_根路径(self, app_client): r = app_client.get("/t/jiancai/", headers={"host": NOPE_HOST}) assert r.status_code == 200 assert "chat" in r.text.lower() def test_静态与公开API全链(self, app_client): r = app_client.get("/t/jiancai/static/widget.js", headers={"host": NOPE_HOST}) assert r.status_code == 200 r = app_client.get("/t/jiancai/api/public/brand", headers={"host": NOPE_HOST}) assert r.status_code == 200 assert r.json()["slug"] == "jiancai" def test_无尾斜杠不剥离_走Host解析404(self, app_client): # 只断言未知 Host 形态:/t/x 在 Host 模式下本就无路由(/ 才是聊天页), # 若带已知 Host 断言 200 会诱导实现去剥离无斜杠形态——违反 spec §3.2 r = app_client.get("/t/jiancai", headers={"host": NOPE_HOST}) assert r.status_code == 404 # 已知 Host 的正常根路径(Host 模式对照,不受路径模式影响) assert app_client.get("/", headers={"host": JIANCAI_HOST}).status_code == 200 def test_编码slug拒绝(self, app_client): # %61uto 解码后即合法 slug auto——解码形态判不出编码,raw_path 直判 # 为准(spec §3.1);不匹配 → 落 Host 解析(未知 Host)→ 404 r = app_client.get("/t/%61uto/", headers={"host": NOPE_HOST}) assert r.status_code == 404 def test_大写与点号拒绝(self, app_client): # 不能用 /t/../:httpx 客户端侧归一点段到不了中间件——用字面点号 assert app_client.get( "/t/Jiancai/", headers={"host": NOPE_HOST}).status_code == 404 assert app_client.get( "/t/jian.cai/", headers={"host": NOPE_HOST}).status_code == 404 def test_未知slug_404(self, app_client): r = app_client.get("/t/no-such/x", headers={"host": NOPE_HOST}) assert r.status_code == 404 def test_停用租户_503(self, disabled_client): # 路径模式同样吃 TenantDisabledError → 503 中性暂停页(与 Host 模式 # 同语义;此处 Host 头无关紧要,解析根本不走到 Host) r = disabled_client.get("/t/jiancai/api/chat/session", headers={"host": NOPE_HOST}) assert r.status_code == 503 assert "暂停" in r.text def test_site参数覆盖在路径模式生效(self, app_client): # jiancai 的 es 站点(空域名站点):默认取首站点 main,?site= 覆盖 # 链路既有(spec §3.4)——brand title 取站点覆盖值即证覆盖真生效 r = app_client.get("/t/jiancai/api/public/brand?site=es", headers={"host": NOPE_HOST}) assert r.status_code == 200 assert r.json()["title"] == "HESUPIPE Soporte" def test_平台豁免优先级不受影响(self, app_client): # /health 豁免不受路径模式影响;/t/x/api/platform/... 剥离后命中 # 平台路由属 by design(spec §3.5)——认证守护兜底,未带凭据 # 401/422/429(POST 空 body → 422) assert app_client.get("/health", headers={"host": NOPE_HOST}).status_code == 200 r = app_client.post("/t/jiancai/api/platform/auth/login", headers={"host": NOPE_HOST}) assert r.status_code in (401, 422, 429) async def test_orig_path与路径改写直接钉位(self, tmp_path): """直接构造 ASGI scope 断言改写(不经 httpx 归一化):capture app 记录中间件改写后的 path/raw_path/state["orig_path"]——spec §3.2/ §3.3 的直证。registry 手工装配,用完 dispose_all(Windows 文件锁)。""" reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path) captured = {} async def capture_app(scope, receive, send): state = scope.get("state") or {} captured.update( path=scope.get("path"), raw_path=scope.get("raw_path"), orig_path=state.get("orig_path"), tenant=(state.get("tenant").config.slug if state.get("tenant") else None), site=state.get("site").slug if state.get("site") else None, ) async def noop_receive(): return {"type": "http.request", "body": b"", "more_body": False} async def noop_send(message): pass def make_scope(rtype, path, raw_path): return { "type": rtype, "asgi": {"version": "3.0"}, "http_version": "1.1", "method": "GET", "scheme": "http", "path": path, "raw_path": raw_path, "query_string": b"", "headers": [(b"host", b"unknown.example")], "client": ("127.0.0.1", 12345), "app": SimpleNamespace(state=SimpleNamespace(registry=reg)), } mw = TenantHostMiddleware(capture_app) try: # http:基本剥离 + orig_path 留痕 + state 写入 await mw(make_scope("http", "/t/jiancai/static/widget.js", b"/t/jiancai/static/widget.js"), noop_receive, noop_send) assert captured["path"] == "/static/widget.js" assert captured["raw_path"] == b"/static/widget.js" assert captured["orig_path"] == "/t/jiancai/static/widget.js" assert captured["tenant"] == "jiancai" assert captured["site"] == "main" # raw_path 重编码而非长度截断(spec §3.2):%2F 解码成 "/" 后, # 截断会得 b"/a%2Fb",重算得 b"/a/b" await mw(make_scope("http", "/t/jiancai/a/b", b"/t/jiancai/a%2Fb"), noop_receive, noop_send) assert captured["path"] == "/a/b" assert captured["raw_path"] == b"/a/b" # 根路径形态:/t/jiancai/ → "/"(剥空兜底) await mw(make_scope("http", "/t/jiancai/", b"/t/jiancai/"), noop_receive, noop_send) assert captured["path"] == "/" assert captured["raw_path"] == b"/" # websocket scope 同构改写(不必真握手,capture 收到即证) await mw(make_scope("websocket", "/t/jiancai/ws/chat", b"/t/jiancai/ws/chat"), noop_receive, noop_send) assert captured["path"] == "/ws/chat" assert captured["orig_path"] == "/t/jiancai/ws/chat" finally: await reg.dispose_all() def test_WS路径模式握手(self, app_client): # WS 握手后服务端先等客户端 init 消息(chat.py:48)——先发 # standalone 再收 init;session_id 即证路径模式 WS 全链打通 with app_client.websocket_connect( "/t/jiancai/ws/chat", headers={"host": NOPE_HOST}) as ws: ws.send_text(json.dumps({"source": "standalone"})) data = ws.receive_json() assert data["type"] == "init" assert "session_id" in data def test_WS未知slug拒绝(self, app_client): with pytest.raises(WebSocketDisconnect) as ei: with app_client.websocket_connect( "/t/no-such/ws/chat", headers={"host": NOPE_HOST}): pass # pragma: no cover — 未 accept 即 close,进不来 assert ei.value.code == 1008 def test_Host模式回归_路径优先(self, app_client): # 已知 Host(AUTO_HOST 属 auto 租户)+ /t/jiancai/ → 出 jiancai # 而非 auto,证明路径前缀解析赢过 Host 解析(spec §2 优先级) r = app_client.get("/t/jiancai/api/public/brand", 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")