浅念
153 lines
6.0 KiB
Python
153 lines
6.0 KiB
Python
# 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
|
|
|
|
import pytest
|
|
from starlette.websockets import WebSocketDisconnect
|
|
|
|
from app.main import app
|
|
from app.platform.host_middleware import TenantHostMiddleware
|
|
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
|