浅念
343 lines
14 KiB
Python
343 lines
14 KiB
Python
"""TenantRegistry 单元契约(Task 5a 文件加载器 + 审查修复)。
|
|
|
|
seed_profiles/<slug>/tenant.json + 同目录 wall.json → Registry;
|
|
Host(小写/去端口)→站点→租户;首次 resolve 懒建 engine(WAL/pool=2)+
|
|
ensure_schema;reload 原子换条目并 dispose 旧 engine;停用租户
|
|
TenantDisabledError(503 语义);加载期校验显性炸。
|
|
|
|
Windows 纪律:每个建过 engine 的用例结尾 await reg.dispose_all()——
|
|
连接不关,文件锁着,tmp_path 清理 PermissionError。
|
|
"""
|
|
import asyncio
|
|
import dataclasses
|
|
import json
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import app.platform.registry as registry_mod
|
|
from app.platform.registry import (
|
|
TenantDisabledError,
|
|
TenantRegistry,
|
|
UnknownTenantError,
|
|
make_tenant_engine,
|
|
)
|
|
from app.platform.tenant_source import FileTenantSource
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
SEED_DIR = REPO_ROOT / "seed_profiles"
|
|
|
|
|
|
def _tenant(slug, domains=("a.example.com",), **extra):
|
|
"""最小合法 tenant.json(dict 形态);extra 覆盖顶层键。"""
|
|
tenant = {"slug": slug, "name": slug, "status": "active",
|
|
"sites": [{"slug": "main", "domains": list(domains)}],
|
|
"config": {}}
|
|
tenant.update(extra)
|
|
return tenant
|
|
|
|
|
|
def _write_seed(tmp_path, tenants):
|
|
"""tmp 下造最小 seed 目录:每租户目录 = 入参 tenant.json + 拷贝真实
|
|
jiancai wall.json 与 prompts/ 四件套(墙/提示词数据不重复维护,
|
|
compile_wall 与 PromptProvider.from_dir 直接可过)。"""
|
|
seed = tmp_path / "seed"
|
|
seed.mkdir()
|
|
for slug, tenant in tenants.items():
|
|
tdir = seed / slug
|
|
tdir.mkdir()
|
|
shutil.copyfile(SEED_DIR / "jiancai" / "wall.json", tdir / "wall.json")
|
|
shutil.copytree(SEED_DIR / "jiancai" / "prompts", tdir / "prompts")
|
|
(tdir / "tenant.json").write_text(
|
|
json.dumps(tenant, ensure_ascii=False), encoding="utf-8")
|
|
return seed
|
|
|
|
|
|
# ---- 基础解析 -------------------------------------------------------------
|
|
|
|
|
|
async def test_resolve_by_host(tmp_path):
|
|
reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path)
|
|
rt = await reg.resolve("chat2.agent.hnxinyouhe.com")
|
|
assert rt.config.slug == "jiancai"
|
|
assert rt.config.data["push"]["prefix"] == "ai-kefu-jc"
|
|
await reg.dispose_all()
|
|
|
|
|
|
async def test_unknown_host_404_semantics(tmp_path):
|
|
reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path)
|
|
with pytest.raises(UnknownTenantError):
|
|
await reg.resolve("nope.example.com")
|
|
await reg.dispose_all()
|
|
|
|
|
|
async def test_tenant_db_isolated_per_slug(tmp_path):
|
|
reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path)
|
|
a = await reg.resolve("chat2.agent.hnxinyouhe.com")
|
|
b = await reg.resolve("chat.agent.hnxinyouhe.com")
|
|
assert a.db_path != b.db_path and a.db_path.name == "jiancai.db"
|
|
await reg.dispose_all()
|
|
|
|
|
|
async def test_reload_swaps_config_atomically(tmp_path):
|
|
reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path)
|
|
rt = await reg.resolve("chat2.agent.hnxinyouhe.com")
|
|
await reg.registry_reload("jiancai") # 重新读盘
|
|
assert await reg.resolve("chat2.agent.hnxinyouhe.com") is not rt
|
|
assert rt.engine is None # 旧 runtime 的 engine 必须已 dispose(Windows 文件锁)
|
|
await reg.dispose_all()
|
|
|
|
|
|
async def test_site_brand_overlay_and_inheritance(tmp_path):
|
|
reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path)
|
|
rt, site = await reg.resolve_site("chat2.agent.hnxinyouhe.com")
|
|
assert site.slug == "main"
|
|
es = rt.site_by_slug("es") # 空域名站点按 slug 取
|
|
assert es.brand["title"] == "HESUPIPE Soporte" # 覆盖生效
|
|
assert es.brand["color"] == rt.config.data["brand"]["color"] # 未覆盖项继承租户默认
|
|
await reg.dispose_all()
|
|
|
|
|
|
async def test_host_normalization_case_and_port(tmp_path):
|
|
"""真实 Host 头带端口/可能大写(spec §4:Host 识别必须先归一)。"""
|
|
reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path)
|
|
rt = await reg.resolve("CHAT2.Agent.hNxinyouHe.com:8018")
|
|
assert rt.config.slug == "jiancai"
|
|
with pytest.raises(UnknownTenantError):
|
|
await reg.resolve("NOPE.example.com:8018")
|
|
await reg.dispose_all()
|
|
|
|
|
|
async def test_concurrent_first_resolve_builds_engine_once(tmp_path, monkeypatch):
|
|
"""两个协程同时首 resolve 同一租户:engine 只建一次(asyncio.Lock 串行)。"""
|
|
calls = []
|
|
real_make = registry_mod.make_tenant_engine
|
|
|
|
def counting_make(db_path):
|
|
calls.append(db_path)
|
|
return real_make(db_path)
|
|
|
|
monkeypatch.setattr(registry_mod, "make_tenant_engine", counting_make)
|
|
|
|
reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path)
|
|
rts = await asyncio.gather(
|
|
*[reg.resolve("chat2.agent.hnxinyouhe.com") for _ in range(5)])
|
|
assert len(calls) == 1 # 无锁则各协程各建一个,前 4 个泄漏
|
|
assert all(rt is rts[0] for rt in rts) # 同一 runtime 实例
|
|
assert rts[0].engine is not None and rts[0].session_factory is not None
|
|
await reg.dispose_all()
|
|
|
|
|
|
async def test_make_tenant_engine_enables_wal(tmp_path):
|
|
"""spec §6.1 硬纪律:每租户库打开即 PRAGMA journal_mode=WAL(读写不互斥)。"""
|
|
from sqlalchemy import text
|
|
|
|
engine = make_tenant_engine(tmp_path / "wal_check.db")
|
|
try:
|
|
async with engine.connect() as conn:
|
|
mode = (await conn.execute(text("PRAGMA journal_mode"))).scalar()
|
|
assert str(mode).lower() == "wal"
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
# ---- 加载期校验(I-4) ------------------------------------------------------
|
|
|
|
|
|
def test_load_empty_seed_dir_raises(tmp_path):
|
|
seed = tmp_path / "seed"
|
|
seed.mkdir()
|
|
with pytest.raises(ValueError, match="没有任何"):
|
|
TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
|
|
|
|
|
|
def test_load_invalid_json_raises_with_path(tmp_path):
|
|
"""坏 JSON 必须带文件路径显性炸(哪份文件坏了一目了然)。"""
|
|
seed = tmp_path / "seed"
|
|
(seed / "jiancai").mkdir(parents=True)
|
|
(seed / "jiancai" / "tenant.json").write_text("{oops", encoding="utf-8")
|
|
with pytest.raises(ValueError, match=r"tenant\.json.*无效 JSON"):
|
|
TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
|
|
|
|
|
|
def test_load_slug_dir_mismatch_raises(tmp_path):
|
|
seed = _write_seed(tmp_path, {"dirslug": _tenant("jsonslug")})
|
|
with pytest.raises(ValueError, match="不一致"):
|
|
TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
|
|
|
|
|
|
def test_load_zero_sites_raises(tmp_path):
|
|
tenant = _tenant("jiancai")
|
|
tenant["sites"] = []
|
|
seed = _write_seed(tmp_path, {"jiancai": tenant})
|
|
with pytest.raises(ValueError, match="至少一个站点"):
|
|
TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
|
|
|
|
|
|
def test_load_domain_conflict_across_tenants_raises(tmp_path):
|
|
seed = _write_seed(tmp_path, {
|
|
"t1": _tenant("t1", ["dup.example.com"]),
|
|
"t2": _tenant("t2", ["dup.example.com"])})
|
|
with pytest.raises(ValueError, match="dup.example.com"):
|
|
TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
|
|
|
|
|
|
def test_load_domain_conflict_within_tenant_raises(tmp_path):
|
|
"""同租户两个站点抢同一域名也要炸,报错含双方 (tenant, site)。"""
|
|
tenant = _tenant("t1", ["dup.example.com"])
|
|
tenant["sites"].append({"slug": "second", "domains": ["dup.example.com"]})
|
|
seed = _write_seed(tmp_path, {"t1": tenant})
|
|
with pytest.raises(ValueError) as excinfo:
|
|
TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
|
|
msg = str(excinfo.value)
|
|
assert "main" in msg and "second" in msg and "t1" in msg
|
|
|
|
|
|
# ---- reload / dispose 行为 -------------------------------------------------
|
|
|
|
|
|
async def test_reload_bad_config_keeps_old_entry_serving(tmp_path):
|
|
"""reload 遇坏配置必须先炸后不动:旧条目(含域名索引)原样服务。"""
|
|
tenant = _tenant("jiancai", ["a.example.com"])
|
|
tenant["config"] = {"push": {"prefix": "v1"}}
|
|
seed = _write_seed(tmp_path, {"jiancai": tenant})
|
|
reg = TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
|
|
rt1 = await reg.resolve("a.example.com")
|
|
assert rt1.config.data["push"]["prefix"] == "v1"
|
|
|
|
tenant["sites"] = [] # 改坏:零站点
|
|
(seed / "jiancai" / "tenant.json").write_text(
|
|
json.dumps(tenant, ensure_ascii=False), encoding="utf-8")
|
|
with pytest.raises(ValueError, match="至少一个站点"):
|
|
await reg.registry_reload("jiancai")
|
|
|
|
rt = await reg.resolve("a.example.com") # 旧条目继续服务
|
|
assert rt is rt1 and rt.config.data["push"]["prefix"] == "v1"
|
|
await reg.dispose_all()
|
|
|
|
|
|
async def test_dispose_all_then_resolve_rebuilds(tmp_path):
|
|
"""dispose_all 复位为未就绪,再 resolve 重新懒建(可安全复用)。"""
|
|
reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path)
|
|
rt = await reg.resolve("chat2.agent.hnxinyouhe.com")
|
|
engine_before = rt.engine
|
|
await reg.dispose_all()
|
|
assert rt.engine is None
|
|
rt2 = await reg.resolve("chat2.agent.hnxinyouhe.com")
|
|
assert rt2 is rt
|
|
assert rt2.engine is not None and rt2.engine is not engine_before
|
|
await reg.dispose_all()
|
|
|
|
|
|
async def test_resolve_waiting_through_reload_serves_new_entry(tmp_path):
|
|
"""I-2 竞态:resolve 排队等锁期间条目被 reload 换掉——持锁后现势
|
|
检查必须放弃旧 runtime 按新条目建库(旧 runtime 永不产生孤儿 engine)。
|
|
|
|
构造:测试占住 per-slug 锁 → reload 与 resolve 先后排到锁等待
|
|
(FIFO:reload 先拿锁换条目)→ 释放后 resolve 持锁发现条目已换。
|
|
"""
|
|
reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path)
|
|
rt_old = reg._by_slug["jiancai"] # 单元测试:直查内部状态
|
|
lock = reg._locks.setdefault("jiancai", asyncio.Lock())
|
|
async with lock:
|
|
reloader = asyncio.create_task(reg.registry_reload("jiancai"))
|
|
resolver = asyncio.create_task(reg.resolve("chat2.agent.hnxinyouhe.com"))
|
|
await asyncio.sleep(0) # 两任务均跑到锁等待
|
|
await asyncio.sleep(0)
|
|
await asyncio.gather(reloader, resolver)
|
|
|
|
rt = resolver.result()
|
|
assert rt is reg._by_slug["jiancai"] # 服务 reload 后的新条目
|
|
assert rt is not rt_old
|
|
assert rt_old.engine is None # 旧 runtime 没被建库(无孤儿句柄)
|
|
assert rt.engine is not None
|
|
await reg.dispose_all()
|
|
|
|
|
|
# ---- 停用租户(I-3) ---------------------------------------------------------
|
|
|
|
|
|
async def test_disabled_tenant_resolve_raises_and_never_builds_engine(
|
|
tmp_path, monkeypatch):
|
|
"""status=disabled:resolve 抛 TenantDisabledError(503 语义,非 404),
|
|
且从不建 engine。"""
|
|
calls = []
|
|
real_make = registry_mod.make_tenant_engine
|
|
|
|
def counting_make(db_path):
|
|
calls.append(db_path)
|
|
return real_make(db_path)
|
|
|
|
monkeypatch.setattr(registry_mod, "make_tenant_engine", counting_make)
|
|
|
|
seed = _write_seed(tmp_path, {"jiancai": _tenant("jiancai", status="disabled")})
|
|
reg = TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
|
|
with pytest.raises(TenantDisabledError):
|
|
await reg.resolve("a.example.com")
|
|
with pytest.raises(TenantDisabledError): # 再 resolve 同样拒绝
|
|
await reg.resolve("a.example.com")
|
|
assert calls == [] # 从未建 engine
|
|
await reg.dispose_all()
|
|
|
|
|
|
async def test_reload_to_disabled_releases_engine(tmp_path):
|
|
"""active→disabled 热切换:旧 engine 随 reload dispose,resolve 转抛。"""
|
|
tenant = _tenant("jiancai", ["a.example.com"])
|
|
seed = _write_seed(tmp_path, {"jiancai": tenant})
|
|
reg = TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
|
|
rt1 = await reg.resolve("a.example.com")
|
|
assert rt1.engine is not None
|
|
|
|
tenant["status"] = "disabled"
|
|
(seed / "jiancai" / "tenant.json").write_text(
|
|
json.dumps(tenant, ensure_ascii=False), encoding="utf-8")
|
|
await reg.registry_reload("jiancai")
|
|
|
|
assert rt1.engine is None # 旧 engine 已释放
|
|
with pytest.raises(TenantDisabledError):
|
|
await reg.resolve("a.example.com")
|
|
await reg.dispose_all()
|
|
|
|
|
|
# ---- 路径前缀模式解析(spec 2026-10-19 §5) ----------------------------------
|
|
|
|
|
|
class TestResolveBySlug:
|
|
"""resolve_by_slug:/t/<slug>/ 路径模式的 slug 直查——语义与 resolve_site
|
|
同款(未知/停用/懒建/reload 竞态取 _ensure_ready 返回值),首站点为默认。"""
|
|
|
|
async def test_已知_slug_返回租户与首站点(self, tmp_path):
|
|
reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path)
|
|
rt, site = await reg.resolve_by_slug("jiancai")
|
|
assert rt.config.slug == "jiancai"
|
|
assert site.slug == rt.config.sites[0].slug # 首站点为默认
|
|
await reg.dispose_all()
|
|
|
|
async def test_未知_slug_UnknownTenantError(self, tmp_path):
|
|
reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path)
|
|
with pytest.raises(UnknownTenantError):
|
|
await reg.resolve_by_slug("no-such")
|
|
await reg.dispose_all()
|
|
|
|
async def test_停用租户_TenantDisabledError(self, tmp_path):
|
|
seed = _write_seed(
|
|
tmp_path, {"jiancai": _tenant("jiancai", status="disabled")})
|
|
reg = TenantRegistry.load_from_dir(seed, data_dir=tmp_path / "data")
|
|
with pytest.raises(TenantDisabledError):
|
|
await reg.resolve_by_slug("jiancai")
|
|
await reg.dispose_all()
|
|
|
|
async def test_空站点表按未知处理(self, tmp_path):
|
|
"""防御分支(spec §5):两源加载期均拒零站点,直装 sites=[] 的
|
|
LoadedTenant 绕过源校验覆盖该分支——正常运行不可达,防将来源
|
|
语义漂移。"""
|
|
seed = _write_seed(tmp_path, {"jiancai": _tenant("jiancai")})
|
|
source = FileTenantSource(seed)
|
|
lt = dataclasses.replace(source.read_tenant("jiancai"), sites=[])
|
|
reg = TenantRegistry(source, data_dir=tmp_path / "data")
|
|
reg._install([lt]) # 手工装配(单元测试:直装内部)
|
|
with pytest.raises(UnknownTenantError):
|
|
await reg.resolve_by_slug("jiancai")
|
|
await reg.dispose_all()
|