From 0c296ca1ba92a1136d57f5ddc27766cb49c2e582 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Mon, 12 Oct 2026 18:54:30 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20Registry=20=E5=AE=A1=E6=9F=A5=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=EF=BC=88JSON=E5=AE=9A=E4=BD=8D/=E6=A0=A1=E9=AA=8C?= =?UTF-8?q?=E6=B5=8B=E8=AF=95/=E7=AB=9E=E6=80=81=E7=AA=97=E5=8F=A3/TenantD?= =?UTF-8?q?isabledError/=E8=B7=AF=E5=BE=84=E9=94=9A=E5=AE=9A/seed=5Fdir?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- app/platform/registry.py | 161 ++++++++++++++++++++----------- seed_profiles/README.md | 30 ++++++ tests/test_registry.py | 199 ++++++++++++++++++++++++++++++++++++++- 3 files changed, 332 insertions(+), 58 deletions(-) create mode 100644 seed_profiles/README.md diff --git a/app/platform/registry.py b/app/platform/registry.py index 93aaed7..746711f 100644 --- a/app/platform/registry.py +++ b/app/platform/registry.py @@ -7,12 +7,16 @@ platform.db 时本模块的对外接口不变(spec §5.3:后台保存→原子 关键行为: - resolve_site(host):host 小写去端口 → 域名索引 → (TenantRuntime, Site); - 未知 Host → UnknownTenantError(spec §4:404,不做兜底租户)。 + 未知 Host → UnknownTenantError(spec §4:404,不做兜底租户);租户停用 → + TenantDisabledError(Task 7 中间件映射为 503 中性"服务暂停"页,非 404)。 resolve(host) 只是它的便捷形式——双 API 一层实现,防两套解析漂移。 - 懒建:首次 resolve 才建 engine(WAL/pool=2)+ session_factory + ensure_schema, - per-slug asyncio.Lock 串行,双检防并发双建;建库失败不留半开 engine。 -- registry_reload(slug):重读磁盘→校验(坏 JSON/墙/域名冲突全在替换前抛, - 旧条目不动)→原子替换 _by_slug 条目并重建域名索引→dispose 旧 engine + per-slug asyncio.Lock 串行,锁内双检防并发双建;持锁后现势检查——条目 + 已被 reload 换掉则放弃本次建库、按当前条目重走(旧 runtime 永不产生 + 孤儿 engine);建库失败 dispose 半开 engine,保持未就绪下次重试。 +- registry_reload(slug):磁盘读取/校验在锁外,swap+dispose 段与懒建/ + dispose_all 取同一把 per-slug 锁。坏配置(无效 JSON/墙编译失败/零站点/ + 域名冲突)在替换前抛,旧条目原样服务;替换后 dispose 旧 engine (Windows 下不关=库文件句柄泄漏)。 测试/teardown 纪律:凡建过 engine 的 Registry 用完必须 await dispose_all() @@ -40,6 +44,24 @@ class UnknownTenantError(LookupError): """Host/域名无法映射到任何租户(调用方语义:聊天/公开端点 404)。""" +class TenantDisabledError(LookupError): + """租户 status != "active"(调用方语义:503 中性"服务暂停"页,非 404; + Task 7 中间件负责映射)。""" + + +def _read_json(path: pathlib.Path): + """读 JSON 文件;读不了/解析不了都带路径抛 ValueError——哪份文件 + 坏了一目了然(裸 FileNotFoundError/JSONDecodeError 不含路径上下文)。""" + try: + text = path.read_text(encoding="utf-8") + except OSError as exc: + raise ValueError(f"{path}: 无法读取: {exc}") from exc + try: + return json.loads(text) + except ValueError as exc: # JSONDecodeError 等 + raise ValueError(f"{path}: 无效 JSON: {exc}") from exc + + def make_tenant_engine(db_path: pathlib.Path) -> AsyncEngine: """每租户 SQLite engine:pool=2 无溢出(单租户低并发),忙等 30s (对齐 jc 现状:后台推送连接可能撞 SMTP 窗口写锁),打开即 WAL @@ -67,12 +89,18 @@ def _normalize_host(host: str) -> str: @dataclass class TenantRuntime: - """单租户运行时。config/db_path/compiled_wall 加载期就绪;engine/ - session_factory 由 Registry 首次 resolve 懒建;manager 由装配线注入 - (Task 5b),prompts 由 Task 6 注入——本任务不含。""" + """单租户运行时。 + + 加载期就绪:config(TenantConfig:data=快照⊕出厂默认,含 sites)/ + db_path(该租户 SQLite 文件路径)/seed_dir(seed_profiles//, + Task 6 PromptProvider 从中读 prompts/*.md)/compiled_wall(墙编译产物)。 + 懒建:engine/session_factory(Registry 首次 resolve,ensure_schema 已跑)。 + manager(Task 5b)由装配线另行构造注入,不在本类持有。 + """ config: TenantConfig db_path: pathlib.Path + seed_dir: pathlib.Path compiled_wall: CompiledWall engine: AsyncEngine | None = None session_factory: async_sessionmaker | None = None @@ -103,8 +131,8 @@ class TenantRegistry: """seed_profiles//tenant.json(+同目录 wall.json)→ Registry。 加载期显性失败(不流入运行期):目录无任何 tenant.json(seed_dir - 路径拼错→静默空注册表→全量 404,必须炸)、slug 与目录名不一致、 - 零站点、wall 编译错误、域名重复绑定。 + 路径拼错→静默空注册表→全量 404,必须炸)、无效 JSON、slug 与目录 + 名不一致、零站点、wall 编译错误、域名重复绑定。 """ reg = cls() reg._seed_dir = pathlib.Path(seed_dir) @@ -122,8 +150,7 @@ class TenantRegistry: def _load_tenant(self, tenant_dir: pathlib.Path) -> TenantRuntime: slug = tenant_dir.name - tenant_json = json.loads( - (tenant_dir / "tenant.json").read_text(encoding="utf-8")) + tenant_json = _read_json(tenant_dir / "tenant.json") if tenant_json.get("slug") != slug: raise ValueError( f"{tenant_dir / 'tenant.json'}: slug 字段 " @@ -133,8 +160,7 @@ class TenantRegistry: raise ValueError(f"{slug}: 每租户至少一个站点(sites 为空)") # wall 并入 snapshot["wall"]:data["wall"] 保留原始词表(后台编辑/ # 回滚用),编译产物挂 runtime.compiled_wall(引擎消费,单一编译) - wall_cfg = json.loads( - (tenant_dir / "wall.json").read_text(encoding="utf-8")) + wall_cfg = _read_json(tenant_dir / "wall.json") snapshot = dict(tenant_json.get("config") or {}) snapshot["wall"] = wall_cfg config = TenantConfig.from_snapshot( @@ -143,6 +169,7 @@ class TenantRegistry: return TenantRuntime( config=config, db_path=self._data_dir / f"{slug}.db", + seed_dir=tenant_dir, compiled_wall=compile_wall(wall_cfg)) def _build_domain_index( @@ -155,27 +182,34 @@ class TenantRegistry: key = _normalize_host(domain) if not key: continue - if index.get(key, (slug, site.slug)) != (slug, site.slug): + if key in index and index[key] != (slug, site.slug): raise ValueError( - f"域名 {key} 同时绑定 {index[key][0]} 与 " - f"{slug} 的站点(域名不可重复绑定)") + f"域名 {key} 已绑定 (tenant={index[key][0]}, " + f"site={index[key][1]}),不能同时绑定 " + f"(tenant={slug}, site={site.slug})") index[key] = (slug, site.slug) return index # ---- 解析(对外 API)-------------------------------------------------- async def resolve_site(self, host: str) -> tuple[TenantRuntime, Site]: - """host → (TenantRuntime, Site)。首次 resolve 懒建该租户 engine。""" + """host → (TenantRuntime, Site)。未知 Host → UnknownTenantError; + 租户停用 → TenantDisabledError;首次 resolve 懒建该租户 engine。 + 返回的 runtime 以 _ensure_ready 的返回值为准(等锁期间条目可能 + 已被 reload 换成新配置)。""" mapped = self._domain_index.get(_normalize_host(host)) if mapped is None: raise UnknownTenantError(f"未知 Host: {host!r}") tenant_slug, site_slug = mapped rt = self._by_slug[tenant_slug] - await self._ensure_ready(rt) + if rt.config.status != "active": + raise TenantDisabledError(f"租户已停用: {tenant_slug!r}") + rt = await self._ensure_ready(rt) site = rt.site_by_slug(site_slug) - if site is None: # 防御:索引与配置同源重建,正常不可达 + if site is None: # reload 竞态:域名命中后站点被换掉,视同未知 raise UnknownTenantError( - f"站点 {site_slug!r} 不存在于租户 {tenant_slug!r}") + f"站点 {site_slug!r} 不存在于租户 {tenant_slug!r}" + f"(可能已被 reload 移除)") return rt, site async def resolve(self, host: str) -> TenantRuntime: @@ -187,50 +221,71 @@ class TenantRegistry: async def _ensure_ready(self, rt: TenantRuntime) -> TenantRuntime: """懒建 engine + session_factory + ensure_schema,per-slug Lock 串行。 - 双检:抢到锁后重看 engine——等锁期间前一个协程可能已建好。 - 建库失败 dispose 半开 engine 并保持未就绪,下次 resolve 重试。 + 持锁后现势检查:_by_slug 条目已被 registry_reload 换掉 → 放弃本次 + 建库(否则旧 runtime 会产出无人 dispose 的孤儿 engine),按当前 + 条目重走;双检:等锁期间前一个协程可能已建好。建库失败 dispose + 半开 engine 并保持未就绪,下次 resolve 重试。 """ - if rt.engine is not None: - return rt - lock = self._locks.setdefault(rt.config.slug, asyncio.Lock()) - async with lock: + slug = rt.config.slug + while True: if rt.engine is not None: return rt - self._data_dir.mkdir(parents=True, exist_ok=True) - engine = make_tenant_engine(rt.db_path) - try: - await ensure_schema(engine) - except Exception: - await engine.dispose() - raise - rt.engine = engine - rt.session_factory = async_sessionmaker( - engine, class_=AsyncSession, expire_on_commit=False) - return rt + lock = self._locks.setdefault(slug, asyncio.Lock()) + async with lock: + current = self._by_slug.get(slug) + if current is None: + raise UnknownTenantError(f"租户已不存在: {slug!r}") + if current is not rt: + rt = current # 条目已被 reload 换掉,按新条目重走 + continue + if rt.engine is not None: # 双检:等锁期间已建好 + return rt + self._data_dir.mkdir(parents=True, exist_ok=True) + engine = make_tenant_engine(rt.db_path) + try: + await ensure_schema(engine) + except Exception: + await engine.dispose() + raise + rt.engine = engine + rt.session_factory = async_sessionmaker( + engine, class_=AsyncSession, expire_on_commit=False) + return rt async def registry_reload(self, slug: str) -> None: """重读磁盘,原子替换条目;旧 runtime 若已建 engine,dispose 掉。 - 先全量校验(重读 tenant.json/wall.json + 候选域名索引)再替换—— - 坏配置抛异常时旧条目原样服务,不留半更新状态。 + 磁盘读取/校验在锁外,swap+dispose 段与 _ensure_ready/dispose_all + 取同一把 per-slug 锁:懒建要么完整发生在替换前(dispose 得到旧 + engine),要么持锁后发现条目已换而放弃——不存在"建到一半被换" + 的孤儿窗口。坏配置在替换前抛,旧条目原样服务,不留半更新状态。 """ if slug not in self._by_slug: raise UnknownTenantError(f"租户不存在: {slug!r}") - old = self._by_slug[slug] rt = self._load_tenant(self._seed_dir / slug) - candidate = dict(self._by_slug) - candidate[slug] = rt - new_index = self._build_domain_index(candidate) - self._by_slug[slug] = rt # 原子替换(单事件循环内无 await) - self._domain_index = new_index - if old.engine is not None: - await old.engine.dispose() - old.engine = old.session_factory = None + lock = self._locks.setdefault(slug, asyncio.Lock()) + async with lock: + old = self._by_slug.get(slug) + candidate = dict(self._by_slug) + candidate[slug] = rt + new_index = self._build_domain_index(candidate) # 冲突在此抛,状态未动 + self._by_slug[slug] = rt # 原子替换(锁内,无 await) + self._domain_index = new_index + if old is not None and old.engine is not None: + await old.engine.dispose() + old.engine = old.session_factory = None async def dispose_all(self) -> None: """关闭全部 engine 并复位为未就绪(测试 teardown 用——Windows 文件锁; - 复位后若再 resolve 会重新懒建,可安全复用)。""" - for rt in self._by_slug.values(): - if rt.engine is not None: - await rt.engine.dispose() - rt.engine = rt.session_factory = None + 复位后若再 resolve 会重新懒建,可安全复用)。 + + 与懒建/reload 取同一把 per-slug 锁:dispose 期间该租户不会有并发 + 建库(否则新建的 engine 会被本次复位漏掉或反过来被关)。迭代前 + 快照 _locks——迭代中并发 resolve 可能 setdefault 新锁。 + """ + for slug, lock in list(self._locks.items()): + async with lock: + rt = self._by_slug.get(slug) + if rt is not None and rt.engine is not None: + await rt.engine.dispose() + rt.engine = rt.session_factory = None diff --git a/seed_profiles/README.md b/seed_profiles/README.md new file mode 100644 index 0000000..83d1c01 --- /dev/null +++ b/seed_profiles/README.md @@ -0,0 +1,30 @@ +# seed_profiles —— 出厂租户档案 + +每个子目录是一个租户的出厂档案（Phase 1 的配置来源；Phase 2 换 platform.db 后作为"新建租户"的模板与"重置出厂"的基准）。 + +## 目录结构 + +``` +seed_profiles// +├── tenant.json # 站点 + 品牌 + 引擎配置（本文件是租户装配入口） +├── wall.json # 安全墙词表/话术（Task 4 由老项目 AST 提取生成） +├── prompts/ # 提示词 4 件套 md（Task 6 拷入） +│ ├── system.md / company_facts.md / extractor.md / evaluator.md +└── brand_values.json # 品牌值机器提取草稿（人工归一参考，运行时不读） +``` + +## tenant.json 要点 + +- **slug 不可变**：客户端存储 key（邀请卡状态、会话令牌）全部以 slug 命名，改动 = 重置所有客户页面记忆；site slug 同理。`slug` 字段必须与目录名一致（加载期校验）。 +- **站点模型**：域名挂在站点上，站点属于租户。每租户**至少一个站点**。站点两种接入方式（jiancai 的两个站点即示范）： + - **绑域名**：`main` 站 `"domains": ["chat2.agent.hnxinyouhe.com", ...]`，按请求 Host 自动识别（大小写/端口会先归一）； + - **空域名 + `data-site`**：`es` 站 `"domains": []`，与 main 共用挂件域名，嵌入代码以 `data-site="es"` 参数选择站点。 +- **`site.brand` 是覆盖层**：缺省项继承 `config.brand`（深合并），只覆盖门面（标题/欢迎语等）；AI 脑子与线索管道留在租户级。 +- **wall 不写在 tenant.json 里**：加载器会把同目录 `wall.json` **隐式并入** `config["wall"]`（原始词表，供后台编辑/回滚），编译产物挂 runtime 的 `compiled_wall` 供引擎消费——词表数据源只有 wall.json 一份。 +- **`config` 内不含 prompts**：提示词由目录内 `prompts/*.md` 提供（Task 6）。 +- 加载期显性失败（不流入运行期）：无效 JSON、slug 与目录名不一致、零站点、wall 编译错误、域名跨站点/跨租户重复绑定。 + +## 纪律 + +- 改动本目录后必须跑全量 pytest（出厂配置影响面大，见 CLAUDE.md）。 +- `brand_values.json` 是 Task 4 的机器提取草稿（`conflict=true` 处需人工归一），`tenant.json` 的品牌值应与其保持一致；它本身不被运行时读取。 diff --git a/tests/test_registry.py b/tests/test_registry.py index 01a5ebd..1f1e1a6 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -1,23 +1,56 @@ -"""TenantRegistry 单元契约(Task 5a 文件加载器)。 +"""TenantRegistry 单元契约(Task 5a 文件加载器 + 审查修复)。 seed_profiles//tenant.json + 同目录 wall.json → Registry; Host(小写/去端口)→站点→租户;首次 resolve 懒建 engine(WAL/pool=2)+ -ensure_schema;reload 原子换条目并 dispose 旧 engine。 +ensure_schema;reload 原子换条目并 dispose 旧 engine;停用租户 +TenantDisabledError(503 语义);加载期校验显性炸。 Windows 纪律:每个建过 engine 的用例结尾 await reg.dispose_all()—— 连接不关,文件锁着,tmp_path 清理 PermissionError。 """ import asyncio +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, ) -SEED_DIR = "seed_profiles" +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(墙数据不重复维护,compile_wall 直接可过)。""" + 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") + (tdir / "tenant.json").write_text( + json.dumps(tenant, ensure_ascii=False), encoding="utf-8") + return seed + + +# ---- 基础解析 ------------------------------------------------------------- async def test_resolve_by_host(tmp_path): @@ -74,8 +107,6 @@ async def test_host_normalization_case_and_port(tmp_path): async def test_concurrent_first_resolve_builds_engine_once(tmp_path, monkeypatch): """两个协程同时首 resolve 同一租户:engine 只建一次(asyncio.Lock 串行)。""" - import app.platform.registry as registry_mod - calls = [] real_make = registry_mod.make_tenant_engine @@ -105,3 +136,161 @@ async def test_make_tenant_engine_enables_wal(tmp_path): 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()