ai-kefu-admin/tests/test_db_tenant_source.py

"""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 PROMPTS_KEYS, 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_config import TenantConfig
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 源测试专属域名(不与文件源种子撞)


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, es_status="active"):
    """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",
                  status=es_status,
                  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()


async def test_db_source_filters_disabled_sites(tmp_path):
    """终审 I-1:site.status != active 的站点不进 LoadedTenant(域名不进
    registry 索引 → 该域名 404;site_by_slug 取不到)。es 停用后只剩
    main;全部停用 → 空 sites(零站点校验只对库表形状)。"""
    engine = await _make_engine(tmp_path)
    try:
        await _add_jiancai(engine, es_status="disabled")
        lt = await DbTenantSource(engine).load_tenant("jiancai")
        assert [s["slug"] for s in lt.sites] == ["main"]   # es 被过滤
        main = lt.sites[0]
        assert main["status"] == "active"
        # TenantConfig 消费侧:Site 数据类带 status,域名索引只含 main
        config = TenantConfig.from_snapshot(
            lt.slug, lt.name, lt.snapshot, sites=lt.sites, status=lt.status)
        assert [s.slug for s in config.sites] == ["main"]
        assert config.sites[0].status == "active"

        # 全部停用 = 合法运营状态(门面全下线),输出空 sites 不炸
        async with AsyncSession(engine, expire_on_commit=False) as session:
            await session.execute(text(
                "UPDATE sites SET status='disabled'"))
            await session.commit()
        lt2 = await DbTenantSource(engine).load_tenant("jiancai")
        assert lt2.sites == []
    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"


def test_file_source_filters_disabled_sites(tmp_path):
    """终审 I-1:tenant.json 站点带 status: "disabled" → 不进
    LoadedTenant(文件源同步行为);无 status 键 → 默认 active,既有
    种子零改动兼容。"""
    _write_min_seed(tmp_path, {"t1": {
        "slug": "t1", "name": "t1", "status": "active",
        "sites": [
            {"slug": "main", "domains": ["a.example.com"]},      # 无键=active
            {"slug": "es", "status": "active", "domains": []},
            {"slug": "down", "status": "disabled",
             "domains": ["down.example.com"], "brand": {"title": "x"}},
        ],
        "config": {}}})
    lt = FileTenantSource(tmp_path / "seed").read_tenant("t1")
    assert [s["slug"] for s in lt.sites] == ["main", "es"]   # down 被滤
    # 消费侧同构:from_snapshot 只见 active 站点
    config = TenantConfig.from_snapshot(
        lt.slug, lt.name, lt.snapshot, sites=lt.sites, status=lt.status)
    assert [s.slug for s in config.sites] == ["main", "es"]
    assert all(s.status == "active" for s in config.sites)


# ---- 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()