diff --git a/app/platform/registry.py b/app/platform/registry.py index 7d6663d..837cc68 100644 --- a/app/platform/registry.py +++ b/app/platform/registry.py @@ -29,6 +29,7 @@ FileTenantSource(seed_profiles/,测试/出厂档案)与 DbTenantSource """ import asyncio import pathlib +import re from dataclasses import dataclass from sqlalchemy import event @@ -49,6 +50,10 @@ from app.platform.tenant_source import ( TenantSource, ) +# 租户 slug 形态(创建侧 tenant_routes 原 SLUG_RE 的单一事实源——路径前缀 +# 解析与创建校验共用,永不漂移;spec 2026-10-19 §3.1) +TENANT_SLUG_RE = re.compile(r"^[a-z0-9-]+$") + class UnknownTenantError(LookupError): """Host/域名无法映射到任何租户(调用方语义:聊天/公开端点 404)。""" @@ -225,6 +230,23 @@ class TenantRegistry: """resolve_site 的便捷形式(双 API 一层实现)。""" return (await self.resolve_site(host))[0] + async def resolve_by_slug(self, slug: str) -> tuple[TenantRuntime, Site]: + """路径前缀模式(/t//,spec 2026-10-19 §5):slug → (runtime, + 首站点)。语义与 resolve_site 同款:未知 slug / 空站点表 → + UnknownTenantError;停用 → TenantDisabledError;懒建 engine、 + reload 竞态取 _ensure_ready 返回值(与 resolve_site 同理)。""" + rt = self._by_slug.get(slug) + if rt is None: + raise UnknownTenantError(f"未知租户 slug: {slug!r}") + if rt.config.status != "active": + raise TenantDisabledError(f"租户已停用: {slug!r}") + rt = await self._ensure_ready(rt) + if not rt.config.sites: + # 创建向导恒建 main 站点,此分支为防御(spec §5) + raise UnknownTenantError( + f"租户 {slug!r} 无站点——路径模式无从选站点") + return rt, rt.config.sites[0] + # ---- 装配期辅助(Task 5b: main.lifespan 逐租户启动清扫)------------------ def runtimes(self) -> tuple[TenantRuntime, ...]: diff --git a/tests/test_registry.py b/tests/test_registry.py index 1654f4e..e68b1be 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -9,6 +9,7 @@ Windows 纪律:每个建过 engine 的用例结尾 await reg.dispose_all()—— 连接不关,文件锁着,tmp_path 清理 PermissionError。 """ import asyncio +import dataclasses import json import shutil from pathlib import Path @@ -22,6 +23,7 @@ from app.platform.registry import ( UnknownTenantError, make_tenant_engine, ) +from app.platform.tenant_source import FileTenantSource REPO_ROOT = Path(__file__).resolve().parents[1] SEED_DIR = REPO_ROOT / "seed_profiles" @@ -296,3 +298,45 @@ async def test_reload_to_disabled_releases_engine(tmp_path): 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 直查——语义与 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()