ai-kefu-admin/tests/test_bootstrap_script.py

"""bootstrap 脚本 + 生产装配切换(Phase 2 Task 11)。

两块契约:
1. scripts/bootstrap_platform.py(照 extract_config.py 的 argparse 风格):
   init(schema+超管,env 凭据)+ 出厂档案幂等导入——跑两遍第二遍零变化
   (DB 行数/版本数不变);导入不建租户管理员(存量迁移形态,计划裁定)。
2. app/main.py 的 TENANT_SOURCE 装配切换:默认 db(DbTenantSource,空
   platform.db 冷启动不炸——空注册表,域名全 404 直到 bootstrap 导入);
   file 回退旧文件源;非法值启动显性炸。conftest 的 registry 预置缝
   优先于 env 装配(既有 558 测试不受默认切换影响)。

脚本测试方式:import + sys.argv 注入(非 subprocess)——同一 main()/
argparse 代码路径确定性执行,免每脚本次冷启 Python+重导 app 包的秒级
开销(幂等测试要连跑多遍);脚本唯一的对外副作用是 DB 文件与 stdout,
两者都被直接断言,真实性不靠进程边界背书。
"""
import asyncio
import importlib.util
import sys
from pathlib import Path

import pytest
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.engine.config import settings
from app.platform.auth import verify_password
from app.platform.db import (
    AdminUser,
    Site,
    SiteDomain,
    Tenant,
    TenantConfigVersion,
    list_admin_users,
    make_platform_engine,
)
from app.platform.registry import TenantRegistry
from app.platform.tenant_source import DbTenantSource
from tests.conftest import AUTO_HOST, JIANCAI_HOST, app_test_client

REPO_ROOT = Path(__file__).resolve().parents[1]


def _load_script():
    spec = importlib.util.spec_from_file_location(
        "bootstrap_platform", REPO_ROOT / "scripts" / "bootstrap_platform.py")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


# 模块级加载一次(脚本 import 期只做 sys.path 锚定,无副作用)
bootstrap_platform = _load_script()


def run_script(monkeypatch, *argv):
    monkeypatch.setattr(sys, "argv", ["bootstrap_platform.py", *argv])
    bootstrap_platform.main()


# ---- 脚本:幂等导入 -----------------------------------------------------------


async def _db_state(db_path: Path) -> tuple[dict, dict]:
    """(四表行数 + 超管数, 租户→当前版本号)——幂等断言的快照形态。"""
    engine = make_platform_engine(db_path)
    try:
        async with AsyncSession(engine) as session:
            counts = {}
            for name, model in (
                    ("tenants", Tenant), ("sites", Site),
                    ("site_domains", SiteDomain),
                    ("versions", TenantConfigVersion),
                    ("admins", AdminUser)):
                counts[name] = (await session.execute(
                    select(func.count()).select_from(model))).scalar()
            versions = dict((await session.execute(
                select(TenantConfigVersion.tenant_id,
                       func.max(TenantConfigVersion.version))
                .group_by(TenantConfigVersion.tenant_id))).all())
        return counts, versions
    finally:
        await engine.dispose()


def test_bootstrap_idempotent_two_runs_zero_change(tmp_path, monkeypatch, capsys):
    """跑两遍:第一遍导入两租户+版本1+超管;第二遍全部跳过——行数/版本数
    逐项不变(幂等不是"不炸",是零变化)。"""
    db = tmp_path / "platform.db"
    run_script(monkeypatch, "--db", str(db))
    first_out = capsys.readouterr().out
    assert "导入" in first_out and "jiancai" in first_out and "auto" in first_out

    counts1, versions1 = asyncio.run(_db_state(db))
    assert counts1 == {"tenants": 2, "sites": 3,      # jiancai main+es / auto main
                       "site_domains": 3,             # jc 2 + auto 1
                       "versions": 2, "admins": 1}
    assert all(v == 1 for v in versions1.values())    # 版本 1,各一行

    run_script(monkeypatch, "--db", str(db))
    second_out = capsys.readouterr().out
    assert "跳过" in second_out

    counts2, versions2 = asyncio.run(_db_state(db))
    assert counts2 == counts1                         # 第二遍零变化
    assert versions2 == versions1


