From c7fd5333207585634254b466639f562f944ded36 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Tue, 13 Oct 2026 08:50:16 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20TenantSource=20=E6=8A=BD=E8=B1=A1?= =?UTF-8?q?=E2=80=94=E2=80=94File=E6=BA=90(=E6=B5=8B=E8=AF=95/=E6=A1=A3?= =?UTF-8?q?=E6=A1=88)=E4=B8=8EDB=E6=BA=90(=E7=94=9F=E4=BA=A7)=EF=BC=8Cregi?= =?UTF-8?q?stry=20=E6=95=B0=E6=8D=AE=E6=BA=90=E5=8F=AF=E5=88=87=E6=8D=A2+r?= =?UTF-8?q?eload=20upsert?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- app/engine/routes/public.py | 5 +- app/engine/services/ai_service.py | 39 +++- app/platform/registry.py | 149 +++++++------ app/platform/tenant_source.py | 224 +++++++++++++++++++ tests/conftest.py | 3 + tests/test_admin_mounting.py | 2 +- tests/test_db_tenant_source.py | 356 ++++++++++++++++++++++++++++++ tests/test_host_routing.py | 3 +- tests/test_prompt_provider.py | 24 ++ 9 files changed, 720 insertions(+), 85 deletions(-) create mode 100644 app/platform/tenant_source.py create mode 100644 tests/test_db_tenant_source.py diff --git a/app/engine/routes/public.py b/app/engine/routes/public.py index 0f8d944..8d316de 100644 --- a/app/engine/routes/public.py +++ b/app/engine/routes/public.py @@ -81,7 +81,10 @@ async def avatar(request: Request): # 任意文件。显式拒绝分隔符与目录点,命中即回退静态兜底。 if name in {".", ".."} or "/" in name or "\\" in name: name = "" - candidate = runtime.seed_dir / "assets" / name if name else None + # seed_dir=None = DB 源租户(Task 4;资产目录尚无 DB 侧存储)——直接 + # 落静态兜底,不让 None 进路径拼接炸 500 + candidate = (runtime.seed_dir / "assets" / name + if name and runtime.seed_dir is not None else None) if candidate is not None and candidate.is_file(): return FileResponse( candidate, headers={"Access-Control-Allow-Origin": "*"}) diff --git a/app/engine/services/ai_service.py b/app/engine/services/ai_service.py index 2e73c54..68e40fe 100644 --- a/app/engine/services/ai_service.py +++ b/app/engine/services/ai_service.py @@ -47,6 +47,11 @@ def _format_deflection_pool(deflections) -> str: ) +# 四件套键名(单一事实源): PromptProvider 构造校验 / FileTenantSource 读盘 +# / DbTenantSource 快照组形态, 三处消费同一清单 +PROMPTS_KEYS = ("system", "company_facts", "extractor", "evaluator") + + @dataclass(frozen=True) class PromptProvider: """租户提示词四件套持有者(Task 6, R1 话术单一事实源)。 @@ -54,9 +59,12 @@ class PromptProvider: frozen(Task 9 收口, Task 6 审查遗留): 实例跨租户运行时/模块级 DEFAULT_PROVIDER 共享, 不可变防共享单例被毒化。 - - from_dir(dir): 读 system/company_facts/extractor/evaluator 四个 md, - 存实例字段——提示词属于租户(seed_profiles//prompts/),不再有 - 代码级 app/engine/prompts 副本; + - from_dir(dir): 读 system/company_facts/extractor/evaluator 四个 md + 委托 from_config——提示词属于租户(seed_profiles//prompts/), + 不再有代码级 app/engine/prompts 副本; + - from_config(md: dict)(Phase 2 Task 4): 快照 prompts 组(dict 四键) + 直接构造——registry 对文件/DB 两源的统一入口(文件源读盘入快照, + DB 源嵌 config_json);缺键/值非字符串 → 显性 ValueError 带键名; - build_system(wall_deflections): system.md 的 $DEFLECTION_POOL 占位符 替换为按语言枚举的话术清单 + "\\n\\n" + company_facts(行为 prompt 在前、事实卡居中、当轮提醒最后拼接——拼接位在 AIService 内); @@ -68,15 +76,28 @@ class PromptProvider: extractor: str evaluator: str + @classmethod + def from_config(cls, md: dict) -> "PromptProvider": + """快照 prompts 组 → PromptProvider。缺键/非字符串显性炸带键名 + (DB 快照坏数据不留到 build_system 才炸)。""" + bad = [k for k in PROMPTS_KEYS if not isinstance((md or {}).get(k), str)] + if bad: + raise ValueError(f"prompts 配置缺少(或非字符串)键: {', '.join(bad)}") + return cls( + system=md["system"], + company_facts=md["company_facts"], + extractor=md["extractor"], + evaluator=md["evaluator"]) + @classmethod def from_dir(cls, prompts_dir) -> "PromptProvider": d = Path(prompts_dir) - return cls( - system=(d / "system.md").read_text(encoding="utf-8"), - company_facts=(d / "company_facts.md").read_text(encoding="utf-8"), - extractor=(d / "extractor.md").read_text(encoding="utf-8"), - evaluator=(d / "evaluator.md").read_text(encoding="utf-8"), - ) + return cls.from_config({ + "system": (d / "system.md").read_text(encoding="utf-8"), + "company_facts": (d / "company_facts.md").read_text(encoding="utf-8"), + "extractor": (d / "extractor.md").read_text(encoding="utf-8"), + "evaluator": (d / "evaluator.md").read_text(encoding="utf-8"), + }) def build_system(self, wall_deflections=None) -> str: system = self.system diff --git a/app/platform/registry.py b/app/platform/registry.py index 9cfc77a..503d38c 100644 --- a/app/platform/registry.py +++ b/app/platform/registry.py @@ -1,9 +1,12 @@ -"""TenantRegistry——进程内租户注册表(Phase 1 文件加载器)。 +"""TenantRegistry——进程内租户注册表(Phase 1 文件加载;Task 4 起数据源 +抽象化)。 -数据源:seed_profiles//tenant.json(站点/品牌/引擎配置)+ 同目录 -wall.json(并入 snapshot["wall"],compile_wall 编译期显性炸)。Phase 2 换 -platform.db 时本模块的对外接口不变(spec §5.3:后台保存→原子替换条目, -进行中的会话下一条消息即用新配置)。 +数据源经 TenantSource 协议注入(app/platform/tenant_source.py): +FileTenantSource(seed_profiles/,测试/出厂档案)与 DbTenantSource +(platform.db,生产 Task 11 起装配)输出同构 LoadedTenant(snapshot = +配置组 ⊕ wall ⊕ prompts),本模块对源差异无感知。load_from_dir 保持 +为文件源便捷构造(既有调用零改动)。spec §5.3 的热加载语义不变:后台 +保存→原子替换条目,进行中的会话下一条消息即用新配置。 关键行为: - resolve_site(host):host 小写去端口 → 域名索引 → (TenantRuntime, Site); @@ -14,16 +17,17 @@ platform.db 时本模块的对外接口不变(spec §5.3:后台保存→原子 per-slug asyncio.Lock 串行,锁内双检防并发双建;持锁后现势检查——条目 已被 reload 换掉则放弃本次建库、按当前条目重走(旧 runtime 永不产生 孤儿 engine);建库失败 dispose 半开 engine,保持未就绪下次重试。 -- registry_reload(slug):磁盘读取/校验在锁外,swap+dispose 段与懒建/ - dispose_all 取同一把 per-slug 锁。坏配置(无效 JSON/墙编译失败/零站点/ - 域名冲突)在替换前抛,旧条目原样服务;替换后 dispose 旧 engine - (Windows 下不关=库文件句柄泄漏)。 +- registry_reload(slug):经 source 重读(锁外),swap+dispose 段与懒建/ + dispose_all 取同一把 per-slug 锁。upsert 语义(Task 4):slug 未加载但 + source 已知(DB/文件新建租户)→ 载入插入;source 也不知道 → + UnknownTenantError。坏配置(无效 JSON/墙编译失败/零站点/域名冲突)在 + 替换前抛,旧条目原样服务;替换后 dispose 旧 engine(Windows 下不关= + 库文件句柄泄漏)。 测试/teardown 纪律:凡建过 engine 的 Registry 用完必须 await dispose_all() (Windows 文件锁,tmp 目录删不掉)。 """ import asyncio -import json import pathlib from dataclasses import dataclass @@ -39,6 +43,11 @@ from app.engine.models import ensure_schema from app.engine.services.ai_service import PromptProvider from app.engine.wall_schema import CompiledWall, compile_wall from app.platform.tenant_config import Site, TenantConfig +from app.platform.tenant_source import ( + FileTenantSource, + LoadedTenant, + TenantSource, +) class UnknownTenantError(LookupError): @@ -50,19 +59,6 @@ class TenantDisabledError(LookupError): 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 @@ -93,16 +89,17 @@ class TenantRuntime: """单租户运行时。 加载期就绪:config(TenantConfig:data=快照⊕出厂默认,含 sites)/ - db_path(该租户 SQLite 文件路径)/seed_dir(seed_profiles//)/ - compiled_wall(墙编译产物)/prompts(Task 6 PromptProvider,from_dir - (seed_dir/prompts) 四件套——$DEFLECTION_POOL 话术池注入的单一消费口)。 + db_path(该租户 SQLite 文件路径)/seed_dir(文件源=租户目录;DB 源 + None——avatar 解析链对其回退静态兜底)/compiled_wall(墙编译产物)/ + prompts(PromptProvider.from_config(snapshot["prompts"]),两源统一 + 入口——$DEFLECTION_POOL 话术池注入的单一消费口)。 懒建:engine/session_factory(Registry 首次 resolve,ensure_schema 已跑)。 manager(Task 5b)由装配线另行构造注入,不在本类持有。 """ config: TenantConfig db_path: pathlib.Path - seed_dir: pathlib.Path + seed_dir: pathlib.Path | None compiled_wall: CompiledWall prompts: PromptProvider engine: AsyncEngine | None = None @@ -120,9 +117,9 @@ class TenantRegistry: """tenant slug → TenantRuntime;域名 → (tenant_slug, site_slug) 索引。 reload 重建索引而非增量维护——租户数是个位数,整建最简且无漂移。""" - def __init__(self): - self._seed_dir: pathlib.Path | None = None - self._data_dir: pathlib.Path | None = None + def __init__(self, source: TenantSource, data_dir): + self._source = source + self._data_dir = pathlib.Path(data_dir) self._by_slug: dict[str, TenantRuntime] = {} self._domain_index: dict[str, tuple[str, str]] = {} self._locks: dict[str, asyncio.Lock] = {} @@ -131,52 +128,52 @@ class TenantRegistry: @classmethod def load_from_dir(cls, seed_dir, data_dir) -> "TenantRegistry": - """seed_profiles//tenant.json(+同目录 wall.json)→ Registry。 + """文件源便捷构造(既有调用零改动):seed_profiles// + tenant.json + wall.json + prompts/ → Registry,同步完成。 加载期显性失败(不流入运行期):目录无任何 tenant.json(seed_dir 路径拼错→静默空注册表→全量 404,必须炸)、无效 JSON、slug 与目录 名不一致、零站点、wall 编译错误、域名重复绑定、prompts 读取失败 - (Task 6: prompts/ 目录缺失或四件套不齐 → PromptProvider.from_dir - 的 FileNotFoundError,加载即炸,不流入运行期)。 + (prompts/ 目录缺失或四件套不齐 → read_text 的 FileNotFoundError, + 加载即炸,不流入运行期)。 """ - reg = cls() - reg._seed_dir = pathlib.Path(seed_dir) - reg._data_dir = pathlib.Path(data_dir) - tenant_jsons = sorted(reg._seed_dir.glob("*/tenant.json")) - if not tenant_jsons: - raise ValueError(f"{reg._seed_dir} 下没有任何 /tenant.json") - for tenant_json in tenant_jsons: - rt = reg._load_tenant(tenant_json.parent) - if rt.config.slug in reg._by_slug: - raise ValueError(f"租户 slug 重复: {rt.config.slug}") - reg._by_slug[rt.config.slug] = rt - reg._domain_index = reg._build_domain_index(reg._by_slug) + source = FileTenantSource(seed_dir) + reg = cls(source, data_dir) + reg._install(source.read_all()) return reg - def _load_tenant(self, tenant_dir: pathlib.Path) -> TenantRuntime: - slug = tenant_dir.name - tenant_json = _read_json(tenant_dir / "tenant.json") - if tenant_json.get("slug") != slug: - raise ValueError( - f"{tenant_dir / 'tenant.json'}: slug 字段 " - f"{tenant_json.get('slug')!r} 与目录名 {slug!r} 不一致") - sites = tenant_json.get("sites") or [] - if not sites: - raise ValueError(f"{slug}: 每租户至少一个站点(sites 为空)") - # wall 并入 snapshot["wall"]:data["wall"] 保留原始词表(后台编辑/ - # 回滚用),编译产物挂 runtime.compiled_wall(引擎消费,单一编译) - wall_cfg = _read_json(tenant_dir / "wall.json") - snapshot = dict(tenant_json.get("config") or {}) - snapshot["wall"] = wall_cfg + async def load_all(self) -> None: + """初始异步加载(DB 源等 async 源的装配入口;文件源走 + load_from_dir 同步路径,不经过这里)。空源由 source 自己显性炸。""" + self._install(await self._source.load_all()) + + def _install(self, loaded: list[LoadedTenant]) -> None: + """LoadedTenant 列表 → 全量替换 _by_slug + 域名索引(整体重建, + 无半更新状态;冲突在赋值前抛,旧状态原样)。""" + if not loaded: + raise ValueError(f"数据源未返回任何租户: {self._source!r}") + by_slug: dict[str, TenantRuntime] = {} + for lt in loaded: + rt = self._build_runtime(lt) + if rt.config.slug in by_slug: + raise ValueError(f"租户 slug 重复: {rt.config.slug}") + by_slug[rt.config.slug] = rt + self._by_slug = by_slug + self._domain_index = self._build_domain_index(by_slug) + + def _build_runtime(self, lt: LoadedTenant) -> TenantRuntime: + """LoadedTenant → TenantRuntime(两源同构快照的单一消费点: + from_snapshot 出厂默认合并 / compile_wall 墙编译 / + PromptProvider.from_config 四件套)。""" config = TenantConfig.from_snapshot( - slug, tenant_json.get("name", slug), snapshot, - sites=sites, status=tenant_json.get("status", "active")) + lt.slug, lt.name, lt.snapshot, + sites=lt.sites, status=lt.status) return TenantRuntime( config=config, - db_path=self._data_dir / f"{slug}.db", - seed_dir=tenant_dir, - compiled_wall=compile_wall(wall_cfg), - prompts=PromptProvider.from_dir(tenant_dir / "prompts")) + db_path=self._data_dir / f"{lt.slug}.db", + seed_dir=lt.seed_dir, + compiled_wall=compile_wall(lt.snapshot.get("wall")), + prompts=PromptProvider.from_config(lt.snapshot.get("prompts"))) def _build_domain_index( self, by_slug: dict[str, TenantRuntime] @@ -272,23 +269,31 @@ class TenantRegistry: return rt async def registry_reload(self, slug: str) -> None: - """重读磁盘,原子替换条目;旧 runtime 若已建 engine,dispose 掉。 + """经 source 重读单租户,原子替换条目;旧 runtime 若已建 engine, + dispose 掉。 - 磁盘读取/校验在锁外,swap+dispose 段与 _ensure_ready/dispose_all + upsert 语义(Task 4,评审修订):slug 不在 _by_slug 但 source 已知 + (新建租户已落库/落盘)→ 载入插入——插入同样走 per-slug 锁 + 整建 + 域名索引,与替换完全同路;source 也不知道(slug 拼错/未保存)→ + UnknownTenantError。 + + source 读取/校验在锁外,swap+dispose 段与 _ensure_ready/dispose_all 取同一把 per-slug 锁:懒建要么完整发生在替换前(dispose 得到旧 engine),要么持锁后发现条目已换而放弃——不存在"建到一半被换" 的孤儿窗口。坏配置在替换前抛,旧条目原样服务,不留半更新状态。 """ - if slug not in self._by_slug: - raise UnknownTenantError(f"租户不存在: {slug!r}") - rt = self._load_tenant(self._seed_dir / slug) + try: + loaded = await self._source.load_tenant(slug) + except LookupError as exc: + raise UnknownTenantError(f"租户不存在: {slug!r}") from exc + rt = self._build_runtime(loaded) 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._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() diff --git a/app/platform/tenant_source.py b/app/platform/tenant_source.py new file mode 100644 index 0000000..ba00fa8 --- /dev/null +++ b/app/platform/tenant_source.py @@ -0,0 +1,224 @@ +"""TenantSource——租户数据源抽象(Phase 2 Task 4)。 + +registry 只认本模块的 TenantSource 协议(async load_all / load_tenant), +不再直连文件系统:测试/出厂档案走 FileTenantSource,生产(Task 11 起装配) +走 DbTenantSource(platform.db)。两源输出同构 LoadedTenant: + +- sites: raw 站点列表,形态 = 文件源 tenant.json 的 sites 数组 + ({slug, name, domains, brand})——DB 源把 sites.brand_overlay 列映射进 + "brand" 键(TenantConfig.from_snapshot 的站点覆盖合并链消费同一形态, + 未覆盖项继承租户级 brand); +- snapshot: 租户配置快照 dict = 配置组(brand/push/ai/...) ⊕ wall(墙 + 词表)⊕ prompts(提示词四件套文本)——两源键形态一致,registry 侧 + 单一消费路径(compile_wall / PromptProvider.from_config)。 + +文件源行为增量(本任务,等价迁移):prompts 四件套并入 snapshot["prompts"] +——旧路径 PromptProvider.from_dir 直读盘、不进快照;对齐后 DB 源(prompts +嵌 config_json)与文件源输出形态一致,registry 不再感知源差异。 + +错误契约(两源一致,registry 依赖): +- 未知 slug → LookupError(registry_reload 映射为 UnknownTenantError); +- 数据坏(缺版本/坏 JSON/零站点/缺 wall·prompts 组)→ ValueError 带 + 租户名(加载期显性失败,不流入运行期)。 +""" +import json +import pathlib +from dataclasses import dataclass +from typing import Protocol + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession + +from app.engine.services.ai_service import PROMPTS_KEYS +from app.platform.db import Site, SiteDomain, Tenant, TenantConfigVersion + + +@dataclass(frozen=True) +class LoadedTenant: + """单租户的源侧加载产物(两源同构)。 + + snapshot 含 wall/prompts 键(见模块 docstring);sites 为 raw dict + 列表(TenantConfig.from_snapshot 直接消费);seed_dir 文件源专属 + (租户目录,DB 源 None——avatar 解析链对 None 回退静态兜底)。 + """ + + slug: str + name: str + status: str + sites: list[dict] + snapshot: dict + seed_dir: pathlib.Path | None = None + + +class TenantSource(Protocol): + """registry 的数据源接口(纯读,无缓存——每次现读)。""" + + async def load_all(self) -> list[LoadedTenant]: + """全量加载(启动/整表重读)。空源 → ValueError(路径拼错静默 + 空注册表 = 全量 404,必须炸)。""" + ... + + async def load_tenant(self, slug: str) -> LoadedTenant: + """单租户重读(registry_reload 用)。未知 slug → LookupError。""" + ... + + +def _read_json(path: pathlib.Path): + """读 JSON 文件;读不了/解析不了都带路径抛 ValueError——哪份文件 + 坏了一目了然(裸 FileNotFoundError/JSONDecodeError 不含路径上下文)。 + (自 registry 原样迁入,报错文案逐字等价。)""" + 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 + + +class FileTenantSource: + """文件源:seed_profiles//(测试/出厂档案)。 + + 行为自 registry._load_tenant 逐字迁入(含校验与报错文案);唯一增量 + = prompts 四件套读入 snapshot["prompts"](两源形态对齐)。读盘是纯 + 同步快操作——同步本体 read_all/read_tenant 供 load_from_dir 便捷构造 + 同步调用;async load_all/load_tenant 为 TenantSource 协议适配 + (registry 的统一 async 消费面,直接返回本体结果)。 + """ + + def __init__(self, seed_dir): + self.seed_dir = pathlib.Path(seed_dir) + + # ---- 同步本体 --------------------------------------------------------- + + def read_all(self) -> list[LoadedTenant]: + tenant_jsons = sorted(self.seed_dir.glob("*/tenant.json")) + if not tenant_jsons: + raise ValueError(f"{self.seed_dir} 下没有任何 /tenant.json") + return [self.read_tenant(p.parent.name) for p in tenant_jsons] + + def read_tenant(self, slug: str) -> LoadedTenant: + tenant_json_path = self.seed_dir / slug / "tenant.json" + if not tenant_json_path.is_file(): + raise LookupError(f"文件源没有租户 {slug!r}(缺 {tenant_json_path})") + tenant_json = _read_json(tenant_json_path) + if tenant_json.get("slug") != slug: + raise ValueError( + f"{tenant_json_path}: slug 字段 " + f"{tenant_json.get('slug')!r} 与目录名 {slug!r} 不一致") + sites = tenant_json.get("sites") or [] + if not sites: + raise ValueError(f"{slug}: 每租户至少一个站点(sites 为空)") + # wall 并入 snapshot["wall"]:data["wall"] 保留原始词表(后台编辑/ + # 回滚用),编译产物挂 runtime.compiled_wall(引擎消费,单一编译) + snapshot = dict(tenant_json.get("config") or {}) + snapshot["wall"] = _read_json(self.seed_dir / slug / "wall.json") + prompts_dir = self.seed_dir / slug / "prompts" + snapshot["prompts"] = { + name: (prompts_dir / f"{name}.md").read_text(encoding="utf-8") + for name in PROMPTS_KEYS + } + return LoadedTenant( + slug=slug, + name=tenant_json.get("name", slug), + status=tenant_json.get("status", "active"), + sites=list(sites), + snapshot=snapshot, + seed_dir=self.seed_dir / slug) + + # ---- 协议适配(registry 的 async 消费面)-------------------------------- + + async def load_all(self) -> list[LoadedTenant]: + return self.read_all() + + async def load_tenant(self, slug: str) -> LoadedTenant: + return self.read_tenant(slug) + + +class DbTenantSource: + """platform.db 源(生产,Task 11 起装配;本任务先交付能力)。 + + tenants JOIN sites(brand_overlay → 站点 brand 覆盖层)/site_domains + + 当前生效版本(MAX(version),db.py 纪律:历史不可变,当前版取最大) + 的 config_json 解析。版本快照由 Task 5 创建向导写入(config 组 ⊕ wall + ⊕ prompts 同构嵌套)。 + + 已知边界:site.status 列暂不参与映射(全量输出)——Task 5 站点 + CRUD/停用接线时定语义。 + """ + + def __init__(self, engine: AsyncEngine): + self._engine = engine + + async def load_all(self) -> list[LoadedTenant]: + async with AsyncSession(self._engine) as session: + tenants = (await session.execute( + select(Tenant).order_by(Tenant.slug))).scalars().all() + if not tenants: + raise ValueError( + "platform.db 没有任何租户(tenants 表为空)——" + "库路径拼错会静默空注册表(全量 404),必须炸") + return [await self._load_one(session, t) for t in tenants] + + async def load_tenant(self, slug: str) -> LoadedTenant: + async with AsyncSession(self._engine) as session: + tenant = (await session.execute( + select(Tenant).where(Tenant.slug == slug))).scalar_one_or_none() + if tenant is None: + raise LookupError(f"platform.db 中不存在租户: {slug!r}") + return await self._load_one(session, tenant) + + async def _load_one( + self, session: AsyncSession, tenant: Tenant) -> LoadedTenant: + sites_rows = (await session.execute( + select(Site).where(Site.tenant_id == tenant.id) + .order_by(Site.id))).scalars().all() + if not sites_rows: + raise ValueError(f"{tenant.slug}: 每租户至少一个站点(sites 为空)") + domains = (await session.execute( + select(SiteDomain) + .where(SiteDomain.site_id.in_([s.id for s in sites_rows])) + .order_by(SiteDomain.id))).scalars().all() + domains_by_site: dict[int, list[str]] = {} + for d in domains: + domains_by_site.setdefault(d.site_id, []).append(d.domain) + + version = (await session.execute( + select(TenantConfigVersion) + .where(TenantConfigVersion.tenant_id == tenant.id) + .order_by(TenantConfigVersion.version.desc()) + .limit(1))).scalar_one_or_none() + if version is None: + raise ValueError( + f"租户 {tenant.name!r}({tenant.slug}) 没有任何配置版本" + f"(tenant_config_versions 无行)——先在后台保存配置出版本 1") + try: + snapshot = json.loads(version.config_json) + except ValueError as exc: + raise ValueError( + f"租户 {tenant.name!r}({tenant.slug}) 版本 " + f"{version.version} 的 config_json 无效 JSON: {exc}") from exc + if not isinstance(snapshot, dict): + raise ValueError( + f"租户 {tenant.name!r}({tenant.slug}) 版本 " + f"{version.version} 的 config_json 不是 JSON 对象" + f"(实际 {type(snapshot).__name__})") + for key in ("wall", "prompts"): + if not isinstance(snapshot.get(key), dict): + raise ValueError( + f"租户 {tenant.name!r}({tenant.slug}) 版本 " + f"{version.version} 的 config_json 缺少 {key!r} 配置组" + f"(或其不是对象)") + + sites = [{"slug": s.slug, "name": s.name, + "domains": list(domains_by_site.get(s.id, [])), + "brand": dict(s.brand_overlay or {})} + for s in sites_rows] + return LoadedTenant( + slug=tenant.slug, + name=tenant.name, + status=tenant.status, + sites=sites, + snapshot=snapshot, + seed_dir=None) diff --git a/tests/conftest.py b/tests/conftest.py index 134fb86..7ed068d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -139,6 +139,9 @@ def build_auto_runtime(): # (与 5b 之前 default_runtime 兜底的语义一致,既有断言零改动) JIANCAI_HOST = "chat2.agent.hnxinyouhe.com" AUTO_HOST = "chat.agent.hnxinyouhe.com" +# 未知 Host 哨兵(原 test_admin_mounting/test_host_routing 各自定义, +# Task 4 收拢到 conftest 单一定义) +NOPE_HOST = "nope.example.com" def build_app(seed_dir, data_dir): diff --git a/tests/test_admin_mounting.py b/tests/test_admin_mounting.py index 94c4f9a..4a75285 100644 --- a/tests/test_admin_mounting.py +++ b/tests/test_admin_mounting.py @@ -9,7 +9,7 @@ # 的 test_tenant_middleware_is_outermost_user_middleware 覆盖 # (user_middleware[0] 断言)——全量回归确认仍绿,此处不重复建。 -NOPE_HOST = "nope.example.com" +from tests.conftest import NOPE_HOST def test_platform_api_bypasses_tenant_resolution(app_client): diff --git a/tests/test_db_tenant_source.py b/tests/test_db_tenant_source.py new file mode 100644 index 0000000..6635e45 --- /dev/null +++ b/tests/test_db_tenant_source.py @@ -0,0 +1,356 @@ +"""TenantSource 抽象与 DB 源(Phase 2 Task 4)。 + +两源输出同构 LoadedTenant:sites(tenant.json sites 形态,DB 源把 +sites.brand_overlay 列映射进站点 "brand" 键)+ snapshot(配置组 ⊕ wall +⊕ prompts 四件套文本——文件源的行为增量:prompts 并入快照,两源对齐)。 + +registry 构造收 source:load_from_dir 退化为 File 源便捷构造(既有调用 +零改动,既有 test_registry 全绿即证文件源行为等价);registry_reload 走 +source 并支持 upsert——source 已知但未加载的 slug 载入插入(Task 5 +创建租户后 reload 依赖此语义),source 也不知道 → UnknownTenantError。 + +Windows 纪律:建过 engine 的用例结尾 await engine.dispose()/dispose_all() +(文件锁,tmp 目录才删得掉)。 +""" +import json +import shutil +from pathlib import Path + +import pytest +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from app.engine.services.ai_service import PromptProvider +from app.platform.db import ( + Site, + SiteDomain, + Tenant, + TenantConfigVersion, + ensure_platform_schema, + make_platform_engine, +) +from app.platform.registry import UnknownTenantError, TenantRegistry +from app.platform.tenant_source import DbTenantSource, FileTenantSource + +REPO_ROOT = Path(__file__).resolve().parents[1] +SEED_DIR = REPO_ROOT / "seed_profiles" +JC_DOMAIN = "db.jc.example.com" # DB 源测试专属域名(不与文件源种子撞) + +_PROMPTS_KEYS = ("system", "company_facts", "extractor", "evaluator") + + +def _jiancai_snapshot(prefix="ai-kefu-jc"): + """版本 config_json 构造助手:seed jiancai 的 config 组 + wall.json + + prompts 四件套合并(Task 5 创建向导将产出同构内容)。""" + tenant = json.loads( + (SEED_DIR / "jiancai" / "tenant.json").read_text(encoding="utf-8")) + prompts_dir = SEED_DIR / "jiancai" / "prompts" + snapshot = dict(tenant["config"]) + snapshot["push"] = dict(snapshot.get("push") or {}, prefix=prefix) + snapshot["wall"] = json.loads( + (SEED_DIR / "jiancai" / "wall.json").read_text(encoding="utf-8")) + snapshot["prompts"] = { + name: (prompts_dir / f"{name}.md").read_text(encoding="utf-8") + for name in _PROMPTS_KEYS + } + return snapshot + + +async def _make_engine(tmp_path): + engine = make_platform_engine(tmp_path / "platform.db") + await ensure_platform_schema(engine) + return engine + + +async def _add_jiancai(engine, prefix="ai-kefu-jc", version=1, + config_json=None, with_version=True): + """jiancai 租户 + main(JC_DOMAIN)+ es(brand_overlay)+ 版本行。""" + async with AsyncSession(engine, expire_on_commit=False) as session: + tenant = Tenant(slug="jiancai", name="HESUPIPE 建材", status="active") + session.add(tenant) + await session.flush() + main = Site(tenant_id=tenant.id, slug="main", name="main") + es = Site(tenant_id=tenant.id, slug="es", name="es", + brand_overlay={"title": "HESUPIPE Soporte"}) + session.add_all([main, es]) + await session.flush() + session.add(SiteDomain(domain=JC_DOMAIN, site_id=main.id)) + if with_version: + session.add(TenantConfigVersion( + tenant_id=tenant.id, version=version, + config_json=config_json if config_json is not None + else json.dumps(_jiancai_snapshot(prefix), ensure_ascii=False))) + await session.commit() + return tenant.id + + +async def _add_version(engine, tenant_id, version, prefix): + async with AsyncSession(engine, expire_on_commit=False) as session: + session.add(TenantConfigVersion( + tenant_id=tenant_id, version=version, + config_json=json.dumps(_jiancai_snapshot(prefix), + ensure_ascii=False))) + await session.commit() + + +async def _add_tenant(engine, slug, domain, prefix): + """极简单租户(无 es 覆盖)——upsert 测试用。""" + async with AsyncSession(engine, expire_on_commit=False) as session: + tenant = Tenant(slug=slug, name=slug) + session.add(tenant) + await session.flush() + site = Site(tenant_id=tenant.id, slug="main", name="main") + session.add(site) + await session.flush() + session.add(SiteDomain(domain=domain, site_id=site.id)) + session.add(TenantConfigVersion( + tenant_id=tenant.id, version=1, + config_json=json.dumps(_jiancai_snapshot(prefix), + ensure_ascii=False))) + await session.commit() + + +@pytest.fixture +async def platform_db_with_tenant(tmp_path): + """tmp platform.db + jiancai(main+es 含 brand_overlay)+域名+版本1。""" + engine = await _make_engine(tmp_path) + await _add_jiancai(engine) + yield engine + await engine.dispose() + + +# ---- DB 源加载 --------------------------------------------------------------- + + +async def test_db_source_loads_tenant(platform_db_with_tenant): + src = DbTenantSource(platform_db_with_tenant) + loaded = await src.load_all() + assert [lt.slug for lt in loaded] == ["jiancai"] + lt = loaded[0] + assert lt.name == "HESUPIPE 建材" and lt.status == "active" + assert lt.snapshot["push"]["prefix"] == "ai-kefu-jc" # 快照组直读 + assert "system" in lt.snapshot["prompts"] # prompts 嵌 config_json + assert lt.seed_dir is None # DB 源无种子目录 + main = [s for s in lt.sites if s["slug"] == "main"][0] + es = [s for s in lt.sites if s["slug"] == "es"][0] + assert main["domains"] == [JC_DOMAIN] + assert es["brand"]["title"] == "HESUPIPE Soporte" # brand_overlay 生效 + assert es["domains"] == [] + + +async def test_db_source_load_tenant_single_and_unknown( + platform_db_with_tenant): + src = DbTenantSource(platform_db_with_tenant) + lt = await src.load_tenant("jiancai") + assert lt.slug == "jiancai" + with pytest.raises(LookupError, match="nope"): + await src.load_tenant("nope") + + +async def test_db_source_empty_db_raises(tmp_path): + engine = await _make_engine(tmp_path) + try: + with pytest.raises(ValueError, match="没有任何租户"): + await DbTenantSource(engine).load_all() + finally: + await engine.dispose() + + +async def test_db_source_no_version_raises(tmp_path): + """有租户无版本 → 显性 ValueError 带租户名(哪个租户缺版本一目了然)。""" + engine = await _make_engine(tmp_path) + try: + await _add_jiancai(engine, with_version=False) + with pytest.raises(ValueError) as excinfo: + await DbTenantSource(engine).load_all() + assert "jiancai" in str(excinfo.value) and "没有任何配置版本" in str( + excinfo.value) + finally: + await engine.dispose() + + +async def test_db_source_bad_json_raises(tmp_path): + """config_json 无效 JSON → 带租户名与版本号的清晰错误。 + + CHECK(json_valid) 挡住正常写入方,坏行只能来自约束启用前的历史/ + 外部直改——经 PRAGMA ignore_check_constraints 直插构造,防御路径 + 仍要显性报错。 + """ + engine = await _make_engine(tmp_path) + try: + tid = await _add_jiancai(engine) + async with engine.connect() as conn: + await conn.exec_driver_sql("PRAGMA ignore_check_constraints=ON") + await conn.execute( + text("INSERT INTO tenant_config_versions " + "(tenant_id, version, config_json, updated_by, created_at) " + "VALUES (:t, :v, :c, 'test', '2026-01-01 00:00:00')"), + {"t": tid, "v": 2, "c": "{oops"}) + await conn.commit() + with pytest.raises(ValueError) as excinfo: + await DbTenantSource(engine).load_tenant("jiancai") + msg = str(excinfo.value) + assert "jiancai" in msg and "版本 2" in msg and "JSON" in msg + finally: + await engine.dispose() + + +async def test_db_source_non_object_json_raises(tmp_path): + """合法 JSON 但非对象(如数组)→ 显性错误带租户名与版本号。""" + engine = await _make_engine(tmp_path) + try: + await _add_jiancai(engine, config_json="[1,2]") + with pytest.raises(ValueError) as excinfo: + await DbTenantSource(engine).load_all() + msg = str(excinfo.value) + assert "jiancai" in msg and "版本 1" in msg and "不是 JSON 对象" in msg + finally: + await engine.dispose() + + +async def test_db_source_missing_prompts_group_raises(tmp_path): + """快照缺 wall/prompts 组 → 显性错误带租户名与版本号(不留给 + compile_wall/from_config 出无上下文的错)。""" + engine = await _make_engine(tmp_path) + try: + snapshot = _jiancai_snapshot() + del snapshot["prompts"] + await _add_jiancai(engine, config_json=json.dumps( + snapshot, ensure_ascii=False)) + with pytest.raises(ValueError) as excinfo: + await DbTenantSource(engine).load_all() + msg = str(excinfo.value) + assert "jiancai" in msg and "prompts" in msg and "版本 1" in msg + finally: + await engine.dispose() + + +# ---- 文件源(直测;等价性由既有 test_registry 全绿证明)----------------------- + + +async def test_file_source_loads_same_snapshot_shape(): + src = FileTenantSource(SEED_DIR) + loaded = await src.load_all() + by_slug = {lt.slug: lt for lt in loaded} + assert set(by_slug) == {"jiancai", "auto"} + jc = by_slug["jiancai"] + assert jc.snapshot["push"]["prefix"] == "ai-kefu-jc" + assert "system" in jc.snapshot["prompts"] # 行为增量:四件套并入快照 + assert jc.seed_dir == SEED_DIR / "jiancai" + es = [s for s in jc.sites if s["slug"] == "es"][0] + assert es["brand"]["title"] == "HESUPIPE Soporte" + # 快照 prompts 与 from_dir 读盘逐字一致(等价迁移直证) + assert (PromptProvider.from_config(jc.snapshot["prompts"]) + == PromptProvider.from_dir(SEED_DIR / "jiancai" / "prompts")) + + +def test_file_source_read_tenant_unknown_slug_raises(tmp_path): + """reload upsert 的语义地基:文件源对未知 slug 报 LookupError + (registry 映射回 UnknownTenantError,不与坏数据的 ValueError 混)。""" + src = FileTenantSource(tmp_path) # 空目录:一切 slug 未知 + with pytest.raises(LookupError, match="ghost"): + src.read_tenant("ghost") + + +def test_file_source_read_tenant_rereads_disk(tmp_path): + """read_tenant 每次现读盘(reload 换配置的数据通路)。""" + tdir = tmp_path / "t1" + 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({ + "slug": "t1", "name": "t1", "status": "active", + "sites": [{"slug": "main", "domains": ["a.example.com"]}], + "config": {"push": {"prefix": "v1"}}}, ensure_ascii=False), encoding="utf-8") + + src = FileTenantSource(tmp_path) + assert src.read_tenant("t1").snapshot["push"]["prefix"] == "v1" + tenant = json.loads((tdir / "tenant.json").read_text(encoding="utf-8")) + tenant["config"]["push"]["prefix"] = "v2" + (tdir / "tenant.json").write_text( + json.dumps(tenant, ensure_ascii=False), encoding="utf-8") + assert src.read_tenant("t1").snapshot["push"]["prefix"] == "v2" + + +# ---- registry × DB 源(reload / upsert)-------------------------------------- + + +async def test_registry_reload_via_db_source(tmp_path): + engine = await _make_engine(tmp_path) + try: + tid = await _add_jiancai(engine, prefix="v1") + reg = TenantRegistry( + DbTenantSource(engine), data_dir=tmp_path / "tenants") + await reg.load_all() + old = await reg.resolve(JC_DOMAIN) + assert old.config.data["push"]["prefix"] == "v1" + assert old.prompts.system # prompts 经 from_config 接线 + + await _add_version(engine, tid, 2, prefix="v2") + await reg.registry_reload("jiancai") + rt = await reg.resolve(JC_DOMAIN) + assert rt is not old + assert rt.config.data["push"]["prefix"] == "v2" # MAX(version) 生效 + assert old.engine is None # 旧 engine 已 dispose + await reg.dispose_all() + finally: + await engine.dispose() + + +async def test_registry_reload_upsert_new_tenant_db_source(tmp_path): + """source 已知但未加载的 slug(DB 新建租户)→ reload 载入插入不炸; + source 也不知道的 slug → UnknownTenantError(原语义保留)。""" + engine = await _make_engine(tmp_path) + try: + await _add_jiancai(engine) + reg = TenantRegistry( + DbTenantSource(engine), data_dir=tmp_path / "tenants") + await reg.load_all() + + await _add_tenant(engine, "demo", "demo.example.com", prefix="dm1") + await reg.registry_reload("demo") # 未加载的 slug:插入 + rt = await reg.resolve("demo.example.com") + assert rt.config.slug == "demo" + assert rt.config.data["push"]["prefix"] == "dm1" + + with pytest.raises(UnknownTenantError): + await reg.registry_reload("ghost") + await reg.dispose_all() + finally: + await engine.dispose() + + +def _write_min_seed(tmp_path, tenants): + """最小 seed 目录(拷真实 jiancai wall/prompts,免数据重复维护)。 + 幂等:第二次调用往同一 seed 下加新租户目录。""" + seed = tmp_path / "seed" + seed.mkdir(exist_ok=True) + for slug, tenant in tenants.items(): + tdir = seed / slug + tdir.mkdir(exist_ok=True) + shutil.copyfile(SEED_DIR / "jiancai" / "wall.json", tdir / "wall.json") + shutil.copytree(SEED_DIR / "jiancai" / "prompts", tdir / "prompts", + dirs_exist_ok=True) + (tdir / "tenant.json").write_text( + json.dumps(tenant, ensure_ascii=False), encoding="utf-8") + return seed + + +async def test_registry_reload_upsert_new_tenant_file_source(tmp_path): + """文件源同款 upsert 语义:新租户目录落盘后 reload 插入可 resolve + (Task 5 创建向导 + File 源部署形态的数据通路)。""" + _write_min_seed(tmp_path, {"t1": { + "slug": "t1", "name": "t1", "status": "active", + "sites": [{"slug": "main", "domains": ["a.example.com"]}], + "config": {}}}) + reg = TenantRegistry.load_from_dir( + tmp_path / "seed", data_dir=tmp_path / "data") + + _write_min_seed(tmp_path, {"t2": { + "slug": "t2", "name": "t2", "status": "active", + "sites": [{"slug": "main", "domains": ["b.example.com"]}], + "config": {}}}) + await reg.registry_reload("t2") + rt = await reg.resolve("b.example.com") + assert rt.config.slug == "t2" + await reg.dispose_all() diff --git a/tests/test_host_routing.py b/tests/test_host_routing.py index e8b0afd..7119672 100644 --- a/tests/test_host_routing.py +++ b/tests/test_host_routing.py @@ -16,13 +16,12 @@ from app.platform.host_middleware import TenantHostMiddleware from tests.conftest import ( AUTO_HOST, JIANCAI_HOST, + NOPE_HOST, SEED_DIR, app_test_client, build_app, ) -NOPE_HOST = "nope.example.com" - # ---- 中间件挂载顺序守护 ------------------------------------------------------- diff --git a/tests/test_prompt_provider.py b/tests/test_prompt_provider.py index 6458c11..1f4891d 100644 --- a/tests/test_prompt_provider.py +++ b/tests/test_prompt_provider.py @@ -136,6 +136,30 @@ def test_build_extractor_handles_none_info(): assert "{}" in out # existing_info or {} 语义保留 +# ---- from_config(Phase 2 Task 4: 快照 prompts 组构造)---------------------- + +def test_from_config_missing_key_raises(): + # 缺一键 → 显性 ValueError 带键名(不留到 build_system 才炸) + with pytest.raises(ValueError, match="company_facts"): + PromptProvider.from_config( + {"system": "s", "extractor": "e", "evaluator": "v"}) + + +def test_from_config_non_string_key_raises(): + # 值非字符串(JSON null 等)同样显性拒绝 + with pytest.raises(ValueError, match="system"): + PromptProvider.from_config( + {"system": None, "company_facts": "c", + "extractor": "e", "evaluator": "v"}) + + +def test_from_config_equals_from_dir(): + d = JC_PROMPTS + cfg = {name: (d / f"{name}.md").read_text(encoding="utf-8") + for name in ("system", "company_facts", "extractor", "evaluator")} + assert PromptProvider.from_config(cfg) == PromptProvider.from_dir(JC_PROMPTS) + + # ---- runtime 接线 -------------------------------------------------------- def test_registry_runtime_carries_prompts(tmp_path):