From ec2e17d5e3152bf93a955468cefb1779cbefc2b0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Tue, 13 Oct 2026 09:20:28 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=A7=9F=E6=88=B7=E7=AE=A1=E7=90=86API?= =?UTF-8?q?=EF=BC=88=E5=88=9B=E5=BB=BA=E5=90=91=E5=AF=BC=E6=B6=88=E8=B4=B9?= =?UTF-8?q?=E5=87=BA=E5=8E=82=E6=A1=A3=E6=A1=88/blank=E4=BF=9D=E7=95=99jc?= =?UTF-8?q?=E5=A2=99/=E5=9F=9F=E5=90=8D/=E5=81=9C=E7=94=A8/slug=E4=B8=8D?= =?UTF-8?q?=E5=8F=AF=E5=8F=98=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /api/platform/tenants: profile ∈ {auto,jiancai,blank}——首版快照复用 FileTenantSource(seed).read_tenant(profile).snapshot(Task 4 审查指定); blank=jiancai 档案 _blankify 去身份组但保留 wall/prompts(compile_wall/ PromptProvider 的硬要求)。建 Tenant+main 站点+版本1+租户管理员 (must_change_password=True),创建后 registry_reload(upsert)立即可用。 站点 CRUD(brand_overlay 整字典赋值,M-2a)/域名绑定解绑(全局唯一, 撞车 400 带现有归属)/PATCH 租户(name/status;slug 不可变 400;停用走 reload 路径 dispose 引擎)。列表/详情管理员概要显式挑字段,凭据永不外泄。 鉴权: 写操作+列表 require_super;GET 详情放宽 require_tenant(slug)。 测试: db_source_app 夹具(registry 用 DbTenantSource 生产形态指向临时 platform.db,跨循环纪律: 直查走独立 NullPool 连接),16 项全绿,全量 519。 Co-Authored-By: Claude Code --- app/main.py | 4 + app/platform/routes/tenant_routes.py | 516 +++++++++++++++++++++++++++ tests/test_tenant_admin_api.py | 504 ++++++++++++++++++++++++++ 3 files changed, 1024 insertions(+) create mode 100644 app/platform/routes/tenant_routes.py create mode 100644 tests/test_tenant_admin_api.py diff --git a/app/main.py b/app/main.py index 3acb72c..5a0a1da 100644 --- a/app/main.py +++ b/app/main.py @@ -81,9 +81,13 @@ from app.engine.routes.chat import router as chat_router from app.engine.routes.chat_rest import router as chat_rest_router from app.engine.routes.public import router as public_router from app.platform.routes.auth_routes import router as platform_auth_router +from app.platform.routes.tenant_routes import router as platform_tenant_router app.include_router(chat_router) app.include_router(chat_rest_router) app.include_router(public_router) +# 租户管理 API(Task 5):创建向导/站点/域名/停用——同样只依赖路由层认证 +# (require_super / require_tenant),Host 豁免面归 Task 3 的前缀规则覆盖。 +app.include_router(platform_tenant_router) # 平台认证路由(Task 2):本任务即挂——已知租户 Host(如 jiancai)下 # /api/platform/auth/* 经 Host 中间件直达,TDD 循环闭环。Task 3 将为 # /api/platform/ 前缀加中间件豁免(平台后台不依赖租户域名解析); diff --git a/app/platform/routes/tenant_routes.py b/app/platform/routes/tenant_routes.py new file mode 100644 index 0000000..02a7417 --- /dev/null +++ b/app/platform/routes/tenant_routes.py @@ -0,0 +1,516 @@ +"""租户管理 API(Phase 2 Task 5):/api/platform/tenants*——超管的创建向导/ +站点/域名/停用,GET 详情放宽到本租户管理员。 + +创建向导的快照复用路径(Task 4 审查指定,勿手搓拼装): +``FileTenantSource(SEED_DIR).read_tenant(profile).snapshot`` 直接产出首版 +快照——auto/jiancai 原样入版本1;**blank** = jiancai 档案经 _blankify 去 +身份组(brand 的 title/whatsapp/email/privacy_url、push.prefix、 +engine.context_seed_keys 置出厂默认),**wall 与 prompts 组保留**(取 +jiancai 档案)——compile_wall 要求 wall 全字段、PromptProvider 要求四件套, +DEFAULT_CONFIG 满足不了。 + +写操作后一律 registry_reload(slug)(Task 4 upsert 语义:新建租户未加载 +也直接插入,创建即可用);停用走 reload 路径顺带 dispose 旧 engine +(引擎释放由 registry 负责,本层不碰)。 + +鉴权:写操作与列表 require_super;GET /{slug} 详情放宽 require_tenant +(slug)(租户管理员看自己概要,403 边界有测试钉住)。 + +安全纪律: +- 密码哈希/salt 绝不进响应——管理员序列化显式挑 {username, status} + (db.list_admin_users 的 M-5 警示); +- 创建出的租户管理员一律 must_change_password=True(引导密码首登改掉)。 + +M-2a 陷阱:Site.brand_overlay 是 ORM JSON 列——写时**整字典赋值*,不做 +按键局部更新(局部改不标记脏,静默丢写)。 +""" +import copy +import json +import pathlib +import re + +from fastapi import APIRouter, Depends, HTTPException, Request +from pydantic import BaseModel, Field +from sqlalchemy import func, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from app.platform.auth import ( + UserContext, + hash_password, + platform_user, + require_super, + require_tenant, +) +from app.platform.db import AdminUser, Site, SiteDomain, Tenant, TenantConfigVersion +from app.platform.defaults import DEFAULT_CONFIG +from app.platform.registry import _normalize_host +from app.platform.tenant_source import FileTenantSource + +router = APIRouter(prefix="/api/platform/tenants", tags=["platform-tenants"]) + +# 出厂档案目录:与 app.main.SEED_DIR 同一锚定规则(仓库根/seed_profiles), +# 但不 import app.main——main 模块级挂路由,反向导入制造循环依赖。 +SEED_DIR = pathlib.Path(__file__).resolve().parents[3] / "seed_profiles" + +PROFILES = ("auto", "jiancai", "blank") +STATUSES = ("active", "disabled") +SLUG_RE = re.compile(r"^[a-z0-9-]+$") + +# blank 身份组清理清单(brand 的这几个键置出厂默认;其余 brand 字段保留) +_IDENTITY_BRAND_KEYS = ("title", "whatsapp", "email", "privacy_url") + + +def _blankify(snapshot: dict) -> dict: + """出厂档案快照 → blank 形态:身份组置出厂默认,其余(含 wall/prompts) + 原样保留。深拷贝后改——纯函数纪律,绝不动入参。""" + out = copy.deepcopy(snapshot) + brand = dict(out.get("brand") or {}) + for key in _IDENTITY_BRAND_KEYS: + brand[key] = DEFAULT_CONFIG["brand"][key] + out["brand"] = brand + push = dict(out.get("push") or {}) + push["prefix"] = DEFAULT_CONFIG["push"]["prefix"] + out["push"] = push + engine = dict(out.get("engine") or {}) + engine["context_seed_keys"] = dict( + DEFAULT_CONFIG["engine"]["context_seed_keys"]) + out["engine"] = engine + return out + + +def _profile_snapshot(profile: str) -> dict: + """profile → 首版快照:auto/jiancai 读对应出厂档案原样;blank 读 jiancai + 档案再去身份组(wall/prompts 保留——见模块 docstring)。""" + base_slug = "jiancai" if profile == "blank" else profile + snapshot = FileTenantSource(SEED_DIR).read_tenant(base_slug).snapshot + return _blankify(snapshot) if profile == "blank" else snapshot + + +# ---- 校验助手 ------------------------------------------------------------------- + + +def _validate_slug(slug: str) -> str: + if not SLUG_RE.fullmatch(slug or ""): + raise HTTPException( + status_code=400, detail="slug 只允许小写字母、数字和连字符") + return slug + + +def _normalize_domain(domain: str) -> str: + """入库前归一(小写/去端口/去尾点——与 resolve 的 host 归一同一事实源); + 归一后为空 → 400。""" + normalized = _normalize_host(domain or "") + if not normalized: + raise HTTPException(status_code=400, detail="域名不能为空") + return normalized + + +def _check_name(name) -> None: + if not isinstance(name, str) or not 1 <= len(name) <= 128: + raise HTTPException(status_code=400, detail="name 须为 1-128 字符") + + +def _check_status(status) -> None: + if status not in STATUSES: + raise HTTPException( + status_code=400, + detail=f"status 只允许 {STATUSES}(实际 {status!r})") + + +def _patch_fields(body: dict, *, allow: tuple[str, ...]) -> dict: + """PATCH 通用校验:slug 键一律 400(不可变);白名单外字段 400;显式 + null 视为缺席;全空 → 400(无意义补丁)。返回只含出现且非 null 的字段。""" + if "slug" in body: + raise HTTPException(status_code=400, detail="slug 不可修改") + unknown = set(body) - set(allow) + if unknown: + raise HTTPException( + status_code=400, + detail=f"不可修改的字段: {', '.join(sorted(unknown))}") + fields = {k: v for k, v in body.items() if v is not None} + if not fields: + raise HTTPException(status_code=400, detail="没有可更新字段") + return fields + + +async def _get_tenant_or_404(session: AsyncSession, slug: str) -> Tenant: + tenant = (await session.execute( + select(Tenant).where(Tenant.slug == slug))).scalar_one_or_none() + if tenant is None: + raise HTTPException(status_code=404, detail=f"租户不存在: {slug!r}") + return tenant + + +async def _get_site_or_404( + session: AsyncSession, tenant_slug: str, site_slug: str) -> Site: + site = (await session.execute( + select(Site).join(Tenant, Site.tenant_id == Tenant.id) + .where(Tenant.slug == tenant_slug, Site.slug == site_slug) + )).scalar_one_or_none() + if site is None: + raise HTTPException( + status_code=404, + detail=f"站点不存在: {tenant_slug}/{site_slug}") + return site + + +async def _assert_domain_free(session: AsyncSession, domain: str) -> None: + """域名全局唯一预检:撞车 400 带现有归属(租户+站点)——调用方能一眼 + 看到域名被谁占了。""" + row = (await session.execute( + select(SiteDomain.domain, Tenant.slug, Site.slug) + .join(Site, SiteDomain.site_id == Site.id) + .join(Tenant, Site.tenant_id == Tenant.id) + .where(SiteDomain.domain == domain))).first() + if row is not None: + raise HTTPException( + status_code=400, + detail=f"域名 {domain} 已绑定租户 {row[1]!r} 的站点 {row[2]!r}") + + +async def _site_summary(session: AsyncSession, site: Site) -> dict: + domains = (await session.execute( + select(SiteDomain.domain).where(SiteDomain.site_id == site.id) + .order_by(SiteDomain.id))).scalars().all() + return { + "slug": site.slug, "name": site.name, "status": site.status, + "domains": list(domains), + "brand_overlay": dict(site.brand_overlay or {}), + } + + +def _admin_summary(admin: AdminUser) -> dict: + # 显式挑字段——password_hash/salt 绝不进响应(M-5) + return {"username": admin.username, "status": admin.status} + + +def _tenant_guard(slug: str, user: platform_user) -> UserContext: + """路径参数版 require_tenant:复用 auth 的依赖工厂,slug 取自路径 + (super 或本租户管理员放行,409 门经 platform_user 连锁)。""" + return require_tenant(slug)(user) + + +# ---- 请求模型 ------------------------------------------------------------------- + + +class CreateTenantRequest(BaseModel): + name: str = Field(min_length=1, max_length=128) + slug: str = Field(min_length=1, max_length=64) + profile: str + admin_username: str = Field(min_length=1, max_length=64) + admin_password: str = Field(min_length=1, max_length=128) + domains: list[str] = Field(default_factory=list) + + +class CreateSiteRequest(BaseModel): + slug: str = Field(min_length=1, max_length=64) + name: str = Field(default="", max_length=128) + brand_overlay: dict = Field(default_factory=dict) + domains: list[str] = Field(default_factory=list) + + +class AddDomainRequest(BaseModel): + domain: str = Field(min_length=1, max_length=255) + + +# ---- 租户:创建 / 列表 / 详情 / 改名改状态 ------------------------------------------ + + +@router.post("", status_code=201) +async def create_tenant( + body: CreateTenantRequest, request: Request, + user=Depends(require_super)): + """创建向导:校验 slug/profile → 出厂档案快照 → 建租户+main 站点+ + 版本1+租户管理员(must_change_password=True)→ registry_reload(upsert, + 创建即可用)。""" + _validate_slug(body.slug) + if body.profile not in PROFILES: + raise HTTPException( + status_code=400, + detail=f"未知 profile: {body.profile!r}(可选 {PROFILES})") + # 请求内去重(保序),免自撞唯一约束 + domains = list(dict.fromkeys( + _normalize_domain(d) for d in body.domains)) + try: + snapshot = _profile_snapshot(body.profile) + except LookupError as exc: + raise HTTPException( + status_code=400, detail=f"出厂档案缺失: {exc}") from exc + + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + if (await session.execute( + select(Tenant.id).where(Tenant.slug == body.slug))).first(): + raise HTTPException( + status_code=400, detail=f"slug 已存在: {body.slug!r}") + if (await session.execute( + select(AdminUser.id).where( + AdminUser.username == body.admin_username))).first(): + raise HTTPException( + status_code=400, + detail=f"管理员用户名已被占用: {body.admin_username!r}") + for domain in domains: + await _assert_domain_free(session, domain) + + tenant = Tenant(slug=body.slug, name=body.name, status="active") + session.add(tenant) + await session.flush() + site = Site(tenant_id=tenant.id, slug="main", name="main") + session.add(site) + await session.flush() + for domain in domains: + session.add(SiteDomain(domain=domain, site_id=site.id)) + password_hash, salt = hash_password(body.admin_password) + session.add(AdminUser( + tenant_id=tenant.id, + username=body.admin_username, + password_hash=password_hash, + salt=salt, + must_change_password=True)) + session.add(TenantConfigVersion( + tenant_id=tenant.id, + version=1, + config_json=json.dumps(snapshot, ensure_ascii=False), + updated_by=f"super#{user.user_id}", + comment=f"创建向导: profile={body.profile}")) + try: + await session.commit() + except IntegrityError as exc: + # 并发竞态兜底(预检与提交之间被人抢先):唯一约束落库层兜住 + await session.rollback() + raise HTTPException( + status_code=409, + detail="并发冲突: slug/用户名/域名已被占用") from exc + + await request.app.state.registry.registry_reload(body.slug) + return { + "slug": tenant.slug, "name": tenant.name, "status": tenant.status, + "profile": body.profile, "version": 1, + "site": {"slug": "main", "domains": domains}, + "admin_username": body.admin_username, + } + + +@router.get("") +async def list_tenants(request: Request, user=Depends(require_super)): + """租户列表:slug/name/status/域名数/当前版本号/管理员概要。""" + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + tenants = (await session.execute( + select(Tenant).order_by(Tenant.slug))).scalars().all() + domain_counts = dict((await session.execute( + select(Site.tenant_id, func.count(SiteDomain.id)) + .join(SiteDomain, SiteDomain.site_id == Site.id) + .group_by(Site.tenant_id))).all()) + versions = dict((await session.execute( + select(TenantConfigVersion.tenant_id, + func.max(TenantConfigVersion.version)) + .group_by(TenantConfigVersion.tenant_id))).all()) + admins = (await session.execute( + select(AdminUser).where(AdminUser.tenant_id.is_not(None)) + .order_by(AdminUser.id))).scalars().all() + + admins_by_tenant: dict[int, list[dict]] = {} + for admin in admins: + admins_by_tenant.setdefault(admin.tenant_id, []).append( + _admin_summary(admin)) + return {"items": [{ + "slug": t.slug, + "name": t.name, + "status": t.status, + "domain_count": domain_counts.get(t.id, 0), + "current_version": versions.get(t.id), + "admins": admins_by_tenant.get(t.id, []), + } for t in tenants]} + + +@router.get("/{slug}") +async def tenant_detail( + slug: str, request: Request, user=Depends(_tenant_guard)): + """租户详情:sites(含 domains/brand_overlay)/管理员用户名/当前版本。 + 鉴权放宽 require_tenant(slug)——租户管理员看自己概要。""" + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + tenant = await _get_tenant_or_404(session, slug) + sites = (await session.execute( + select(Site).where(Site.tenant_id == tenant.id) + .order_by(Site.id))).scalars().all() + domains = (await session.execute( + select(SiteDomain) + .where(SiteDomain.site_id.in_([s.id for s in sites])) + .order_by(SiteDomain.id))).scalars().all() + current_version = (await session.execute( + select(func.max(TenantConfigVersion.version)) + .where(TenantConfigVersion.tenant_id == tenant.id))).scalar() + admins = (await session.execute( + select(AdminUser).where(AdminUser.tenant_id == tenant.id) + .order_by(AdminUser.id))).scalars().all() + + domains_by_site: dict[int, list[str]] = {} + for d in domains: + domains_by_site.setdefault(d.site_id, []).append(d.domain) + return { + "slug": tenant.slug, + "name": tenant.name, + "status": tenant.status, + "current_version": current_version, + "sites": [{ + "slug": s.slug, "name": s.name, "status": s.status, + "domains": list(domains_by_site.get(s.id, [])), + "brand_overlay": dict(s.brand_overlay or {}), + } for s in sites], + "admins": [_admin_summary(a) for a in admins], + } + + +@router.patch("/{slug}") +async def patch_tenant( + slug: str, body: dict, request: Request, user=Depends(require_super)): + """改名/停用复启用。slug 不可变(400);status≠active → registry reload + 后 resolve 抛 TenantDisabledError,引擎释放由 reload 路径处理(dispose)。""" + fields = _patch_fields(body, allow=("name", "status")) + if "name" in fields: + _check_name(fields["name"]) + if "status" in fields: + _check_status(fields["status"]) + + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + tenant = await _get_tenant_or_404(session, slug) + if "name" in fields: + tenant.name = fields["name"] + if "status" in fields: + tenant.status = fields["status"] + await session.commit() + result = {"slug": tenant.slug, "name": tenant.name, + "status": tenant.status} + + await request.app.state.registry.registry_reload(slug) + return result + + +# ---- 站点:建站 / 改门面 ----------------------------------------------------------- + + +@router.post("/{slug}/sites", status_code=201) +async def create_site( + slug: str, body: CreateSiteRequest, request: Request, + user=Depends(require_super)): + """建站:slug/name/brand_overlay/domains 可选。brand_overlay 是 ORM + JSON 列——整字典赋值(M-2a)。""" + _validate_slug(body.slug) + domains = list(dict.fromkeys( + _normalize_domain(d) for d in body.domains)) + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + tenant = await _get_tenant_or_404(session, slug) + if (await session.execute( + select(Site.id).where(Site.tenant_id == tenant.id, + Site.slug == body.slug))).first(): + raise HTTPException( + status_code=400, + detail=f"站点 slug 已存在: {body.slug!r}") + for domain in domains: + await _assert_domain_free(session, domain) + + site = Site(tenant_id=tenant.id, slug=body.slug, + name=body.name or body.slug, + brand_overlay=dict(body.brand_overlay)) + session.add(site) + await session.flush() + for domain in domains: + session.add(SiteDomain(domain=domain, site_id=site.id)) + try: + await session.commit() + except IntegrityError as exc: + await session.rollback() + raise HTTPException( + status_code=409, + detail="并发冲突: 站点 slug 或域名已被占用") from exc + result = await _site_summary(session, site) + + await request.app.state.registry.registry_reload(slug) + return result + + +@router.patch("/{slug}/sites/{site_slug}") +async def patch_site( + slug: str, site_slug: str, body: dict, request: Request, + user=Depends(require_super)): + """改站点:停用/改名/改 overlay(整字典赋值,旧键不残留)。站点 slug + 同样不可变(400)。""" + fields = _patch_fields(body, allow=("name", "status", "brand_overlay")) + if "name" in fields: + _check_name(fields["name"]) + if "status" in fields: + _check_status(fields["status"]) + if "brand_overlay" in fields and not isinstance( + fields["brand_overlay"], dict): + raise HTTPException( + status_code=400, detail="brand_overlay 必须是 JSON 对象") + + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + site = await _get_site_or_404(session, slug, site_slug) + if "name" in fields: + site.name = fields["name"] + if "status" in fields: + site.status = fields["status"] + if "brand_overlay" in fields: + site.brand_overlay = dict(fields["brand_overlay"]) # M-2a 整赋值 + await session.commit() + result = await _site_summary(session, site) + + await request.app.state.registry.registry_reload(slug) + return result + + +# ---- 域名:绑定 / 解绑 ------------------------------------------------------------ + + +@router.post("/{slug}/sites/{site_slug}/domains", status_code=201) +async def add_domain( + slug: str, site_slug: str, body: AddDomainRequest, request: Request, + user=Depends(require_super)): + """域名绑定:全局唯一(撞车 400 带现有归属);绑定后 registry reload + 生效。""" + domain = _normalize_domain(body.domain) + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + site = await _get_site_or_404(session, slug, site_slug) + await _assert_domain_free(session, domain) + session.add(SiteDomain(domain=domain, site_id=site.id)) + try: + await session.commit() + except IntegrityError as exc: + await session.rollback() + raise HTTPException( + status_code=409, + detail=f"并发冲突: 域名 {domain} 已被占用") from exc + + await request.app.state.registry.registry_reload(slug) + return {"domain": domain, "site": site_slug} + + +@router.delete("/{slug}/sites/{site_slug}/domains/{domain}") +async def delete_domain( + slug: str, site_slug: str, domain: str, request: Request, + user=Depends(require_super)): + """域名解绑:只解本站点的该域名(不存在 404);解绑后 reload 生效。""" + domain = _normalize_domain(domain) + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + site = await _get_site_or_404(session, slug, site_slug) + row = (await session.execute( + select(SiteDomain).where(SiteDomain.site_id == site.id, + SiteDomain.domain == domain) + )).scalar_one_or_none() + if row is None: + raise HTTPException( + status_code=404, detail=f"域名不存在: {domain!r}") + await session.delete(row) + await session.commit() + + await request.app.state.registry.registry_reload(slug) + return {"ok": True, "domain": domain, "site": site_slug} diff --git a/tests/test_tenant_admin_api.py b/tests/test_tenant_admin_api.py new file mode 100644 index 0000000..4109f52 --- /dev/null +++ b/tests/test_tenant_admin_api.py @@ -0,0 +1,504 @@ +"""租户管理 API(Phase 2 Task 5):超管的创建向导/站点/域名/停用 + 403 边界。 + +db_source_app 夹具: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.main import app +from app.platform.db import ( + AdminUser, + Site, + SiteDomain, + Tenant, + TenantConfigVersion, + make_platform_engine, +) +from app.platform.defaults import DEFAULT_CONFIG +from app.platform.registry import TenantDisabledError, TenantRegistry +from app.platform.routes.tenant_routes import _blankify +from app.platform.tenant_source import DbTenantSource, FileTenantSource +from tests.conftest import SEED_DIR, app_test_client + +TENANTS = "/api/platform/tenants" +LOGIN = "/api/platform/auth/login" +CHANGE_PW = "/api/platform/auth/change-password" + + +# ---- 夹具 ---------------------------------------------------------------------- + + +@pytest.fixture +def db_source_app(tmp_path): + """app + 生产形态 registry(DbTenantSource 指向临时 platform.db)。 + + platform_engine 也指同一临时库(lifespan 在 portal 循环里 ensure_schema + + bootstrap 超管 admin/admin123)。teardown 复位 app.state(照抄 + app_test_client 的关残留窗纪律;engine dispose 由 lifespan finally)。 + """ + engine = make_platform_engine(tmp_path / "platform.db") + app.state.registry = TenantRegistry( + DbTenantSource(engine), data_dir=tmp_path / "tenants") + app.state.platform_engine = engine + try: + yield app + finally: + app.state.registry = None + app.state.platform_engine = None + + +@pytest.fixture +def super_client(db_source_app, tmp_path): + """超管 client:lifespan bootstrap 的 admin/admin123 登录 + 首登改密 + (过 must_change 409 门)。token 挂 client.platform_token。""" + with app_test_client(db_source_app, "http://testserver", tmp_path) as c: + r = c.post(LOGIN, json={"username": "admin", "password": "admin123"}) + assert r.status_code == 200 + c.platform_token = r.json()["token"] + r = c.post(CHANGE_PW, headers=_auth(c), json={ + "old_password": "admin123", "new_password": "sup3r-pass-1"}) + assert r.status_code == 200 + yield c + + +# ---- 助手 ---------------------------------------------------------------------- + + +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}"