def test_bootstrap_superuser_uses_settings_credentials(tmp_path, monkeypatch):
    """init 的超管凭据来自 settings(env/.env)——非硬编码;空表才建 +
    must_change_password=True(与 lifespan 同一语义)。"""
    monkeypatch.setattr(settings, "ADMIN_USERNAME", "bootadmin")
    monkeypatch.setattr(settings, "ADMIN_PASSWORD", "s3cret-pw-9")
    db = tmp_path / "platform.db"
    run_script(monkeypatch, "--db", str(db))

    async def probe():
        engine = make_platform_engine(db)
        try:
            return await list_admin_users(engine)
        finally:
            await engine.dispose()

    users = asyncio.run(probe())
    assert [u.username for u in users] == ["bootadmin"]
    assert verify_password("s3cret-pw-9", users[0].password_hash, users[0].salt)
    assert users[0].must_change_password is True


def test_bootstrap_unknown_profile_fails_loudly(tmp_path, monkeypatch):
    """--profiles 给不存在的档案 → 显性 SystemExit 带档案名(不静默跳过
    ——拼错名静默空平台 = 全量 404 的部署事故)。"""
    with pytest.raises(SystemExit, match="ghost"):
        run_script(monkeypatch, "--db", str(tmp_path / "p.db"),
                   "--profiles", "ghost")


# ---- 脚本产物 → db 源启动冒烟 ---------------------------------------------------


def test_bootstrap_then_db_source_registry_serves_both_tenants(tmp_path, monkeypatch):
    """bootstrap → DbTenantSource → TenantRegistry(async load_all)→ 两租户
    域名 resolve 全就位。抽查明细(计划指定):jiancai 品牌蓝/墙 17 条/
    prefix ai-kefu-jc,auto 橙/13 条/ai-kefu,es 站点覆盖层在位。"""
    db = tmp_path / "platform.db"
    run_script(monkeypatch, "--db", str(db))

    async def scenario():
        engine = make_platform_engine(db)
        try:
            reg = TenantRegistry(
                DbTenantSource(engine), data_dir=tmp_path / "tenants")
            await reg.load_all()

            jc_rt, jc_site = await reg.resolve_site(JIANCAI_HOST)
            assert jc_rt.config.slug == "jiancai" and jc_site.slug == "main"
            assert jc_rt.config.data["brand"]["color"] == "#066fc0"   # 品牌蓝
            assert len(jc_rt.compiled_wall.price_patterns) == 17
            assert jc_rt.config.data["push"]["prefix"] == "ai-kefu-jc"
            es = jc_rt.site_by_slug("es")                 # 站点覆盖层随导入落位
            assert es.brand["title"] == "HESUPIPE Soporte"
            assert es.brand["color"] == "#066fc0"          # 未覆盖项继承租户级

            auto_rt, auto_site = await reg.resolve_site(AUTO_HOST)
            assert auto_rt.config.slug == "auto" and auto_site.slug == "main"
            assert auto_rt.config.data["brand"]["color"] == "#f08a31"  # 橙
            assert len(auto_rt.compiled_wall.price_patterns) == 13
            assert auto_rt.config.data["push"]["prefix"] == "ai-kefu"

            await reg.dispose_all()                       # Windows 文件锁纪律
        finally:
            await engine.dispose()

    asyncio.run(scenario())


# ---- main.py 装配切换(TENANT_SOURCE)--------------------------------------------


def _patch_main_paths(monkeypatch, tmp_path):
    """main 模块的 PLATFORM_DB/TENANTS_DIR 锚到 tmp——默认装配路径在仓库根
    建真实 platform.db/租户库,测试绝不许写仓库目录。"""
    import app.main as main_mod
    monkeypatch.setattr(main_mod, "PLATFORM_DB", tmp_path / "platform.db")
    monkeypatch.setattr(main_mod, "TENANTS_DIR", tmp_path / "tenants")
    return main_mod


