ai-kefu-admin/tests/test_tenant_admin_api.py

"""租户管理 API(Phase 2 Task 5):超管的创建向导/站点/域名/停用 + 403 边界。

db_source_app/super_client 夹具(Task 6 起收拢到 conftest,与配置 API 测试
共用):app.state.registry 用 DbTenantSource(生产形态)指向临时
platform.db——创建 API 写库后 registry_reload 走 DB 源闭环(Task 4
的 upsert 语义:未加载的 slug 直接插入,创建即可用)。platform.db 起始
为空(schema+bootstrap 超管由 lifespan 跑)——租户全部经创建 API 进入,
registry 不做初始 load_all(生产装配的启动次序归 Task 11)。

跨循环纪律:platform engine(pool=2 常驻连接)只在 TestClient portal
循环内使用(lifespan+路由);测试线程直查 platform.db 一律走 _db_fetch
(独立 NullPool engine,即用即弃,绝不复用 portal 循环的常驻连接)。
"""
import asyncio
import json

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.pool import NullPool

from app.engine.services.ai_service import PromptProvider
from app.engine.wall_schema import compile_wall
from app.platform.db import (
    AdminUser,
    Site,
    SiteDomain,
    Tenant,
    TenantConfigVersion,
)
from app.platform.defaults import DEFAULT_CONFIG
from app.platform.registry import TenantDisabledError
from app.platform.tenant_source import FileTenantSource, _blankify
from tests.conftest import SEED_DIR

TENANTS = "/api/platform/tenants"
LOGIN = "/api/platform/auth/login"
CHANGE_PW = "/api/platform/auth/change-password"


# ---- 助手 ----------------------------------------------------------------------


def _auth(client):
    return {"Authorization": f"Bearer {client.platform_token}"}


def _bearer(token):
    return {"Authorization": f"Bearer {token}"}


def _create(client, slug="demo", profile="jiancai", name=None, admin=None,
            password=None, domains=None):
    """创建租户的便捷 POST(admin/password 缺省按 slug 派生,免撞用户名)。"""
    return client.post(TENANTS, headers=_auth(client), json={
        "name": name or f"{slug} 租户",
        "slug": slug,
        "profile": profile,
        "admin_username": admin or f"{slug}-admin",
        "admin_password": password or f"{slug}-pass-1",
        "domains": domains or [],
    })


def _tenant_admin_token(client, username, old_password, new_password="tn-pass-1"):
    """租户管理员登录 + 首登改密(过 409 门),返回可用 token。"""
    r = client.post(LOGIN, json={"username": username, "password": old_password})
    assert r.status_code == 200
    token = r.json()["token"]
    r = client.post(CHANGE_PW, headers=_bearer(token), json={
        "old_password": old_password, "new_password": new_password})
    assert r.status_code == 200
    return token


def _db_fetch(root, stmt):
    """测试线程直查 platform.db:独立 NullPool engine,同一 asyncio.run 内
    查完即 dispose(跨循环纪律,见模块 docstring)。"""
    engine = create_async_engine(
        f"sqlite+aiosqlite:///{(root / 'platform.db').as_posix()}",
        poolclass=NullPool)

    async def _run():
        try:
            async with AsyncSession(engine) as session:
                return list((await session.execute(stmt)).all())
        finally:
            await engine.dispose()

    return asyncio.run(_run())


def _current_snapshot(root, slug):
    """该租户当前生效版本(MAX)的 config_json 解析。"""
    rows = _db_fetch(root, select(TenantConfigVersion)
                     .join(Tenant, TenantConfigVersion.tenant_id == Tenant.id)
                     .where(Tenant.slug == slug)
                     .order_by(TenantConfigVersion.version.desc()))
    assert rows, f"租户 {slug} 没有配置版本"
    return json.loads(rows[0][0].config_json)


def _jc_snapshot():
    return FileTenantSource(SEED_DIR).read_tenant("jiancai").snapshot


# ---- 核心 6 项 ------------------------------------------------------------------


