diff --git a/app/main.py b/app/main.py index 5a0a1da..5aedb98 100644 --- a/app/main.py +++ b/app/main.py @@ -81,6 +81,10 @@ 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.config_routes import ( + router as platform_config_router, + settings_router as platform_settings_router, +) from app.platform.routes.tenant_routes import router as platform_tenant_router app.include_router(chat_router) app.include_router(chat_rest_router) @@ -88,6 +92,11 @@ app.include_router(public_router) # 租户管理 API(Task 5):创建向导/站点/域名/停用——同样只依赖路由层认证 # (require_super / require_tenant),Host 豁免面归 Task 3 的前缀规则覆盖。 app.include_router(platform_tenant_router) +# 配置 API(Task 6):分组校验管线/整份版本/回滚/试金石/重置出厂(鉴权 +# require_tenant,租户全开)+ 平台设置(require_super)——/api/platform/ +# 前缀的 Host 豁免由 Task 3 的中间件规则统一覆盖。 +app.include_router(platform_config_router) +app.include_router(platform_settings_router) # 平台认证路由(Task 2):本任务即挂——已知租户 Host(如 jiancai)下 # /api/platform/auth/* 经 Host 中间件直达,TDD 循环闭环。Task 3 将为 # /api/platform/ 前缀加中间件豁免(平台后台不依赖租户域名解析); diff --git a/app/platform/config_validation.py b/app/platform/config_validation.py new file mode 100644 index 0000000..4a3b4de --- /dev/null +++ b/app/platform/config_validation.py @@ -0,0 +1,256 @@ +"""配置校验管线(Phase 2 Task 6)——PUT /config/{group} 的分组校验。 + +validate_group(group, payload, context) -> list[str]:错误清单(中文,带 +字段路径),空清单 = 通过;路由层把非空清单拼进 400 detail。 + +纪律: +- **键 allowlist**:各组未知键显性拒绝(与 wall_schema._KNOWN_FIELDS 同款 + 防呆——配置编辑器时代拼错键 = 深合并静默忽略,这里前移到保存期报错)。 + wall 组例外:未知键交 compile_wall 管(它的报错自带 wall 侧字段路径)。 +- **单一事实源**:各组合法键从 DEFAULT_CONFIG 派生(prompts 除外—— + DEFAULT 不含 prompts,取 ai_service.PROMPTS_KEYS 四件套)。 +- **wall 组不重复造校验**:深拷贝 payload 交 compile_wall,复用全部既有 + 校验(键 allowlist/占位符白名单/flags/正则编译/必填池)。 +- **prompts 组**:四键字符串 + 大小上限(company_facts ≤50KB 软上限, + spec §5.7——每轮整段拼入 system prompt,过大门线性放大 LLM 成本; + 其余三件 ≤100KB)。字节口径按 UTF-8 编码长度。 +- **push.prefix 跨租户查重**(spec E1,external_key 前缀全局唯一防 CRM + 线索空间串号):context.other_prefixes 由路由层从 DB 各租户当前版本 + 收集(排除自身;空串 = 未设置,不参战——出厂默认本就多租户同空)。 +- 数值字段(None = "未设置"哨兵,与 DEFAULT_CONFIG 同款纪律):None 或 + 范围内数值;bool 不是 int(isinstance 陷阱,显式排除)。 +""" +import copy +from dataclasses import dataclass, field + +from app.engine.services.ai_service import PROMPTS_KEYS +from app.engine.wall_schema import WallConfigError, compile_wall +from app.platform.defaults import DEFAULT_CONFIG + +# 各组合法键 = DEFAULT_CONFIG 同组键(单一事实源);prompts 特例见上。 +_ALLOWED_KEYS: dict[str, frozenset] = { + group: frozenset(keys) for group, keys in DEFAULT_CONFIG.items()} +_ALLOWED_KEYS["prompts"] = frozenset(PROMPTS_KEYS) + +# 合法组名(排序输出进错误信息,稳定可测) +CONFIG_GROUPS = frozenset(_ALLOWED_KEYS) + +# prompts 大小上限(字节,UTF-8 口径):company_facts 50KB 软上限,其余 100KB +PROMPTS_SIZE_CAPS = { + name: 50 * 1024 if name == "company_facts" else 100 * 1024 + for name in PROMPTS_KEYS} + +_BRAND_STR_KEYS = ("title", "color", "avatar", "welcome", "placeholder", + "whatsapp", "email", "privacy_url") + + +@dataclass(frozen=True) +class ValidationContext: + """跨租户校验上下文(路由层按请求现势收集,校验管线不自查 DB)。""" + other_prefixes: frozenset = field(default_factory=frozenset) + + +def _type_name(value) -> str: + return type(value).__name__ + + +def _want_str(value, path: str, errors: list) -> None: + if not isinstance(value, str): + errors.append(f"{path}: 须为字符串(实际 {_type_name(value)})") + + +def _want_bool(value, path: str, errors: list) -> None: + if not isinstance(value, bool): + errors.append(f"{path}: 须为布尔(实际 {_type_name(value)})") + + +def _want_int(value, path: str, errors: list) -> None: + # bool 是 int 的子类——True 会被 isinstance 放过,显式排除 + if not isinstance(value, int) or isinstance(value, bool): + errors.append(f"{path}: 须为整数(实际 {_type_name(value)})") + + +def _want_str_list(value, path: str, errors: list) -> None: + if not isinstance(value, list): + errors.append(f"{path}: 须为字符串列表(实际 {_type_name(value)})") + return + for i, item in enumerate(value): + if not isinstance(item, str): + errors.append(f"{path}[{i}]: 项须为字符串(实际 {_type_name(item)})") + + +def _check_brand(payload: dict, ctx: ValidationContext, errors: list) -> None: + for key in _BRAND_STR_KEYS: + if key in payload: + _want_str(payload[key], f"brand.{key}", errors) + if "topics" in payload: + _want_str_list(payload["topics"], "brand.topics", errors) + + +def _check_wall(payload: dict, ctx: ValidationContext, errors: list) -> None: + """全量交 compile_wall(深拷贝防意外 mutation):键 allowlist/占位符/ + flags/正则编译/必填池全套复用,WallConfigError 转清单项。""" + try: + compile_wall(copy.deepcopy(payload)) + except WallConfigError as exc: + errors.append(str(exc)) + + +def _check_prompts(payload: dict, ctx: ValidationContext, + errors: list) -> None: + for key in PROMPTS_KEYS: + if key not in payload: + errors.append(f"prompts.{key}: 缺少必填键(四件套须整组提交)") + continue + text = payload[key] + if not isinstance(text, str): + errors.append(f"prompts.{key}: 须为字符串(实际 {_type_name(text)})") + continue + size = len(text.encode("utf-8")) + cap = PROMPTS_SIZE_CAPS[key] + if size > cap: + errors.append( + f"prompts.{key}: 内容 {size} 字节超过上限 {cap} 字节" + f"({cap // 1024}KB,spec §5.7——每轮拼入 system prompt," + f"请精炼)") + + +def _check_contact(payload: dict, ctx: ValidationContext, + errors: list) -> None: + for key in ("cc_completion", "email_light_ask"): + if key in payload: + _want_bool(payload[key], f"contact.{key}", errors) + if "template_languages" in payload: + _want_str_list(payload["template_languages"], + "contact.template_languages", errors) + + +def _check_email(payload: dict, ctx: ValidationContext, errors: list) -> None: + if "sender_name" in payload: + _want_str(payload["sender_name"], "email.sender_name", errors) + if "smtp" in payload and not (payload["smtp"] is None + or isinstance(payload["smtp"], dict)): + errors.append(f"email.smtp: 须为映射或 null(实际 " + f"{_type_name(payload['smtp'])})") + if "notify_emails" in payload: + value = payload["notify_emails"] + if not isinstance(value, list): + errors.append(f"email.notify_emails: 须为列表(实际 " + f"{_type_name(value)})") + return + for i, item in enumerate(value): + if not isinstance(item, str) or "@" not in item: + errors.append(f"email.notify_emails[{i}]: 项须为含 @ 的邮箱" + f"地址(实际 {item!r})") + + +def _check_push(payload: dict, ctx: ValidationContext, errors: list) -> None: + if "enabled" in payload: + _want_bool(payload["enabled"], "push.enabled", errors) + for key in ("url", "api_key", "prefix"): + if key in payload: + _want_str(payload[key], f"push.{key}", errors) + if "timeout" in payload: + _want_positive_number(payload["timeout"], "push.timeout", errors) + if "retry_backoff_base" in payload: + _want_positive_number(payload["retry_backoff_base"], + "push.retry_backoff_base", errors) + if "max_attempts" in payload: + value = payload["max_attempts"] + if value is not None and (not isinstance(value, int) + or isinstance(value, bool) + or not 1 <= value <= 10): + errors.append(f"push.max_attempts: 须为 1-10 整数或 null(实际 " + f"{value!r})") + # spec E1: external_key 前缀全局唯一(空串 = 未设置,不参战) + prefix = payload.get("prefix") + if isinstance(prefix, str) and prefix and prefix in ctx.other_prefixes: + errors.append(f"push.prefix: {prefix!r} 已被其他租户占用" + f"(external_key 前缀全局唯一)") + + +def _want_positive_number(value, path: str, errors: list) -> None: + if value is None: + return + if (not isinstance(value, (int, float)) or isinstance(value, bool) + or value <= 0): + errors.append(f"{path}: 须为正数或 null(实际 {value!r})") + + +def _check_ai(payload: dict, ctx: ValidationContext, errors: list) -> None: + for key in ("provider", "model"): + if key in payload: + _want_str(payload[key], f"ai.{key}", errors) + for key in ("api_key", "base_url"): + if key in payload: + value = payload[key] + if value is not None and not isinstance(value, str): + errors.append(f"ai.{key}: 须为字符串或 null(实际 " + f"{_type_name(value)})") + + +def _check_engine(payload: dict, ctx: ValidationContext, + errors: list) -> None: + for key in ("rate_limit_per_hour", "ws_conn_limit"): + if key in payload: + value = payload[key] + _want_int(value, f"engine.{key}", errors) + if isinstance(value, int) and not isinstance(value, bool) \ + and value < 1: + errors.append(f"engine.{key}: 须为正整数(实际 {value})") + if "context_seed_keys" in payload: + value = payload["context_seed_keys"] + if not isinstance(value, dict): + errors.append(f"engine.context_seed_keys: 须为映射(实际 " + f"{_type_name(value)})") + return + for k, v in value.items(): + if not isinstance(v, str): + errors.append(f"engine.context_seed_keys.{k}: 须为字符串" + f"(实际 {_type_name(v)})") + + +def _check_misc(payload: dict, ctx: ValidationContext, errors: list) -> None: + if "name_to_region_extras" in payload: + value = payload["name_to_region_extras"] + if not isinstance(value, dict): + errors.append(f"misc.name_to_region_extras: 须为映射(实际 " + f"{_type_name(value)})") + return + for k, v in value.items(): + if not isinstance(v, str): + errors.append(f"misc.name_to_region_extras.{k}: 须为字符串" + f"(实际 {_type_name(v)})") + + +_GROUP_CHECKERS = { + "brand": _check_brand, + "wall": _check_wall, + "contact": _check_contact, + "email": _check_email, + "push": _check_push, + "ai": _check_ai, + "engine": _check_engine, + "misc": _check_misc, + "prompts": _check_prompts, +} + + +def validate_group(group: str, payload, context: ValidationContext) -> list: + """校验单个配置组的 payload(整组替换语义:只查出现键的类型/形状, + 缺省键由消费侧深合并出厂默认补)。返回错误清单(中文,带字段路径), + 空清单 = 通过。未知组名/非对象 payload 也走清单(路由统一转 400)。""" + checker = _GROUP_CHECKERS.get(group) + if checker is None: + return [f"未知配置组: {group!r}(合法: " + f"{', '.join(sorted(CONFIG_GROUPS))})"] + if not isinstance(payload, dict): + return [f"{group}: 须为 JSON 对象(实际 {_type_name(payload)})"] + errors: list = [] + if group != "wall": # wall 的未知键交 compile_wall(报错自带其字段路径) + unknown = set(payload) - _ALLOWED_KEYS[group] + if unknown: + errors.append(f"{group}: 未知键 {', '.join(sorted(unknown))}" + f"(合法: {', '.join(sorted(_ALLOWED_KEYS[group]))})") + checker(payload, context, errors) + return errors diff --git a/app/platform/db.py b/app/platform/db.py index e98ede8..a288332 100644 --- a/app/platform/db.py +++ b/app/platform/db.py @@ -26,6 +26,7 @@ PlatformBase.metadata,engine/models.ensure_schema 只跑 Base.metadata—— (Windows 文件锁,同 registry)。 """ import datetime +import json import pathlib from sqlalchemy import ( @@ -38,6 +39,7 @@ from sqlalchemy import ( Text, UniqueConstraint, event, + func, select, ) from sqlalchemy.exc import IntegrityError @@ -69,6 +71,10 @@ class Tenant(PlatformBase): slug: Mapped[str] = mapped_column(String(64), unique=True, index=True) name: Mapped[str] = mapped_column(String(128)) status: Mapped[str] = mapped_column(String(16), default="active") + # 创建向导选定的出厂档案(Task 6 起):重置出厂按它取对应档案组 + # (auto/jiancai 档案 / blank = jiancai 去身份组)。缺省 blank—— + # 正常路径一律经创建向导写入,缺省只是手工插行的兜底语义。 + profile: Mapped[str] = mapped_column(String(16), default="blank") created_at: Mapped[datetime.datetime] = mapped_column(default=_utcnow) @@ -176,6 +182,29 @@ async def ensure_platform_schema(engine: AsyncEngine) -> None: # ---- 密码哈希:定义在 auth.py(见文件头 re-export)--------------------------- +# ---- 配置版本写入(Task 6 抽公共:创建向导与配置 API 的单一版本写路径) -------- + + +async def write_config_version( + session: AsyncSession, tenant_id: int, snapshot: dict, *, + updated_by: str = "", comment: str | None = None) -> int: + """写新配置版本:MAX(version)+1,整份快照直存(spec §5.2 历史不可变, + 当前生效 = MAX)。Task 5 创建向导(版本 1)与 Task 6 配置 API(PUT 分组 + 编辑/rollback 复制旧版/reset 重置出厂)的公共写入路径——两处各自 + MAX+1 会漂移,收敛为单一事实源。调用方负责 commit;返回新版本号。""" + current = (await session.execute( + select(func.max(TenantConfigVersion.version)) + .where(TenantConfigVersion.tenant_id == tenant_id))).scalar() + version = (current or 0) + 1 + session.add(TenantConfigVersion( + tenant_id=tenant_id, + version=version, + config_json=json.dumps(snapshot, ensure_ascii=False), + updated_by=updated_by, + comment=comment)) + return version + + # ---- 超管引导 --------------------------------------------------------------- async def bootstrap_superuser( diff --git a/app/platform/routes/config_routes.py b/app/platform/routes/config_routes.py new file mode 100644 index 0000000..30d346c --- /dev/null +++ b/app/platform/routes/config_routes.py @@ -0,0 +1,433 @@ +"""配置 API(Phase 2 Task 6):/api/platform/tenants/{slug}/config* 与 +/api/platform/settings。 + +能力面(校验管线见 config_validation.py): +- GET /config:当前生效快照(MAX 版本)+ version。返回**原样快照**—— + DEFAULT_CONFIG 不入版本(出厂模板演进不被旧版本锁死,spec E4),消费侧 + (registry → TenantConfig.from_snapshot)深合并; +- PUT /config/{group}:校验 → 读当前快照 → **整组替换** → write_config_version + (MAX+1)→ registry_reload 热生效。updated_by 取认证主体(super#id / + tenant#id),comment 走查询参数(可选); +- POST /config/{group}/reset:重置出厂 = 代码库当前模板(spec §5.6,随平台 + 升级演进)——按 Tenant.profile 取 _profile_snapshot 的对应组,档案缺的组 + 取 DEFAULT_CONFIG 整组; +- GET /config/versions(列表:版本/时间/作者/备注,不含 config 本体)与 + /config/versions/{n}(详情含整份); +- POST /config/rollback {version}:目标版本复制为新版本(历史不可变), + reload 生效; +- POST /wall/test(试金石):body 可带 wall 覆盖(先 compile_wall 校验)或 + 缺省用当前墙——逐句调 boundary_filter 纯函数给判定,**不落库不 bump 版本** + (reset/编辑器的预览入口)。判定优先级与 conversation.py 执行侧一致: + 价格墙 → 零号码墙 → 再索要墙(仅 has_contact); +- GET/PUT /api/platform/settings:PlatformSetting CRUD,仅超管。本阶段只落 + 存储与接口(value 以 JSON Text 直存,读回还原类型);resolver 的平台默认 + 回退链接入归 Phase 3。 + +鉴权矩阵(spec §3):config 系 = require_tenant(slug)(租户全开——本租户 +管理员可自助改配置);settings = require_super。409 门经 platform_user 连锁。 + +版本写入一律走 db.write_config_version(与 Task 5 创建向导同一 MAX+1 路径)。 +""" +import copy +import json + +from fastapi import APIRouter, Depends, HTTPException, Request +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.engine.services import boundary_filter +from app.engine.wall_schema import WallConfigError, compile_wall +from app.platform.auth import UserContext, require_super +from app.platform.config_validation import ( + CONFIG_GROUPS, + ValidationContext, + validate_group, +) +from app.platform.db import ( + PlatformSetting, + Tenant, + TenantConfigVersion, + write_config_version, +) +from app.platform.defaults import DEFAULT_CONFIG +from app.platform.routes.tenant_routes import _get_tenant_or_404, _tenant_guard +from app.platform.tenant_source import _profile_snapshot + +router = APIRouter(prefix="/api/platform/tenants", tags=["platform-config"]) +settings_router = APIRouter( + prefix="/api/platform/settings", tags=["platform-settings"]) + + +# ---- 助手 ----------------------------------------------------------------------- + + +async def _json_body(request: Request) -> dict: + """手动解析 JSON 对象请求体:非 JSON/非对象一律 400(带实际类型), + 不让 FastAPI 的 422 抢答——配置 API 的错误契约是"400 带字段路径"。""" + try: + body = await request.json() + except ValueError as exc: + raise HTTPException(status_code=400, + detail="请求体必须是合法 JSON 对象") from exc + if not isinstance(body, dict): + raise HTTPException( + status_code=400, + detail=f"请求体必须是 JSON 对象(实际 {type(body).__name__})") + return body + + +async def _current_version_row( + session: AsyncSession, tenant_id: int) -> TenantConfigVersion | None: + """当前生效版本行(= MAX(version),db.py 纪律)。""" + return (await session.execute( + select(TenantConfigVersion) + .where(TenantConfigVersion.tenant_id == tenant_id) + .order_by(TenantConfigVersion.version.desc()) + .limit(1))).scalar_one_or_none() + + +async def _load_current( + session: AsyncSession, tenant_id: int, slug: str +) -> tuple[TenantConfigVersion, dict]: + """当前生效版本行 + 解析后的快照。无版本(防御——创建向导必写版本 1) + → 404;坏 JSON(库层数据损坏)→ 500 显性报,不静默。""" + row = await _current_version_row(session, tenant_id) + if row is None: + raise HTTPException( + status_code=404, detail=f"租户 {slug!r} 没有任何配置版本") + try: + snapshot = json.loads(row.config_json) + except ValueError as exc: + raise HTTPException( + status_code=500, + detail=f"租户 {slug!r} 版本 {row.version} 的 config_json 损坏" + ) from exc + return row, snapshot + + +async def _other_prefixes( + session: AsyncSession, exclude_tenant_id: int) -> frozenset: + """各租户**当前版本**的 push.prefix 集合(spec E1 查重数据源,排除自身; + 空/坏数据不参战)。MAX(version) 子查询定位每租户当前版。""" + max_versions = ( + select(TenantConfigVersion.tenant_id.label("tid"), + func.max(TenantConfigVersion.version).label("mv")) + .group_by(TenantConfigVersion.tenant_id).subquery()) + rows = (await session.execute( + select(TenantConfigVersion.tenant_id, TenantConfigVersion.config_json) + .join(max_versions, (TenantConfigVersion.tenant_id == max_versions.c.tid) + & (TenantConfigVersion.version == max_versions.c.mv)))).all() + prefixes = set() + for tenant_id, config_json in rows: + if tenant_id == exclude_tenant_id: + continue + try: + snapshot = json.loads(config_json) + except ValueError: + continue + push = snapshot.get("push") if isinstance(snapshot, dict) else None + prefix = push.get("prefix") if isinstance(push, dict) else None + if isinstance(prefix, str) and prefix: + prefixes.add(prefix) + return frozenset(prefixes) + + +def _updated_by(user: UserContext) -> str: + return f"{user.role}#{user.user_id}" + + +# ---- 配置快照 / 分组编辑 ---------------------------------------------------------- + + +@router.get("/{slug}/config") +async def get_config(slug: str, request: Request, + user=Depends(_tenant_guard)): + """当前生效快照(MAX 版本)+ version。原样快照,非合并视图——缺省键 + 由消费侧(registry)深合并 DEFAULT_CONFIG。""" + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + tenant = await _get_tenant_or_404(session, slug) + row, snapshot = await _load_current(session, tenant.id, slug) + return {"slug": slug, "version": row.version, "config": snapshot} + + +@router.put("/{slug}/config/{group}") +async def put_config_group( + slug: str, group: str, request: Request, comment: str | None = None, + user=Depends(_tenant_guard)): + """分组编辑:校验(config_validation 管线)→ 整组替换当前快照该组 → + MAX+1 写新版本 → registry_reload 热生效。快照存用户配置**原样组** + (组内缺省键消费侧深合并出厂默认补)。""" + body = await _json_body(request) + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + tenant = await _get_tenant_or_404(session, slug) + context = ValidationContext( + other_prefixes=await _other_prefixes(session, tenant.id)) + errors = validate_group(group, body, context) + if errors: + raise HTTPException(status_code=400, detail="; ".join(errors)) + _, current = await _load_current(session, tenant.id, slug) + new_snapshot = dict(current) + new_snapshot[group] = copy.deepcopy(body) + version = await write_config_version( + session, tenant.id, new_snapshot, + updated_by=_updated_by(user), comment=comment) + await session.commit() + + await request.app.state.registry.registry_reload(slug) + return {"slug": slug, "group": group, "version": version, + "config": new_snapshot[group]} + + +@router.post("/{slug}/config/{group}/reset") +async def reset_config_group( + slug: str, group: str, request: Request, user=Depends(_tenant_guard)): + """重置出厂(spec §5.6 = 代码库当前模板,非创建时的版本 1 快照): + 按租户创建 profile 取 _profile_snapshot 对应组;档案快照缺的组 + (contact/misc/ai)取 DEFAULT_CONFIG 整组。产新版本 + reload。""" + if group not in CONFIG_GROUPS: + raise HTTPException( + status_code=400, + detail=f"未知配置组: {group!r}(合法: " + f"{', '.join(sorted(CONFIG_GROUPS))})") + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + tenant = await _get_tenant_or_404(session, slug) + try: + factory = _profile_snapshot(tenant.profile) + except LookupError as exc: + raise HTTPException( + status_code=400, detail=f"出厂档案缺失: {exc}") from exc + if group in factory: + value = copy.deepcopy(factory[group]) + else: + value = copy.deepcopy(DEFAULT_CONFIG[group]) + _, current = await _load_current(session, tenant.id, slug) + new_snapshot = dict(current) + new_snapshot[group] = value + version = await write_config_version( + session, tenant.id, new_snapshot, + updated_by=_updated_by(user), + comment=f"重置出厂: {group}") + await session.commit() + + await request.app.state.registry.registry_reload(slug) + return {"slug": slug, "group": group, "version": version, "config": value} + + +# ---- 版本历史 / 回滚 --------------------------------------------------------------- + + +@router.get("/{slug}/config/versions") +async def list_config_versions( + slug: str, request: Request, user=Depends(_tenant_guard)): + """版本列表(升序):版本/时间/作者/备注——不含 config_json 本体 + (列表几十 KB × N 版本会白白撑爆响应)。""" + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + tenant = await _get_tenant_or_404(session, slug) + rows = (await session.execute( + select(TenantConfigVersion) + .where(TenantConfigVersion.tenant_id == tenant.id) + .order_by(TenantConfigVersion.version))).scalars().all() + return {"slug": slug, "items": [{ + "version": row.version, + "created_at": row.created_at.isoformat(), + "updated_by": row.updated_by, + "comment": row.comment, + } for row in rows]} + + +@router.get("/{slug}/config/versions/{version}") +async def get_config_version( + slug: str, version: int, request: Request, user=Depends(_tenant_guard)): + """单版本详情(含整份 config)。不存在 → 404。""" + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + tenant = await _get_tenant_or_404(session, slug) + row = (await session.execute( + select(TenantConfigVersion) + .where(TenantConfigVersion.tenant_id == tenant.id, + TenantConfigVersion.version == version) + )).scalar_one_or_none() + if row is None: + raise HTTPException( + status_code=404, detail=f"版本不存在: {version}") + return { + "slug": slug, "version": row.version, + "created_at": row.created_at.isoformat(), + "updated_by": row.updated_by, "comment": row.comment, + "config": json.loads(row.config_json), + } + + +@router.post("/{slug}/config/rollback") +async def rollback_config( + slug: str, request: Request, user=Depends(_tenant_guard)): + """回滚:目标版本整份复制为新版本(历史不可变——旧版本行永不改写), + reload 生效。""" + body = await _json_body(request) + target = body.get("version") + if not isinstance(target, int) or isinstance(target, bool) or target < 1: + raise HTTPException( + status_code=400, detail="version: 须为正整数(实际 " + f"{target!r})") + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + tenant = await _get_tenant_or_404(session, slug) + row = (await session.execute( + select(TenantConfigVersion) + .where(TenantConfigVersion.tenant_id == tenant.id, + TenantConfigVersion.version == target) + )).scalar_one_or_none() + if row is None: + raise HTTPException( + status_code=404, detail=f"版本不存在: {target}") + snapshot = json.loads(row.config_json) + new_version = await write_config_version( + session, tenant.id, snapshot, + updated_by=_updated_by(user), + comment=f"回滚自版本 {target}") + await session.commit() + + await request.app.state.registry.registry_reload(slug) + return {"slug": slug, "rollback_to": target, "version": new_version, + "config": snapshot} + + +# ---- 试金石(wall/test) ----------------------------------------------------------- + + +_WALL_TEST_KEYS = frozenset({"sentences", "wall", "has_contact", "source"}) + + +def _judge_sentence(sentence: str, wall, *, has_contact: bool, + source: str) -> dict: + """单句判定——优先级与 conversation.py 执行侧逐句门控一致: + 价格墙(掐断+替换)→ 零号码墙(替换)→ 再索要墙(仅已留资,静默丢弃)。 + 替换话术同款取自墙词表:deflection_for(首句形态,emitted 为空的分支)/ + no_contact 按 has_contact·source 分支 / 再索要无替换(丢句不掐断)。""" + if boundary_filter.violates_boundary(sentence, wall=wall): + return {"sentence": sentence, "action": "blocked", "reason": "boundary", + "replacement": boundary_filter.deflection_for( + sentence, has_contact=has_contact, wall=wall)} + if boundary_filter.no_contact_value(sentence, wall=wall): + replacement = ( + wall.no_contact["already"] if has_contact + else wall.no_contact["app"] if source == "app" + else wall.no_contact["web"]) + return {"sentence": sentence, "action": "replaced", + "reason": "no_contact", "replacement": replacement} + if has_contact and boundary_filter.asks_contact(sentence, wall=wall): + return {"sentence": sentence, "action": "blocked", "reason": "reask", + "replacement": None} + return {"sentence": sentence, "action": "pass", "reason": None, + "replacement": None} + + +@router.post("/{slug}/wall/test") +async def test_wall(slug: str, request: Request, user=Depends(_tenant_guard)): + """试金石:逐句墙判定预览。body: + - sentences: 非空字符串列表(必填); + - wall: 覆盖墙(先 compile_wall 校验,坏墙 400 带字段路径)——缺省用 + 当前生效墙(reset/编辑器改墙前的预览入口); + - has_contact: 客户已留联系方式(默认 False——启用再索要墙/换话术池); + - source: "web" | "app"(默认 web——零号码墙替换句分支)。 + 纯内存判定:不落库、不 bump 版本、不 reload。""" + body = await _json_body(request) + unknown = set(body) - _WALL_TEST_KEYS + if unknown: + raise HTTPException( + status_code=400, + detail=f"未知字段: {', '.join(sorted(unknown))}(合法: " + f"{', '.join(sorted(_WALL_TEST_KEYS))})") + sentences = body.get("sentences") + if not isinstance(sentences, list) or not sentences: + raise HTTPException(status_code=400, + detail="sentences: 须为非空字符串列表") + for i, sentence in enumerate(sentences): + if not isinstance(sentence, str): + raise HTTPException( + status_code=400, + detail=f"sentences[{i}]: 项须为字符串(实际 " + f"{type(sentence).__name__})") + has_contact = body.get("has_contact", False) + if not isinstance(has_contact, bool): + raise HTTPException( + status_code=400, + detail=f"has_contact: 须为布尔(实际 {type(has_contact).__name__})") + source = body.get("source", "web") + if source not in ("web", "app"): + raise HTTPException( + status_code=400, detail=f"source: 只允许 web/app(实际 {source!r})") + + engine = request.app.state.platform_engine + wall_override = body.get("wall") + async with AsyncSession(engine, expire_on_commit=False) as session: + tenant = await _get_tenant_or_404(session, slug) + if wall_override is not None: + wall_source = "override" + candidate = wall_override + else: + wall_source = "current" + _, snapshot = await _load_current(session, tenant.id, slug) + candidate = snapshot.get("wall") + try: + compiled = compile_wall(copy.deepcopy(candidate)) + except WallConfigError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + results = [_judge_sentence(s, compiled, has_contact=has_contact, + source=source) for s in sentences] + return {"slug": slug, "wall_source": wall_source, "results": results} + + +# ---- 平台设置(仅超管;resolver 回退链接入归 Phase 3) ----------------------------- + + +def _decode_setting(raw: str): + """Text 列还原类型:json.loads 失败(手工 SQL 塞的裸文本)回退原串。""" + try: + return json.loads(raw) + except ValueError: + return raw + + +@settings_router.get("") +async def list_settings(request: Request, user=Depends(require_super)): + """平台级默认(集团 AI 凭据/默认 SMTP 等,spec §5.2)——本阶段只落 + 存储与接口;消费(resolver 平台默认回退链)归 Phase 3。""" + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + rows = (await session.execute( + select(PlatformSetting).order_by(PlatformSetting.key)) + ).scalars().all() + return {"items": [{"key": row.key, "value": _decode_setting(row.value)} + for row in rows]} + + +@settings_router.put("") +async def put_setting(request: Request, user=Depends(require_super)): + """upsert 单个设置项:key 1-64 字符;value 任意 JSON(json.dumps 入 + Text,读回还原)。同 key 再写 = 整值替换。""" + body = await _json_body(request) + key = body.get("key") + if not isinstance(key, str) or not 1 <= len(key) <= 64: + raise HTTPException( + status_code=400, detail="key: 须为 1-64 字符字符串") + if "value" not in body: + raise HTTPException(status_code=400, detail="value: 缺少(必填)") + encoded = json.dumps(body["value"], ensure_ascii=False) + + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + row = (await session.execute( + select(PlatformSetting).where(PlatformSetting.key == key) + )).scalar_one_or_none() + if row is None: + session.add(PlatformSetting(key=key, value=encoded)) + else: + row.value = encoded + await session.commit() + return {"key": key, "value": body["value"]} diff --git a/app/platform/routes/tenant_routes.py b/app/platform/routes/tenant_routes.py index 02a7417..acea819 100644 --- a/app/platform/routes/tenant_routes.py +++ b/app/platform/routes/tenant_routes.py @@ -23,10 +23,12 @@ DEFAULT_CONFIG 满足不了。 M-2a 陷阱:Site.brand_overlay 是 ORM JSON 列——写时**整字典赋值*,不做 按键局部更新(局部改不标记脏,静默丢写)。 + +Task 6 增量:出厂档案快照助手(_profile_snapshot/_blankify)迁至 +tenant_source.py(配置 API 的重置出厂同样消费);版本 1 写入改走 +db.write_config_version(与配置 API 的 PUT/rollback/reset 同一版本写入 +路径,单一事实源);Tenant 落 profile 列(重置出厂按它取对应档案组)。 """ -import copy -import json -import pathlib import re from fastapi import APIRouter, Depends, HTTPException, Request @@ -42,50 +44,23 @@ from app.platform.auth import ( require_super, require_tenant, ) -from app.platform.db import AdminUser, Site, SiteDomain, Tenant, TenantConfigVersion -from app.platform.defaults import DEFAULT_CONFIG +from app.platform.db import ( + AdminUser, + Site, + SiteDomain, + Tenant, + TenantConfigVersion, + write_config_version, +) from app.platform.registry import _normalize_host -from app.platform.tenant_source import FileTenantSource +from app.platform.tenant_source import _profile_snapshot 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 - # ---- 校验助手 ------------------------------------------------------------------- @@ -253,7 +228,8 @@ async def create_tenant( for domain in domains: await _assert_domain_free(session, domain) - tenant = Tenant(slug=body.slug, name=body.name, status="active") + tenant = Tenant(slug=body.slug, name=body.name, status="active", + profile=body.profile) session.add(tenant) await session.flush() site = Site(tenant_id=tenant.id, slug="main", name="main") @@ -268,12 +244,13 @@ async def create_tenant( 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), + # 版本写入走公共 helper(Task 6:与配置 API 的 PUT/rollback/reset + # 同一条 MAX(version)+1 路径;新租户无历史版本 → 落版本 1) + version = await write_config_version( + session, tenant.id, snapshot, updated_by=f"super#{user.user_id}", - comment=f"创建向导: profile={body.profile}")) + comment=f"创建向导: profile={body.profile}") + assert version == 1 try: await session.commit() except IntegrityError as exc: diff --git a/app/platform/tenant_source.py b/app/platform/tenant_source.py index f0906c8..4787517 100644 --- a/app/platform/tenant_source.py +++ b/app/platform/tenant_source.py @@ -23,6 +23,7 @@ registry 只认本模块的 TenantSource 协议(async load_all / load_tenant), - 文件源数据坏:坏 JSON/slug 与目录不一致/零站点 → ValueError 带路径, wall/prompts 文件缺失 → FileNotFoundError 带路径(read_text 原样)。 """ +import copy import json import pathlib from dataclasses import dataclass @@ -33,6 +34,7 @@ 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 +from app.platform.defaults import DEFAULT_CONFIG @dataclass(frozen=True) @@ -224,3 +226,45 @@ class DbTenantSource: sites=sites, snapshot=snapshot, seed_dir=None) + + +# ---- 出厂档案快照(Task 5 创建向导/Task 6 重置出厂的公共路径) ---------------------- +# 自 tenant_routes 迁入(Task 6 微审注记:配置 API 的 reset 也要用,挪公共位置 +# 供两个路由模块消费——此处已是文件源的地盘,零新增依赖)。 + +# 出厂档案目录:与 app.main.SEED_DIR 同一锚定规则(仓库根/seed_profiles), +# 但不 import app.main——main 模块级挂路由,反向导入制造循环依赖。 +SEED_DIR = pathlib.Path(__file__).resolve().parents[2] / "seed_profiles" + +# 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 保留——见 Task 5 模块 docstring)。 + + 重置出厂(spec §5.6"代码库当前模板,随平台升级演进")与创建向导共用 + 本函数:每次调用现读 seed 目录,不缓存。未知 profile → LookupError + (read_tenant 的文件源契约,调用方映射 400)。""" + 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 diff --git a/tests/conftest.py b/tests/conftest.py index 7ed068d..a91491e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -143,6 +143,52 @@ AUTO_HOST = "chat.agent.hnxinyouhe.com" # Task 4 收拢到 conftest 单一定义) NOPE_HOST = "nope.example.com" +# ---- Phase 2 Task 6: DB 源 app/超管 client 夹具 ----------------------------------- +# 自 test_tenant_admin_api 收拢到 conftest(Task 5 时代该文件私有,Task 6 的 +# 配置 API/试金石测试与租户管理测试共用同一套 DB 源闭环环境)。 +# 跨循环纪律见 db_source_app docstring 与 test_tenant_admin_api 模块头。 + + +@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)。 + """ + from app.platform.tenant_source import DbTenantSource + + 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("/api/platform/auth/login", + json={"username": "admin", "password": "admin123"}) + assert r.status_code == 200 + c.platform_token = r.json()["token"] + r = c.post("/api/platform/auth/change-password", headers=_bearer_of(c), + json={"old_password": "admin123", + "new_password": "sup3r-pass-1"}) + assert r.status_code == 200 + yield c + + +def _bearer_of(client): + return {"Authorization": f"Bearer {client.platform_token}"} + def build_app(seed_dir, data_dir): """app 工厂(Task 7): 建 Registry → 预置 app.state.registry → 返回 app。 diff --git a/tests/test_config_api.py b/tests/test_config_api.py new file mode 100644 index 0000000..b7faf23 --- /dev/null +++ b/tests/test_config_api.py @@ -0,0 +1,431 @@ +"""配置 API(Phase 2 Task 6):分组校验管线/整份版本/回滚/重置出厂/平台设置。 + +夹具 db_source_app/super_client 自 conftest(DB 源闭环);创建租户等助手 +复用 test_tenant_admin_api(_create/_auth/_tenant_admin_token)。 + +鉴权矩阵(spec §3):config GET/PUT/reset/versions/rollback/wall-test = +require_tenant(slug)(租户全开);settings = require_super。 +""" +from app.platform.tenant_source import FileTenantSource +from tests.conftest import SEED_DIR +from tests.test_tenant_admin_api import ( + _auth, + _bearer, + _create, + _tenant_admin_token, +) + +CONFIG = "/api/platform/tenants/{slug}/config" +SETTINGS = "/api/platform/settings" +LOGIN = "/api/platform/auth/login" + + +def _cfg(slug): + return CONFIG.format(slug=slug) + + +def _get_config(client, slug): + r = client.get(_cfg(slug), headers=_auth(client)) + assert r.status_code == 200, r.text + return r.json() + + +def _put_group(client, slug, group, payload, comment=None): + url = f"{_cfg(slug)}/{group}" + if comment is not None: + url += f"?comment={comment}" + return client.put(url, headers=_auth(client), json=payload) + + +def _jc_snapshot(): + return FileTenantSource(SEED_DIR).read_tenant("jiancai").snapshot + + +# ---- 核心 7 项 ------------------------------------------------------------------ + + +def test_put_group_bumps_version_and_hot_reloads(super_client, db_source_app): + _create(super_client, slug="demo", domains=["demo.example.com"]) + # 旧值可见(顺带在 client 循环里懒建 engine——之后断言 reload dispose 它) + r = super_client.get("/api/public/brand", + headers={"host": "demo.example.com"}) + assert r.status_code == 200 + assert r.json()["color"] == "#066fc0" # jc 档案出厂色 + registry = db_source_app.state.registry + rt_old = registry._by_slug["demo"] + assert rt_old.engine is not None + + r = _put_group(super_client, "demo", "brand", {"color": "#ff0000"}) + assert r.status_code == 200, r.text + assert r.json()["version"] == 2 + + # 热生效:旧 runtime 已被 reload dispose,新 runtime 服务新值 + assert rt_old.engine is None + r = super_client.get("/api/public/brand", + headers={"host": "demo.example.com"}) + assert r.status_code == 200 + assert r.json()["color"] == "#ff0000" + # 整组替换语义:快照存用户配置原样组,缺省键由消费侧深合并出厂默认 + assert r.json()["title"] == "AI Support" + + # GET config:新值 + version(原样快照,不是合并视图) + body = _get_config(super_client, "demo") + assert body["version"] == 2 + assert body["config"]["brand"] == {"color": "#ff0000"} + + +def test_put_wall_bad_regex_400_with_field(super_client): + _create(super_client, slug="demo") + wall = _get_config(super_client, "demo")["config"]["wall"] + wall["price_patterns"] = ["([unclosed"] + r = _put_group(super_client, "demo", "wall", wall) + assert r.status_code == 400 + detail = r.json()["detail"] + assert "price_patterns[0]" in detail # 字段路径 + assert "regex" in detail # regex 字样 + # 校验失败版本不涨 + assert _get_config(super_client, "demo")["version"] == 1 + + +def test_prompts_size_cap(super_client): + _create(super_client, slug="demo") + prompts = dict(_get_config(super_client, "demo")["config"]["prompts"]) + # company_facts 50KB 软上限(spec §5.7):超一节 → 400 带字段路径与上限 + fat = dict(prompts, company_facts="x" * (50 * 1024 + 1)) + r = _put_group(super_client, "demo", "prompts", fat) + assert r.status_code == 400 + assert "company_facts" in r.json()["detail"] + assert "50" in r.json()["detail"] + # 其余三件 100KB 上限 + fat = dict(prompts, system="x" * (100 * 1024 + 1)) + r = _put_group(super_client, "demo", "prompts", fat) + assert r.status_code == 400 + assert "system" in r.json()["detail"] + # 恰在上限内 → 通过(company_facts 正好 50KB) + ok = dict(prompts, company_facts="x" * (50 * 1024)) + r = _put_group(super_client, "demo", "prompts", ok) + assert r.status_code == 200 + assert r.json()["version"] == 2 + # prompts 传字符串(Task 4 守卫兜过的形态)→ 400,不是 500/422 + r = super_client.put(f"{_cfg('demo')}/prompts", headers=_auth(super_client), + json="just a string") + assert r.status_code == 400 + # 缺键 → 400 + r = _put_group(super_client, "demo", "prompts", {"system": "x"}) + assert r.status_code == 400 + assert "company_facts" in r.json()["detail"] + + +def test_prefix_uniqueness_across_tenants(super_client): + _create(super_client, slug="jc1", profile="jiancai") # prefix ai-kefu-jc + _create(super_client, slug="au1", profile="auto") # prefix ai-kefu + _create(super_client, slug="bl1", profile="blank") # prefix "" + # auto 租户改成 jc1 的前缀 → 400(spec E1 跨租户查重) + r = _put_group(super_client, "au1", "push", {"prefix": "ai-kefu-jc"}) + assert r.status_code == 400 + assert "prefix" in r.json()["detail"] + assert "ai-kefu-jc" in r.json()["detail"] + # 新建的 blank 租户(空 prefix)同样撞车 → 400 + r = _put_group(super_client, "bl1", "push", {"prefix": "ai-kefu-jc"}) + assert r.status_code == 400 + # 自身当前前缀重写 → 200(不与自己撞);版本照涨 + r = _put_group(super_client, "au1", "push", {"prefix": "ai-kefu"}) + assert r.status_code == 200 + # 换新前缀后,别人再撞新前缀也 400(查重集合随 DB 当前版本刷新) + r = _put_group(super_client, "au1", "push", {"prefix": "ai-kefu-new"}) + assert r.status_code == 200 + r = _put_group(super_client, "bl1", "push", {"prefix": "ai-kefu-new"}) + assert r.status_code == 400 + # 空串(未设置)不参战:多个租户可同为 "" + r = _put_group(super_client, "bl1", "push", {"prefix": ""}) + assert r.status_code == 200 + + +def test_rollback_restores(super_client): + _create(super_client, slug="demo", domains=["demo.example.com"]) + assert _put_group(super_client, "demo", "brand", + {"color": "#ff0000"}).status_code == 200 # v2 + + # 版本列表:元数据齐全,不含 config 本体 + r = super_client.get(f"{_cfg('demo')}/versions", headers=_auth(super_client)) + assert r.status_code == 200 + items = r.json()["items"] + assert [i["version"] for i in items] == [1, 2] + assert "config_json" not in r.text and '"config"' not in r.text + assert items[0]["updated_by"].startswith("super#") + assert items[0]["comment"] == "创建向导: profile=jiancai" + + # 回滚到 v1 → 生效值回 v1,产生新版本 v3(历史不可变) + r = super_client.post(f"{_cfg('demo')}/rollback", headers=_auth(super_client), + json={"version": 1}) + assert r.status_code == 200 + assert r.json()["version"] == 3 + body = _get_config(super_client, "demo") + assert body["version"] == 3 + assert body["config"]["brand"]["color"] == "#066fc0" # v1 的值 + # 公开门面也回到 v1 色(reload 已生效) + r = super_client.get("/api/public/brand", + headers={"host": "demo.example.com"}) + assert r.json()["color"] == "#066fc0" + + # v2 历史仍在(整份可查),回滚目标不存在 → 404 + r = super_client.get(f"{_cfg('demo')}/versions/2", headers=_auth(super_client)) + assert r.status_code == 200 + assert r.json()["config"]["brand"]["color"] == "#ff0000" + r = super_client.post(f"{_cfg('demo')}/rollback", headers=_auth(super_client), + json={"version": 99}) + assert r.status_code == 404 + + +def test_tenant_admin_scoped_config(super_client): + _create(super_client, slug="jc2", admin="jc2-admin", + password="jc2-pass-1") + _create(super_client, slug="auto2", profile="auto") + token = _tenant_admin_token(super_client, "jc2-admin", "jc2-pass-1") + + # 本租户全开:GET/PUT/versions/rollback/reset/wall-test + assert super_client.get( + _cfg("jc2"), headers=_bearer(token)).status_code == 200 + r = super_client.put(f"{_cfg('jc2')}/brand", headers=_bearer(token), + json={"color": "#123456"}) + assert r.status_code == 200 + assert r.json()["version"] == 2 + assert super_client.get( + f"{_cfg('jc2')}/versions", headers=_bearer(token)).status_code == 200 + r = super_client.post(f"{_cfg('jc2')}/rollback", headers=_bearer(token), + json={"version": 1}) + assert r.status_code == 200 + r = super_client.post(f"{_cfg('jc2')}/brand/reset", headers=_bearer(token)) + assert r.status_code == 200 + r = super_client.post("/api/platform/tenants/jc2/wall/test", + headers=_bearer(token), json={"sentences": ["ok."]}) + assert r.status_code == 200 + + # 改别的租户 → 403 + assert super_client.put( + f"{_cfg('auto2')}/brand", headers=_bearer(token), + json={"color": "#000000"}).status_code == 403 + assert super_client.get( + _cfg("auto2"), headers=_bearer(token)).status_code == 403 + assert super_client.post( + "/api/platform/tenants/auto2/wall/test", headers=_bearer(token), + json={"sentences": ["ok."]}).status_code == 403 + # 平台设置超管专属 + assert super_client.get( + SETTINGS, headers=_bearer(token)).status_code == 403 + assert super_client.put( + SETTINGS, headers=_bearer(token), + json={"key": "k", "value": 1}).status_code == 403 + # 未认证 401 + assert super_client.get(_cfg("jc2")).status_code == 401 + + +# ---- 校验管线:类型/未知键/范围 --------------------------------------------------- + + +def test_validation_type_errors(super_client): + _create(super_client, slug="demo") + cases = [ + # (group, payload, detail 必含片段) + ("nope", {"x": 1}, "未知配置组"), + ("brand", {"unknown_key": 1}, "未知键"), + ("brand", {"color": 123}, "brand.color"), + ("brand", {"topics": ["a", 2]}, "brand.topics"), + ("contact", {"cc_completion": "yes"}, "cc_completion"), + ("contact", {"template_languages": "en"}, "template_languages"), + ("email", {"notify_emails": ["no-at-sign"]}, "notify_emails"), + ("email", {"smtp": "host"}, "smtp"), + ("push", {"timeout": 0}, "timeout"), + ("push", {"max_attempts": 11}, "max_attempts"), + ("push", {"retry_backoff_base": -1}, "retry_backoff_base"), + ("push", {"enabled": 1}, "enabled"), + ("ai", {"model": 42}, "model"), + ("engine", {"rate_limit_per_hour": "30"}, "rate_limit_per_hour"), + # bool 不是 int(isinstance 陷阱) + ("engine", {"rate_limit_per_hour": True}, "rate_limit_per_hour"), + ("engine", {"ws_conn_limit": 0}, "ws_conn_limit"), + ("engine", {"context_seed_keys": {"k": 1}}, "context_seed_keys.k"), + ("misc", {"name_to_region_extras": "x"}, "name_to_region_extras"), + ] + for group, payload, needle in cases: + r = _put_group(super_client, "demo", group, payload) + assert r.status_code == 400, f"{group}/{payload}: {r.text}" + assert needle in r.json()["detail"], f"{group}/{payload}: {r.text}" + # 全军覆没后版本仍是 1 + assert _get_config(super_client, "demo")["version"] == 1 + + +def test_valid_groups_pass(super_client): + _create(super_client, slug="demo") + ok_cases = [ + ("contact", {"cc_completion": False, "email_light_ask": True, + "template_languages": ["en", "zh"]}), + ("email", {"sender_name": "Demo", "notify_emails": ["a@b.com"], + "smtp": {"host": "smtp.example.com", "port": 465}}), + ("email", {"smtp": None}), + ("push", {"timeout": 5.5, "max_attempts": 3, + "retry_backoff_base": 2}), + ("push", {"timeout": None, "max_attempts": None}), + ("ai", {"api_key": None, "base_url": None, "model": "glm-4-flash"}), + ("engine", {"rate_limit_per_hour": 60, "ws_conn_limit": 2, + "context_seed_keys": {"k": "v"}}), + ("misc", {"name_to_region_extras": {"US": "加州"}}), + ] + version = 1 + for group, payload in ok_cases: + r = _put_group(super_client, "demo", group, payload) + assert r.status_code == 200, f"{group}: {r.text}" + version += 1 + assert r.json()["version"] == version + assert r.json()["config"] == payload + + +def test_put_unknown_tenant_and_misc(super_client): + _create(super_client, slug="demo") + assert super_client.get( + _cfg("ghost"), headers=_auth(super_client)).status_code == 404 + r = super_client.put(f"{_cfg('ghost')}/brand", headers=_auth(super_client), + json={"color": "#000"}) + assert r.status_code == 404 + # rollback 坏 body → 400(非 422/500) + r = super_client.post(f"{_cfg('demo')}/rollback", headers=_auth(super_client), + json={"version": "one"}) + assert r.status_code == 400 + r = super_client.post(f"{_cfg('demo')}/rollback", headers=_auth(super_client), + json={}) + assert r.status_code == 400 + # PUT comment 落版本元数据 + _put_group(super_client, "demo", "brand", {"color": "#000000"}, + comment="换主题色") + items = _get_config(super_client, "demo") + r = super_client.get(f"{_cfg('demo')}/versions", headers=_auth(super_client)) + assert r.json()["items"][-1]["comment"] == "换主题色" + + +# ---- 重置出厂 -------------------------------------------------------------------- + + +def test_reset_group_to_factory(super_client): + _create(super_client, slug="demo", domains=["demo.example.com"]) + _put_group(super_client, "demo", "brand", {"color": "#ff0000"}) # v2 + r = super_client.post(f"{_cfg('demo')}/brand/reset", + headers=_auth(super_client)) + assert r.status_code == 200 + assert r.json()["version"] == 3 + # 回到 jc 档案的整组 brand(重置出厂 = 代码库当前模板,spec §5.6) + assert r.json()["config"] == _jc_snapshot()["brand"] + body = _get_config(super_client, "demo") + assert body["config"]["brand"]["color"] == "#066fc0" + r = super_client.get("/api/public/brand", + headers={"host": "demo.example.com"}) + assert r.json()["color"] == "#066fc0" + assert r.json()["title"] == "HESUPIPE Support" + # 未知组 reset → 400 + r = super_client.post(f"{_cfg('demo')}/nope/reset", + headers=_auth(super_client)) + assert r.status_code == 400 + + +def test_reset_blank_tenant_groups(super_client): + """blank 租户:wall/prompts 重置取 jiancai 档案;档案没有的组取出厂默认。""" + _create(super_client, slug="bl1", profile="blank") + _put_group(super_client, "bl1", "brand", {"color": "#ff0000"}) + # brand → blankified 档案(身份键出厂默认,其余 jc) + r = super_client.post(f"{_cfg('bl1')}/brand/reset", + headers=_auth(super_client)) + assert r.status_code == 200 + brand = r.json()["config"] + assert brand["title"] == "AI Support" # 身份键 = DEFAULT + assert brand["whatsapp"] == "" + assert brand["color"] == "#066fc0" # 非身份键 = jc 档案 + # wall → jc 档案 + r = super_client.post(f"{_cfg('bl1')}/wall/reset", headers=_auth(super_client)) + assert r.status_code == 200 + assert r.json()["config"] == _jc_snapshot()["wall"] + # 档案缺的组 → DEFAULT_CONFIG 整组 + from app.platform.defaults import DEFAULT_CONFIG + r = super_client.post(f"{_cfg('bl1')}/contact/reset", + headers=_auth(super_client)) + assert r.status_code == 200 + assert r.json()["config"] == DEFAULT_CONFIG["contact"] + + +# ---- 平台设置(仅超管;本阶段只落存储与接口) ---------------------------------------- + + +def test_platform_settings_crud(super_client): + r = super_client.get(SETTINGS, headers=_auth(super_client)) + assert r.status_code == 200 + assert r.json() == {"items": []} + # 写入结构化值(JSON Text 存储,读回还原类型) + value = {"host": "smtp.example.com", "port": 465, "tls": True} + r = super_client.put(SETTINGS, headers=_auth(super_client), + json={"key": "default_smtp", "value": value}) + assert r.status_code == 200 + assert r.json() == {"key": "default_smtp", "value": value} + r = super_client.get(SETTINGS, headers=_auth(super_client)) + assert r.json()["items"] == [{"key": "default_smtp", "value": value}] + # 同 key 再写 = 更新(upsert,不重复行) + r = super_client.put(SETTINGS, headers=_auth(super_client), + json={"key": "default_smtp", "value": "second"}) + assert r.status_code == 200 + r = super_client.get(SETTINGS, headers=_auth(super_client)) + assert r.json()["items"] == [{"key": "default_smtp", "value": "second"}] + # 标量值也可 + r = super_client.put(SETTINGS, headers=_auth(super_client), + json={"key": "glm_timeout", "value": 30}) + assert r.status_code == 200 + # 坏 key/缺 value → 400 + assert super_client.put( + SETTINGS, headers=_auth(super_client), + json={"value": 1}).status_code == 400 + assert super_client.put( + SETTINGS, headers=_auth(super_client), + json={"key": "", "value": 1}).status_code == 400 + # 未认证 401 + assert super_client.get(SETTINGS).status_code == 401 + + +# ---- 版本列表形态 ----------------------------------------------------------------- + + +def test_versions_listing_shape(super_client, tmp_path): + _create(super_client, slug="demo") + _put_group(super_client, "demo", "brand", {"color": "#ff0000"}) + r = super_client.get(f"{_cfg('demo')}/versions", headers=_auth(super_client)) + items = r.json()["items"] + assert len(items) == 2 + for item in items: + assert set(item) == {"version", "created_at", "updated_by", "comment"} + assert item["created_at"] # ISO 时间戳 + # 详情 404 + r = super_client.get(f"{_cfg('demo')}/versions/9", headers=_auth(super_client)) + assert r.status_code == 404 + + +def test_wall_group_round_trip_and_body_hygiene(super_client): + """整墙组原样 PUT 回去必须通过(compile_wall 全量校验的 happy path); + 坏 JSON 原文/组名撞 versions 字面量一律 400;未改密租户管理员 409 门 + 经 platform_user 连锁生效。""" + _create(super_client, slug="demo") + wall = _get_config(super_client, "demo")["config"]["wall"] + r = _put_group(super_client, "demo", "wall", wall) + assert r.status_code == 200 + assert r.json()["config"] == wall + + # 组名撞 versions 字面量 → 走未知组 400(不与版本路由混淆) + r = _put_group(super_client, "demo", "versions", {"a": 1}) + assert r.status_code == 400 + assert "未知配置组" in r.json()["detail"] + # 坏 JSON 原文 → 400(手动解析,不让 422/500 抢答) + r = super_client.put(f"{_cfg('demo')}/brand", headers=_auth(super_client), + content=b"{not json") + assert r.status_code == 400 + + # 首登未改密的租户管理员 → 409(与登录/改密之外的受保护端点同款门) + _create(super_client, slug="t2", admin="t2-admin", password="t2-pass-1") + token = super_client.post( + LOGIN, json={"username": "t2-admin", "password": "t2-pass-1"} + ).json()["token"] + r = super_client.get(_cfg("t2"), headers=_bearer(token)) + assert r.status_code == 409 diff --git a/tests/test_tenant_admin_api.py b/tests/test_tenant_admin_api.py index 4109f52..1139546 100644 --- a/tests/test_tenant_admin_api.py +++ b/tests/test_tenant_admin_api.py @@ -1,7 +1,8 @@ """租户管理 API(Phase 2 Task 5):超管的创建向导/站点/域名/停用 + 403 边界。 -db_source_app 夹具:app.state.registry 用 DbTenantSource(生产形态)指向 -临时 platform.db——创建 API 写库后 registry_reload 走 DB 源闭环(Task 4 +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)。 @@ -20,62 +21,23 @@ 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 +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" -# ---- 夹具 ---------------------------------------------------------------------- - - -@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 - - # ---- 助手 ---------------------------------------------------------------------- diff --git a/tests/test_touchstone.py b/tests/test_touchstone.py new file mode 100644 index 0000000..250ac7b --- /dev/null +++ b/tests/test_touchstone.py @@ -0,0 +1,136 @@ +"""试金石(wall/test,Phase 2 Task 6):逐句墙判定预览。 + +复用 test_wall_execution_tenant 的同款姿势——boundary_filter 纯函数 + +jc/auto 两墙的词表差异做判别。判定优先级与 conversation.py 执行侧一致: +价格墙 → 零号码墙 → 再索要墙(仅已留资)。 + +红线:不落库、不 bump 版本(reset/编辑器共用的预览入口)。 +""" +from app.engine.services import boundary_filter +from app.platform.tenant_source import FileTenantSource +from tests.conftest import SEED_DIR +from tests.test_tenant_admin_api import _auth, _create + +WALL_TEST = "/api/platform/tenants/{slug}/wall/test" + +# 判别句选取依据(两墙 diff 实证,勿随手改文案——判别力依赖词表差异): +# - PRICE 句撞 jc 独有的 [€£]\s?\d 组(auto 词表无此组); +# - EMAIL 句撞零号码墙 any_email(替换句 = jc no_contact["web"]); +# - CLEAN 句三墙全放行。 +PRICE = "The total is €500 for one set." +EMAIL = "Sure, you can also email me at sales@example.com anytime." +CLEAN = "We ship worldwide every day." +REASK = "What's your WhatsApp number?" + +AUTO_WALL = FileTenantSource(SEED_DIR).read_tenant("auto").snapshot["wall"] +JC_WALL = FileTenantSource(SEED_DIR).read_tenant("jiancai").snapshot["wall"] + + +def _post(client, slug, payload): + return client.post(WALL_TEST.format(slug=slug), + headers=_auth(client), json=payload) + + +def _config_version(client, slug): + r = client.get(f"/api/platform/tenants/{slug}/config", + headers=_auth(client)) + assert r.status_code == 200 + return r.json()["version"] + + +def test_touchstone_verdicts(super_client): + _create(super_client, slug="demo") + version_before = _config_version(super_client, "demo") + + r = _post(super_client, "demo", {"sentences": [PRICE, EMAIL, CLEAN]}) + assert r.status_code == 200, r.text + body = r.json() + assert body["wall_source"] == "current" + results = body["results"] + assert len(results) == 3 + + # 价格墙: blocked + 替换话术来自 jc deflections 池(en 首句) + assert results[0]["action"] == "blocked" + assert results[0]["reason"] == "boundary" + assert results[0]["replacement"] == boundary_filter.DEFLECTIONS_EN[0] + # 零号码墙: replaced + 替换句精确等于 jc no_contact["web"] + assert results[1]["action"] == "replaced" + assert results[1]["reason"] == "no_contact" + assert results[1]["replacement"] == boundary_filter.NO_CONTACT_WEB + # 干净句: pass,无替换 + assert results[2]["action"] == "pass" + assert results[2]["reason"] is None + assert results[2]["replacement"] is None + + # 不落库不 bump 版本 + assert _config_version(super_client, "demo") == version_before + + +def test_touchstone_has_contact_and_source_variants(super_client): + """已留资/app 渠道两开关照搬执行侧语义:替换句换池/换变体,再索要墙启用。""" + _create(super_client, slug="demo") + # has_contact=True: 价格墙替换句来自 deflections_contact 池 + r = _post(super_client, "demo", {"sentences": [PRICE, EMAIL, REASK], + "has_contact": True}) + results = r.json()["results"] + assert results[0]["replacement"] == boundary_filter.DEFLECTIONS_EN_CONTACT[0] + # 零号码墙替换句换 already 变体(已留资不再索要) + assert results[1]["replacement"] == boundary_filter.NO_CONTACT_ALREADY + # 再索要句被丢(has_contact 才启用): blocked 无替换 + assert results[2]["action"] == "blocked" + assert results[2]["reason"] == "reask" + assert results[2]["replacement"] is None + + # source=app: 零号码墙替换句用 app 变体 + r = _post(super_client, "demo", {"sentences": [EMAIL], "source": "app"}) + assert r.json()["results"][0]["replacement"] == boundary_filter.NO_CONTACT_APP + # has_contact=False(默认): 再索要句放行(墙未启用) + r = _post(super_client, "demo", {"sentences": [REASK]}) + assert r.json()["results"][0]["action"] == "pass" + + +def test_touchstone_wall_override(super_client): + """body 带 wall 覆盖 = reset/编辑器的预览:同句换墙判定翻转,不动库。""" + _create(super_client, slug="demo") + version_before = _config_version(super_client, "demo") + + # 覆盖为 auto 墙(无 € 组): 价格句放行 + r = _post(super_client, "demo", + {"sentences": [PRICE], "wall": AUTO_WALL}) + assert r.status_code == 200 + body = r.json() + assert body["wall_source"] == "override" + assert body["results"][0]["action"] == "pass" + + # 覆盖墙坏 regex → 400 带字段路径与 regex 字样 + bad = dict(JC_WALL) + bad["price_patterns"] = ["([unclosed"] + r = _post(super_client, "demo", {"sentences": [PRICE], "wall": bad}) + assert r.status_code == 400 + assert "price_patterns[0]" in r.json()["detail"] + assert "regex" in r.json()["detail"] + + # 覆盖实验全程不 bump 版本 + assert _config_version(super_client, "demo") == version_before + + +def test_touchstone_input_validation(super_client): + _create(super_client, slug="demo") + # sentences 必须是非空字符串列表 + assert _post(super_client, "demo", {"sentences": []}).status_code == 400 + assert _post(super_client, "demo", + {"sentences": "not a list"}).status_code == 400 + assert _post(super_client, "demo", + {"sentences": ["ok", 42]}).status_code == 400 + assert _post(super_client, "demo", {}).status_code == 400 + # source 只认 web/app;has_contact 必须 bool;未知键拒绝 + assert _post(super_client, "demo", + {"sentences": ["ok."], "source": "sms"}).status_code == 400 + assert _post(super_client, "demo", + {"sentences": ["ok."], + "has_contact": "yes"}).status_code == 400 + assert _post(super_client, "demo", + {"sentences": ["ok."], "surprise": 1}).status_code == 400 + # 未知租户 404 + assert _post(super_client, "ghost", + {"sentences": ["ok."]}).status_code == 404