feat: 生产装配切换DbTenantSource + bootstrap脚本（超管/两租户幂等导入）
- app/main.py: lifespan registry 装配按 env TENANT_SOURCE 切换（db|file， 默认 db）——db 走 TenantRegistry(DbTenantSource(platform_engine)) + schema 就绪后 async load_all；file 回退旧 load_from_dir。空 platform.db 冷启动不炸：显式预检 tenants 表（非捕 ValueError——坏数据同样抛 ValueError，一揽子吞会把数据损坏静默成空平台），空表保持空注册表 （域名全 404 直到 bootstrap 导入），清扫循环对空 runtimes() 安全。 非法值在创建任何 engine 前显性炸。conftest 预置缝优先于 env 装配， 既有 558 测试零改动全绿。 - scripts/bootstrap_platform.py: init（schema+超管，env/.env 凭据，空表 才建）+ 出厂档案幂等导入（已存在 slug 跳过；FileTenantSource.read_tenant 同构快照 → Tenant+sites+domains+版本1，版本走 write_config_version 单一 路径；域名 _normalize_host 归一）。不建租户管理员——导入是存量迁移 形态，租户管理员按需走后台创建向导补。CLI 照 extract_config.py 的 argparse 风格（--db/--profiles/--seed）。 - tests/test_bootstrap_script.py: 幂等两遍零变化（行数/版本数逐项断言）、 超管 env 凭据、未知档案显性炸、bootstrap→DbTenantSource→registry 两租户 resolve 冒烟（jc 蓝/17 条/ai-kefu-jc，auto 橙/13 条/ai-kefu，es 覆盖层）、 TENANT_SOURCE 三态装配（db 空/库冷启动、file 回退、非法值炸）。 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
+43
-5
@@ -1,18 +1,23 @@
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
import logging
|
||||
import os
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.engine.config import STATIC_DIR, settings
|
||||
from app.platform.db import (
|
||||
Tenant,
|
||||
bootstrap_superuser,
|
||||
ensure_platform_schema,
|
||||
make_platform_engine,
|
||||
)
|
||||
from app.platform.host_middleware import TenantHostMiddleware
|
||||
from app.platform.registry import TenantRegistry
|
||||
from app.platform.tenant_source import DbTenantSource
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -25,22 +30,38 @@ PLATFORM_DB = BASE_DIR / "platform.db" # .gitignore 已忽略 platform.db*
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
# 测试注入缝:conftest 预置过 app.state.registry(临时 data_dir 的独立
|
||||
# Registry)则沿用,不再加载仓库 seed——测试租户库与生产互不污染
|
||||
# registry 装配(Task 11 生产切换):测试注入缝优先——conftest 预置过
|
||||
# app.state.registry 则沿用(既有测试全走预置,不受 env 默认切换影响),
|
||||
# 预置路径完全不读 TENANT_SOURCE。未预置才按 env 装配:db(默认,生产
|
||||
# 形态)DbTenantSource(platform.db);file 回退旧文件源(冷备/本地调试)。
|
||||
# 非法值在创建任何 engine 之前显性炸——拼错名静默跑错数据源,与"空源
|
||||
# 必须炸"同一纪律。
|
||||
registry = getattr(app.state, "registry", None)
|
||||
if registry is None:
|
||||
registry = TenantRegistry.load_from_dir(SEED_DIR, data_dir=TENANTS_DIR)
|
||||
app.state.registry = registry
|
||||
assemble_from_env = registry is None
|
||||
tenant_source = os.environ.get("TENANT_SOURCE", "db").strip().lower()
|
||||
if assemble_from_env and tenant_source not in ("db", "file"):
|
||||
raise RuntimeError(
|
||||
f"TENANT_SOURCE 只允许 db|file(实际 {tenant_source!r})")
|
||||
|
||||
# 平台库测试缝(Phase 2 Task 1,同 registry 缝模式):预置过
|
||||
# app.state.platform_engine(临时库)则沿用——不预置则每个 TestClient
|
||||
# 测试都会在仓库根建出真实 platform.db。schema+bootstrap 无条件跑
|
||||
# (两者幂等):预置引擎同样被建到可用,Task 2 起认证测试直接可用。
|
||||
# 装配次序(Task 11 起)engine 先于 registry:db 源 registry 构造即需
|
||||
# 持有 platform_engine。
|
||||
platform_engine = getattr(app.state, "platform_engine", None)
|
||||
if platform_engine is None:
|
||||
platform_engine = make_platform_engine(PLATFORM_DB)
|
||||
app.state.platform_engine = platform_engine
|
||||
|
||||
if assemble_from_env:
|
||||
if tenant_source == "db":
|
||||
registry = TenantRegistry(
|
||||
DbTenantSource(platform_engine), data_dir=TENANTS_DIR)
|
||||
else:
|
||||
registry = TenantRegistry.load_from_dir(SEED_DIR, data_dir=TENANTS_DIR)
|
||||
app.state.registry = registry
|
||||
|
||||
try:
|
||||
await ensure_platform_schema(platform_engine)
|
||||
await bootstrap_superuser(
|
||||
@@ -48,6 +69,23 @@ async def lifespan(app: FastAPI):
|
||||
username=settings.ADMIN_USERNAME,
|
||||
password=settings.ADMIN_PASSWORD)
|
||||
|
||||
# db 源初始加载(Task 11):schema 就绪后现势加载(空库无表,先建
|
||||
# schema 才能查)。空平台(platform.db 无租户——init 后未导入)不炸:
|
||||
# 保持空注册表启动(域名全 404 直到 bootstrap 导入;下方清扫循环对
|
||||
# runtimes() 空元组安全)。探空用显式预检而非捕 ValueError——
|
||||
# load_all 对坏数据(缺版本/坏 JSON/缺配置组)同样抛 ValueError,
|
||||
# 一揽子吞会把数据损坏静默成"空平台",必须区分。
|
||||
if assemble_from_env and tenant_source == "db":
|
||||
async with AsyncSession(platform_engine) as session:
|
||||
has_tenant = (await session.execute(
|
||||
select(Tenant.id).limit(1))).first() is not None
|
||||
if has_tenant:
|
||||
await registry.load_all()
|
||||
else:
|
||||
logger.warning(
|
||||
"platform.db 无租户——空注册表启动(域名全 404);"
|
||||
"跑 scripts/bootstrap_platform.py 导入出厂租户")
|
||||
|
||||
# 询盘推送(设计 8.2): 清扫崩溃遗留的 stale pending → failed。逐租户、
|
||||
# 用该租户自己的 session_factory;db 文件不存在的租户跳过——全新库
|
||||
# 无遗留行,保持 engine 懒建纪律(租户 engine 由 Host 中间件按请求
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
"""平台引导脚本(Phase 2 Task 11):生产装配(platform.db 源)的一键落地。
|
||||
|
||||
两步,全部幂等,可安全反复跑:
|
||||
1. init:make_platform_engine + ensure_platform_schema + bootstrap_superuser
|
||||
(凭据取 env/.env 的 ADMIN_USERNAME/ADMIN_PASSWORD,空表才建,与
|
||||
main.lifespan 同一 bootstrap——谁先跑结果一致);
|
||||
2. 导入:遍历出厂档案(--profiles,默认 auto jiancai)——
|
||||
FileTenantSource.read_tenant(slug) 产出 LoadedTenant(配置组⊕wall⊕
|
||||
prompts 同构快照,Task 4 起 registry 单一消费形态)→ tenants 表已有
|
||||
该 slug 则跳过 → 写 Tenant+sites+domains+配置版本 1(版本写入走
|
||||
db.write_config_version,与创建向导/配置 API 同一条 MAX+1 路径)。
|
||||
|
||||
**不建租户管理员**(计划裁定):导入是"存量迁移形态"——线上超管已由
|
||||
init 步骤引导,各租户管理员按需走后台创建向导(Task 5)补,脚本不造
|
||||
任何租户账号。
|
||||
|
||||
用法(Windows 须 PYTHONUTF8=1,输出含中文):
|
||||
PYTHONUTF8=1 python scripts/bootstrap_platform.py [--db PATH]
|
||||
[--profiles SLUG ...] [--seed DIR]
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# 仓库根入 sys.path(照 extract_config.py 锚定纪律:脚本可从任意 CWD 跑)
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.engine.config import settings
|
||||
from app.platform.db import (
|
||||
Site,
|
||||
SiteDomain,
|
||||
Tenant,
|
||||
bootstrap_superuser,
|
||||
ensure_platform_schema,
|
||||
make_platform_engine,
|
||||
write_config_version,
|
||||
)
|
||||
# 域名归一与路由层同一事实源(registry._normalize_host;tenant_routes 的
|
||||
# _normalize_domain 即它的 HTTP 包装)
|
||||
from app.platform.registry import _normalize_host
|
||||
from app.platform.tenant_source import FileTenantSource
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_PROFILES = ("auto", "jiancai")
|
||||
|
||||
|
||||
async def _import_profile(engine, seed_dir: Path, slug: str) -> str:
|
||||
"""单档案导入:已存在跳过(幂等),否则写 Tenant+sites+domains+版本1。
|
||||
|
||||
返回结果文案(导入明细/跳过原因);档案缺失 → LookupError 由调用方
|
||||
转 SystemExit(拼错档案名静默空平台 = 全量 404 的部署事故,必须显性炸)。
|
||||
"""
|
||||
loaded = FileTenantSource(seed_dir).read_tenant(slug) # LookupError 显性抛
|
||||
async with AsyncSession(engine, expire_on_commit=False) as session:
|
||||
if (await session.execute(
|
||||
select(Tenant.id).where(Tenant.slug == slug))).first() is not None:
|
||||
return f"{slug}: 跳过(tenants 已存在)"
|
||||
|
||||
# profile 列 = 档案名(重置出厂按它取对应档案组;导入的租户即其
|
||||
# 出厂档案的原生形态)
|
||||
tenant = Tenant(slug=loaded.slug, name=loaded.name,
|
||||
status=loaded.status, profile=slug)
|
||||
session.add(tenant)
|
||||
await session.flush()
|
||||
domain_count = 0
|
||||
for site_dict in loaded.sites:
|
||||
site = Site(tenant_id=tenant.id, slug=site_dict["slug"],
|
||||
name=site_dict.get("name") or site_dict["slug"],
|
||||
brand_overlay=dict(site_dict.get("brand") or {}))
|
||||
session.add(site)
|
||||
await session.flush()
|
||||
for domain in site_dict.get("domains") or []:
|
||||
normalized = _normalize_host(domain)
|
||||
if normalized: # 空串跳过(与域名索引同判)
|
||||
session.add(SiteDomain(domain=normalized, site_id=site.id))
|
||||
domain_count += 1
|
||||
version = await write_config_version(
|
||||
session, tenant.id, loaded.snapshot,
|
||||
updated_by="bootstrap", comment=f"bootstrap 导入: profile={slug}")
|
||||
assert version == 1 # 新租户无历史版本,必为 1
|
||||
await session.commit()
|
||||
return (f"{slug}: 导入(站点 {len(loaded.sites)},"
|
||||
f"域名 {domain_count},版本 {version})")
|
||||
|
||||
|
||||
async def run(args) -> None:
|
||||
db_path = Path(args.db)
|
||||
# sqlite 不建父目录(仓库根默认存在,--db 指新目录时会炸——conftest
|
||||
# build_app 同坑注记);照 registry._ensure_ready 纪律先建目录
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
engine = make_platform_engine(db_path)
|
||||
try:
|
||||
# ---- init(schema + 超管,均幂等)------------------------------------
|
||||
await ensure_platform_schema(engine)
|
||||
await bootstrap_superuser(
|
||||
engine,
|
||||
username=settings.ADMIN_USERNAME,
|
||||
password=settings.ADMIN_PASSWORD)
|
||||
print(f"platform.db: {db_path}")
|
||||
print(f"超管: {settings.ADMIN_USERNAME}(空表才建,已存在则不动)")
|
||||
|
||||
# ---- 导入(逐档案;缺档案显性炸)-------------------------------------
|
||||
seed_dir = Path(args.seed)
|
||||
results = []
|
||||
for slug in args.profiles:
|
||||
try:
|
||||
results.append(await _import_profile(engine, seed_dir, slug))
|
||||
except LookupError as exc:
|
||||
raise SystemExit(f"导入失败: 出厂档案缺失: {exc}") from exc
|
||||
for line in results:
|
||||
print(line)
|
||||
finally:
|
||||
await engine.dispose() # Windows 文件锁纪律
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument(
|
||||
"--db", default=str(REPO_ROOT / "platform.db"),
|
||||
help="platform.db 路径(默认仓库根,与 app/main.py 的 PLATFORM_DB "
|
||||
"同一锚定;测试/分环境可覆盖)")
|
||||
parser.add_argument(
|
||||
"--profiles", nargs="+", default=list(DEFAULT_PROFILES),
|
||||
help=f"导入的出厂档案 slug 列表(默认 {' '.join(DEFAULT_PROFILES)})")
|
||||
parser.add_argument(
|
||||
"--seed", default=str(REPO_ROOT / "seed_profiles"),
|
||||
help="出厂档案目录(默认仓库根 seed_profiles)")
|
||||
args = parser.parse_args()
|
||||
asyncio.run(run(args))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,249 @@
|
||||
"""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_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())
|
||||
Reference in New Issue
Block a user