def test_create_tenant_from_profile(super_client, db_source_app, tmp_path):
    r = _create(super_client, slug="demo", profile="jiancai",
                domains=["demo.example.com"])
    assert r.status_code == 201
    body = r.json()
    assert body["slug"] == "demo"
    assert body["status"] == "active"
    assert body["version"] == 1
    assert body["site"] == {"slug": "main", "domains": ["demo.example.com"]}

    # platform.db: tenant + main 站点 + 版本1(config = seed 档案快照)
    rows = _db_fetch(tmp_path, select(Tenant).where(Tenant.slug == "demo"))
    assert len(rows) == 1
    tenant = rows[0][0]
    assert tenant.status == "active"
    sites = _db_fetch(tmp_path, select(Site).where(Site.tenant_id == tenant.id))
    assert [(row[0].slug, row[0].name) for row in sites] == [("main", "main")]
    versions = _db_fetch(
        tmp_path, select(TenantConfigVersion)
        .where(TenantConfigVersion.tenant_id == tenant.id))
    assert len(versions) == 1
    assert versions[0][0].version == 1
    assert json.loads(versions[0][0].config_json) == _jc_snapshot()
    domains = _db_fetch(
        tmp_path, select(SiteDomain).join(Site, SiteDomain.site_id == Site.id)
        .where(Site.tenant_id == tenant.id))
    assert [row[0].domain for row in domains] == ["demo.example.com"]

    # 租户管理员可登录(role=tenant, tenant_slug=demo, 引导改密标志)
    admins = _db_fetch(
        tmp_path, select(AdminUser).where(AdminUser.tenant_id == tenant.id))
    assert len(admins) == 1
    admin = admins[0][0]
    assert admin.username == "demo-admin"
    assert admin.must_change_password is True
    assert admin.password_hash.startswith("scrypt$")   # 哈希在库,不在响应
    r = super_client.post(LOGIN, json={
        "username": "demo-admin", "password": "demo-pass-1"})
    assert r.status_code == 200
    data = r.json()
    assert data["role"] == "tenant"
    assert data["tenant_slug"] == "demo"
    assert data["must_change_password"] is True

    # 创建后立即可用: reload upsert 后经中间件 resolve 命中(DB 源闭环)
    r = super_client.get("/api/public/brand", headers={"host": "demo.example.com"})
    assert r.status_code == 200
    assert r.json()["slug"] == "demo"


def test_create_blank_tenant(super_client, tmp_path):
    r = _create(super_client, slug="blank1", profile="blank",
                domains=["blank.example.com"])
    assert r.status_code == 201
    # 版本1快照: 身份组清零,wall/prompts 取 jiancai 档案
    snapshot = _current_snapshot(tmp_path, "blank1")
    jc = _jc_snapshot()
    db_brand = DEFAULT_CONFIG["brand"]
    assert snapshot["brand"]["title"] == db_brand["title"]
    assert snapshot["brand"]["whatsapp"] == db_brand["whatsapp"] == ""
    assert snapshot["brand"]["email"] == db_brand["email"] == ""
    assert snapshot["brand"]["privacy_url"] == db_brand["privacy_url"] == ""
    assert snapshot["push"]["prefix"] == ""
    assert snapshot["engine"]["context_seed_keys"] == {}
    # 非身份字段与 wall/prompts 保留(jiancai 档案)
    assert snapshot["brand"]["color"] == jc["brand"]["color"]
    assert snapshot["brand"]["topics"] == jc["brand"]["topics"]
    assert snapshot["wall"] == jc["wall"]
    assert snapshot["prompts"] == jc["prompts"]
    # wall/prompts 可消费(compile_wall / PromptProvider 能过)
    compile_wall(snapshot["wall"])
    PromptProvider.from_config(snapshot["prompts"])
    # E2E: blank 租户对外门面是出厂身份
    r = super_client.get("/api/public/brand", headers={"host": "blank.example.com"})
    assert r.status_code == 200
    assert r.json()["title"] == db_brand["title"]
    assert r.json()["whatsapp"] == ""


def test_slug_immutable(super_client, tmp_path):
    _create(super_client, slug="demo", name="原名")
    r = super_client.patch(f"{TENANTS}/demo", headers=_auth(super_client),
                           json={"slug": "demo2", "name": "新名"})
    assert r.status_code == 400
    assert "slug" in r.json()["detail"]
    # 400 先于字段应用: 名字没被捎带改掉,库里也只有原租户
    rows = _db_fetch(tmp_path, select(Tenant))
    assert [(row[0].slug, row[0].name) for row in rows] == [("demo", "原名")]


