ai-kefu-admin/tests/test_admin_serving.py

# Phase 4 Task 1: admin 生产服务(/admin SPA)+ /health 分级。
#
# dist 的测试纪律: 夹具造最小假 dist(index.html + assets/app.js),经
# ADMIN_DIST 环境变量注入——不依赖也不污染仓库 admin/dist(pnpm build
# 前后测试行为一致);真 dist 的浏览器冒烟归验收阶段,不在单测内。
#
# 覆盖:
# - /admin/ 静态 SPA 服务(html=True 语义:目录→index.html);
# - SPA fallback 只吞 /admin 内的非 /admin/api 404;/api 未知路径保持
#   FastAPI 默认 JSON 404(不被 index.html 越界吞掉);
# - Host 中间件 /admin 豁免(未知 Host 可达);/admin-evil 仍走租户解析;
# - dist 缺失 → 503 说明页,应用其余功能照常(降级不炸);
# - /health 分级: ok / 懒建未探(engine_ok=None) / 单租户库损坏只
#   degraded 不整体 500。

import pytest
from sqlalchemy.ext.asyncio import async_sessionmaker

from app.platform.registry import make_tenant_engine
from tests.conftest import (
    AUTO_HOST,
    JIANCAI_HOST,
    NOPE_HOST,
    SEED_DIR,
    app_test_client,
    build_app,
)

ADMIN_INDEX_HTML = (
    "<!DOCTYPE html><html><head><title>admin</title>"
    '<script type="module" src="/admin/assets/app.js"></script>'
    "</head><body>admin-spa-marker</body></html>"
)


@pytest.fixture
def admin_client(tmp_path, monkeypatch):
    """app_client 变体: ADMIN_DIST 指向 tmp 假 dist。

    /admin 挂载对 dist 请求期现势解析(见 app.main.AdminSPA)——env 覆盖
    即换目录,测试与真实构建产物解耦。"""
    dist = tmp_path / "admin-dist"
    (dist / "assets").mkdir(parents=True)
    (dist / "index.html").write_text(ADMIN_INDEX_HTML, encoding="utf-8")
    (dist / "assets" / "app.js").write_text("// admin bundle", encoding="utf-8")
    monkeypatch.setenv("ADMIN_DIST", str(dist))
    a = build_app(SEED_DIR, tmp_path)
    with app_test_client(a, f"http://{JIANCAI_HOST}", tmp_path) as c:
        yield c


@pytest.fixture
def missing_dist_client(tmp_path, monkeypatch):
    """ADMIN_DIST 指向不存在路径——dist 缺失降级(503 说明页)的确定性
    夹具:不依赖仓库 admin/dist 的有无(冒烟构建后它会真实存在)。"""
    monkeypatch.setenv("ADMIN_DIST", str(tmp_path / "no-such-dist"))
    a = build_app(SEED_DIR, tmp_path)
    with app_test_client(a, f"http://{JIANCAI_HOST}", tmp_path) as c:
        yield c


# ---- /admin 静态服务 ---------------------------------------------------------


def test_admin_index_served(admin_client):
    # /admin(无斜杠)→ Mount 规范化重定向 → /admin/ 服务 index.html
    r = admin_client.get("/admin", follow_redirects=True)
    assert r.status_code == 200
    assert "admin-spa-marker" in r.text
    assert r.headers["content-type"].startswith("text/html")
    # 直连 /admin/ 同样 200
    r = admin_client.get("/admin/")
    assert r.status_code == 200
    assert "admin-spa-marker" in r.text


def test_admin_assets_served(admin_client):
    r = admin_client.get("/admin/assets/app.js")
    assert r.status_code == 200
    assert "admin bundle" in r.text


# ---- SPA fallback 边界 -------------------------------------------------------