def test_main_default_db_source_empty_platform_boots(tmp_path, monkeypatch):
    """TENANT_SOURCE 未设(默认 db)+ 空 platform.db:启动不炸——空注册表
    (域名全 404 直到 bootstrap 导入),/health 豁免 200,清扫循环对空
    runtimes() 安全。生产冷启动形态(init 后、导入前)。"""
    from app.main import app
    _patch_main_paths(monkeypatch, tmp_path)
    monkeypatch.delenv("TENANT_SOURCE", raising=False)

    with app_test_client(app, "http://testserver", tmp_path) as c:
        assert c.get("/health").status_code == 200
        assert c.get("/", headers={"host": "nope.example.com"}).status_code == 404
        assert app.state.registry.runtimes() == ()       # 空注册表不炸


def test_platform_db_defaults_into_data_dir(tmp_path, monkeypatch):
    """PLATFORM_DB 落 <repo>/data/(P4T1 审查修复):data/ 是 compose 的
    bind mount——容器/镜像重建不再丢租户注册表。两层断言:
    1. 常量形状钉死(回退仓库根即红,compose 持久化前提);
    2. 功能:锚到 tmp 的 data/ 子目录(目录刻意不预建=冷机形态),
       lifespan 启动后平台库文件确实建在 data/ 内(父目录兜底 mkdir)。"""
    import app.main as main_mod
    assert main_mod.PLATFORM_DB == main_mod.BASE_DIR / "data" / "platform.db"
    target = tmp_path / "data" / "platform.db"
    assert not target.parent.exists()          # 冷机:data/ 尚不存在
    monkeypatch.setattr(main_mod, "PLATFORM_DB", target)
    monkeypatch.setattr(main_mod, "TENANTS_DIR", tmp_path / "tenants")
    monkeypatch.delenv("TENANT_SOURCE", raising=False)

    with app_test_client(main_mod.app, "http://testserver", tmp_path) as c:
        assert c.get("/health").status_code == 200
        # 断言须在 client 存活期内:app_test_client 退出时会清掉整个 tmp
        # (含本文件)——文件在应用运行中确实存在即为所证
        assert target.exists()                 # lifespan 建库于 data/ 内


def test_main_default_db_source_bootstrapped_platform(tmp_path, monkeypatch):
    """默认 db + bootstrap 导入后的 platform.db:启动即装配两租户(registry
    经 async load_all),jiancai 域名真实请求走通(Host 中间件→懒建租户库
    于 tmp)——生产正形态端到端。"""
    from app.main import app
    run_script(monkeypatch, "--db", str(tmp_path / "platform.db"))
    _patch_main_paths(monkeypatch, tmp_path)
    monkeypatch.delenv("TENANT_SOURCE", raising=False)

    with app_test_client(app, f"http://{JIANCAI_HOST}", tmp_path) as c:
        assert c.get("/health").status_code == 200
        assert {rt.config.slug for rt in app.state.registry.runtimes()} == {
            "jiancai", "auto"}
        assert c.get("/").status_code == 200            # 租户域名解析 + chat.html


def test_main_tenant_source_file_fallback(tmp_path, monkeypatch):
    """TENANT_SOURCE=file 回退路径仍工作:registry 走 load_from_dir(仓库
    seed_profiles),两租户就位。"""
    from app.main import app
    _patch_main_paths(monkeypatch, tmp_path)
    monkeypatch.setenv("TENANT_SOURCE", "file")

    with app_test_client(app, f"http://{JIANCAI_HOST}", tmp_path) as c:
        assert c.get("/health").status_code == 200
        assert {rt.config.slug for rt in app.state.registry.runtimes()} == {
            "jiancai", "auto"}
        assert c.get("/").status_code == 200


def test_main_tenant_source_invalid_raises(tmp_path, monkeypatch):
    """TENANT_SOURCE 非法值 → 启动显性炸(不静默回退默认——拼错名静默
    跑错数据源与"空源必须炸"同一纪律)。"""
    from fastapi.testclient import TestClient
    from app.main import app
    _patch_main_paths(monkeypatch, tmp_path)
    monkeypatch.setenv("TENANT_SOURCE", "bogus")

    with pytest.raises(RuntimeError, match="TENANT_SOURCE"):
        with TestClient(app):
            pass
    # 启动半途失败的状态复位(engine 可能已挂 state,不留脏缝给后续测试)
    app.state.registry = None
    if app.state.platform_engine is not None:
        engine = app.state.platform_engine
        app.state.platform_engine = None
        asyncio.run(engine.dispose())