def test_add_domain_and_resolve(super_client):
    _create(super_client, slug="demo")            # domains=[]
    # 加域名前: 未知 Host 404
    r = super_client.get("/api/public/brand", headers={"host": "demo.example.com"})
    assert r.status_code == 404
    r = super_client.post(f"{TENANTS}/demo/sites/main/domains",
                          headers=_auth(super_client),
                          json={"domain": "demo.example.com"})
    assert r.status_code == 201
    # registry reload 后 resolve 命中
    r = super_client.get("/api/public/brand", headers={"host": "demo.example.com"})
    assert r.status_code == 200
    assert r.json()["slug"] == "demo"


def test_disable_tenant_releases(super_client, db_source_app):
    _create(super_client, slug="demo", domains=["demo.example.com"])
    assert super_client.get(
        "/api/public/brand", headers={"host": "demo.example.com"}).status_code == 200
    registry = db_source_app.state.registry
    rt_old = registry._by_slug["demo"]
    assert rt_old.engine is not None       # 懒建已发生(portal 循环内)

    r = super_client.patch(f"{TENANTS}/demo", headers=_auth(super_client),
                           json={"status": "disabled"})
    assert r.status_code == 200
    assert r.json()["status"] == "disabled"
    assert rt_old.engine is None           # reload 路径已 dispose 旧 engine

    # resolve 抛 TenantDisabledError(停用检查先于懒建,无 engine 副作用)
    with pytest.raises(TenantDisabledError):
        asyncio.run(registry.resolve_site("demo.example.com"))
    # 中间件映射 503 中性页
    r = super_client.get("/api/public/brand", headers={"host": "demo.example.com"})
    assert r.status_code == 503

    # 复启用 → resolve 恢复(新 runtime 重新懒建)
    r = super_client.patch(f"{TENANTS}/demo", headers=_auth(super_client),
                           json={"status": "active"})
    assert r.status_code == 200
    assert super_client.get(
        "/api/public/brand", headers={"host": "demo.example.com"}).status_code == 200


def test_tenant_admin_cannot_create_tenant(super_client):
    _create(super_client, slug="demo", admin="demo-admin",
            password="demo-pass-1")
    token = _tenant_admin_token(super_client, "demo-admin", "demo-pass-1")
    r = super_client.post(TENANTS, headers=_bearer(token), json={
        "name": "越权租户", "slug": "evil", "profile": "jiancai",
        "admin_username": "evil-admin", "admin_password": "evil-pass-1"})
    assert r.status_code == 403
    # 对照: 超管可建
    assert _create(super_client, slug="ok1").status_code == 201


# ---- _blankify 单测 --------------------------------------------------------------


def test_blankify_strips_identity_keeps_wall_prompts():
    jc = _jc_snapshot()
    out = _blankify(jc)
    assert out is not jc                                  # 不改入参
    assert jc["brand"]["title"] == "HESUPIPE Support"     # 入参原样
    db_brand = DEFAULT_CONFIG["brand"]
    for key in ("title", "whatsapp", "email", "privacy_url"):
        assert out["brand"][key] == db_brand[key]
    assert out["push"]["prefix"] == ""
    assert out["engine"]["context_seed_keys"] == {}
    # 非身份字段保留
    assert out["brand"]["color"] == jc["brand"]["color"]
    assert out["brand"]["welcome"] == jc["brand"]["welcome"]
    assert out["brand"]["topics"] == jc["brand"]["topics"]
    assert out["email"] == jc["email"]
    # wall/prompts 完整可消费
    assert out["wall"] == jc["wall"]
    assert out["prompts"] == jc["prompts"]
    compile_wall(out["wall"])
    PromptProvider.from_config(out["prompts"])


# ---- 校验与冲突 -----------------------------------------------------------------


def test_slug_validation_and_collision(super_client, tmp_path):
    assert _create(super_client, slug="Bad_Slug").status_code == 400
    assert _create(super_client, slug="下划线_x").status_code == 400
    assert _create(super_client, slug="demo").status_code == 201
    r = _create(super_client, slug="demo")
    assert r.status_code == 400                           # 撞现有租户
    assert "demo" in r.json()["detail"]
    # 管理员用户名全局唯一
    r = _create(super_client, slug="other", admin="demo-admin")
    assert r.status_code == 400
    # 未知 profile
    assert _create(super_client, slug="x9", profile="vip").status_code == 400