def test_admin_fallback_only_admin(admin_client):
    # /admin 内未知路径 → index.html 200(hash 路由下的加固:直连/刷新深链
    # 不 404 白屏)
    r = admin_client.get("/admin/unknown/deep/link")
    assert r.status_code == 200
    assert "admin-spa-marker" in r.text
    # /admin/api 子前缀不吞:未来 admin 侧 API 的路由 404 必须按默认 JSON
    # 暴露,不被 index.html 掩盖
    r = admin_client.get("/admin/api/nope")
    assert r.status_code == 404
    assert r.headers["content-type"].startswith("application/json")
    assert r.json() == {"detail": "Not Found"}
    # 其余路径保持 FastAPI 默认 JSON 404——SPA fallback 不得越界
    r = admin_client.get("/api/definitely-not-a-route")
    assert r.status_code == 404
    assert r.headers["content-type"].startswith("application/json")
    assert r.json() == {"detail": "Not Found"}


# ---- Host 中间件豁免 ---------------------------------------------------------


def test_admin_exempt_from_tenant_middleware(admin_client):
    # 未知 Host 下 /admin 可达(不 404)——admin 是平台面,不依赖租户域名
    # 解析(与 /api/platform/ 前缀同一安全模型:豁免≠放行)
    r = admin_client.get("/admin/", headers={"host": NOPE_HOST})
    assert r.status_code == 200
    assert "admin-spa-marker" in r.text
    # 前缀精确到斜杠:/admin-evil 仍走租户解析 → 中间件 404(text/plain)
    r = admin_client.get("/admin-evil", headers={"host": NOPE_HOST})
    assert r.status_code == 404
    assert r.headers["content-type"].startswith("text/plain")


# ---- dist 缺失降级 ----------------------------------------------------------


def test_admin_dist_missing_degrades(missing_dist_client):
    r = missing_dist_client.get("/admin/")
    assert r.status_code == 503
    assert "未构建" in r.text
    assert "pnpm dev" in r.text
    # 降级不炸应用:健康检查与租户功能照常
    assert missing_dist_client.get("/health").status_code == 200
    assert missing_dist_client.post(
        "/api/chat/session", json={}).status_code == 200


# ---- /health 分级 -----------------------------------------------------------


def test_health_tiered_ok_and_lazy(app_client):
    r = app_client.get("/health")
    assert r.status_code == 200
    body = r.json()
    assert body["status"] == "ok"
    assert {t["slug"] for t in body["tenants"]} == {"jiancai", "auto"}
    # 懒建未就绪:engine_ok=None——健康检查不触发全租户建库
    assert all(t["engine_ok"] is None for t in body["tenants"])
    # 域名绑定数(jiancai 2 + auto 1,种子档案)
    assert body["domains"] == 3


@pytest.fixture
def corrupt_jiancai_client(tmp_path):
    """/health 分级的坏库夹具: jiancai 的 engine 直接指向损坏库文件。

    绕过 ensure_ready 的建库路径(那会因 ensure_schema 失败把 engine 留
    None),手工把垃圾字节写进 db_path 再挂 engine——模拟"运行中库文件
    损坏"的形态:engine 在,一探就炸。lifespan 启动清扫对该租户失败只记
    日志(main.py 既有行为),不影响本夹具。"""
    a = build_app(SEED_DIR, tmp_path)
    rt = next(r for r in a.state.registry.runtimes()
              if r.config.slug == "jiancai")
    rt.db_path.parent.mkdir(parents=True, exist_ok=True)
    rt.db_path.write_bytes(b"definitely not a sqlite database")
    rt.engine = make_tenant_engine(rt.db_path)
    rt.session_factory = async_sessionmaker(rt.engine, expire_on_commit=False)
    with app_test_client(a, f"http://{JIANCAI_HOST}", tmp_path) as c:
        yield c


def test_health_tiered_degraded_on_corrupt_db(corrupt_jiancai_client):
    c = corrupt_jiancai_client
    r = c.get("/health")
    assert r.status_code == 200          # 单租户损坏 → degraded,不整体 500
    body = r.json()
    assert body["status"] == "degraded"
    by_slug = {t["slug"]: t for t in body["tenants"]}
    assert by_slug["jiancai"]["engine_ok"] is False
    assert by_slug["auto"]["engine_ok"] is None
    # 其余租户不受牵连:auto 懒建好库正常服务
    r = c.post("/api/chat/session", json={}, headers={"host": AUTO_HOST})
    assert r.status_code == 200