def test_create_auto_profile(super_client, tmp_path):
    r = _create(super_client, slug="auto2", profile="auto",
                domains=["auto2.example.com"])
    assert r.status_code == 201
    assert _current_snapshot(tmp_path, "auto2") == FileTenantSource(
        SEED_DIR).read_tenant("auto").snapshot
    r = super_client.get("/api/public/brand", headers={"host": "auto2.example.com"})
    assert r.status_code == 200
    assert r.json()["title"] == "HESU Support"            # auto 档案身份


def test_domain_collision_reports_owner(super_client):
    _create(super_client, slug="demo", domains=["demo.example.com"])
    _create(super_client, slug="other")
    # 大写归一后同域 → 400 带现有归属(tenant + site)
    r = super_client.post(f"{TENANTS}/other/sites/main/domains",
                          headers=_auth(super_client),
                          json={"domain": "Demo.Example.com"})
    assert r.status_code == 400
    detail = r.json()["detail"]
    assert "demo" in detail and "main" in detail
    # 创建向导的初始域名同样预检
    r = _create(super_client, slug="third", domains=["demo.example.com"])
    assert r.status_code == 400


def test_delete_domain(super_client):
    _create(super_client, slug="demo",
            domains=["a.example.com", "b.example.com"])
    r = super_client.delete(f"{TENANTS}/demo/sites/main/domains/a.example.com",
                            headers=_auth(super_client))
    assert r.status_code == 200
    assert super_client.get(
        "/api/public/brand", headers={"host": "a.example.com"}).status_code == 404
    assert super_client.get(
        "/api/public/brand", headers={"host": "b.example.com"}).status_code == 200
    # 重复删/不存在 → 404
    assert super_client.delete(
        f"{TENANTS}/demo/sites/main/domains/a.example.com",
        headers=_auth(super_client)).status_code == 404
    assert super_client.delete(
        f"{TENANTS}/demo/sites/main/domains/nope.example.com",
        headers=_auth(super_client)).status_code == 404


# ---- 站点 CRUD / brand_overlay 整赋值 ---------------------------------------------


def test_site_create_patch_overlay(super_client, tmp_path):
    _create(super_client, slug="demo", domains=["demo.example.com"])
    r = super_client.post(f"{TENANTS}/demo/sites", headers=_auth(super_client),
                          json={"slug": "es", "name": "ES 站",
                                "brand_overlay": {"title": "Demo ES",
                                                  "welcome": "Hola"},
                                "domains": ["es.demo.example.com"]})
    assert r.status_code == 201
    # 域名直连该站点(reload 后 resolve 命中站点)
    r = super_client.get("/api/public/brand", headers={"host": "es.demo.example.com"})
    assert r.status_code == 200
    assert r.json()["title"] == "Demo ES"

    # PATCH 改名 + overlay 整字典赋值(M-2a: 换 dict,旧键不残留)
    r = super_client.patch(f"{TENANTS}/demo/sites/es",
                           headers=_auth(super_client),
                           json={"name": "ES2",
                                 "brand_overlay": {"title": "Demo ES v2"}})
    assert r.status_code == 200
    rows = _db_fetch(tmp_path, select(Site)
                     .join(Tenant, Site.tenant_id == Tenant.id)
                     .where(Tenant.slug == "demo", Site.slug == "es"))
    site = rows[0][0]
    assert site.name == "ES2"
    assert site.brand_overlay == {"title": "Demo ES v2"}   # welcome 键没了
    r = super_client.get("/api/public/brand", headers={"host": "es.demo.example.com"})
    assert r.json()["title"] == "Demo ES v2"
    # welcome 回退租户级 brand(覆盖层不再含它)
    assert r.json()["welcome"] != "Hola"

    # 站点 slug 不可改;站点内 slug 撞车 400;未知站点 404
    assert super_client.patch(
        f"{TENANTS}/demo/sites/es", headers=_auth(super_client),
        json={"slug": "es2"}).status_code == 400
    assert super_client.post(
        f"{TENANTS}/demo/sites", headers=_auth(super_client),
        json={"slug": "es"}).status_code == 400
    assert super_client.patch(
        f"{TENANTS}/demo/sites/ghost", headers=_auth(super_client),
        json={"name": "x"}).status_code == 404

    # 站点停用持久化(解析级语义后续任务接线,此处钉字段落库+reload 不炸)
    r = super_client.patch(f"{TENANTS}/demo/sites/es",
                           headers=_auth(super_client),
                           json={"status": "disabled"})
    assert r.status_code == 200
    rows = _db_fetch(tmp_path, select(Site)
                     .join(Tenant, Site.tenant_id == Tenant.id)
                     .where(Tenant.slug == "demo", Site.slug == "es"))
    assert rows[0][0].status == "disabled"


def test_patch_tenant_name_and_validation(super_client, tmp_path):
    _create(super_client, slug="demo", name="原名")
    r = super_client.patch(f"{TENANTS}/demo", headers=_auth(super_client),
                           json={"name": "新名"})
    assert r.status_code == 200
    rows = _db_fetch(tmp_path, select(Tenant).where(Tenant.slug == "demo"))
    assert rows[0][0].name == "新名"
    # 非法 status / 空补丁 / 未知租户
    assert super_client.patch(f"{TENANTS}/demo", headers=_auth(super_client),
                              json={"status": "paused"}).status_code == 400
    assert super_client.patch(f"{TENANTS}/demo", headers=_auth(super_client),
                              json={}).status_code == 400
    assert super_client.patch(f"{TENANTS}/ghost", headers=_auth(super_client),
                              json={"name": "x"}).status_code == 404


# ---- 列表/详情/凭据不外泄/403 边界 -------------------------------------------------


def test_list_tenants(super_client):
    _create(super_client, slug="demo",
            domains=["a.example.com", "b.example.com"])
    _create(super_client, slug="other")
    r = super_client.get(TENANTS, headers=_auth(super_client))
    assert r.status_code == 200
    items = {i["slug"]: i for i in r.json()["items"]}
    assert set(items) == {"demo", "other"}
    assert items["demo"]["status"] == "active"
    assert items["demo"]["domain_count"] == 2
    assert items["demo"]["current_version"] == 1
    assert items["demo"]["admins"] == [
        {"username": "demo-admin", "status": "active"}]
    assert items["other"]["domain_count"] == 0


def test_detail_shape_and_scope(super_client):
    _create(super_client, slug="demo", admin="demo-admin",
            password="demo-pass-1", domains=["demo.example.com"])
    _create(super_client, slug="other")
    r = super_client.get(f"{TENANTS}/demo", headers=_auth(super_client))
    assert r.status_code == 200
    d = r.json()
    assert d["slug"] == "demo"
    assert d["current_version"] == 1
    assert [s["slug"] for s in d["sites"]] == ["main"]
    assert d["sites"][0]["domains"] == ["demo.example.com"]
    assert d["admins"] == [{"username": "demo-admin", "status": "active"}]

    # 详情放宽到本租户管理员;看别的租户 403;列表(超管专属)403
    token = _tenant_admin_token(super_client, "demo-admin", "demo-pass-1")
    assert super_client.get(
        f"{TENANTS}/demo", headers=_bearer(token)).status_code == 200
    assert super_client.get(
        f"{TENANTS}/other", headers=_bearer(token)).status_code == 403
    assert super_client.get(
        TENANTS, headers=_bearer(token)).status_code == 403
    # 未认证 401
    assert super_client.get(f"{TENANTS}/demo").status_code == 401


def test_no_credentials_leak_in_responses(super_client):
    """密码哈希/salt/明文密码绝不出现在列表与详情响应(M-5 红线)。"""
    _create(super_client, slug="demo", admin="demo-admin",
            password="demo-pass-1", domains=["demo.example.com"])
    for r in (super_client.get(TENANTS, headers=_auth(super_client)),
              super_client.get(f"{TENANTS}/demo", headers=_auth(super_client))):
        assert r.status_code == 200
        for needle in ("password_hash", "salt", "scrypt", "demo-pass-1"):
            assert needle not in r.text, f"响应泄露凭据字段: {needle}"