浅念
0c25287c19
- config_routes: push.api_key/ai.api_key/email.smtp.password 掩码助手
(MASKED="__SET__";非空才掩,空串/None 原样)。作用于 GET config、
GET versions/{n}、POST rollback 响应、GET settings(计划四处),并盖
PUT/reset/PUT settings 的写回显——哨兵还原后的库值不经任何响应外泄。
- PUT 哨兵还原: 密钥字段收到 __SET__ → 当前快照原值替换后再写版本
(前端"不改就原样传回");传新值照常覆盖;空串传回仍=清空(既有语义
不变)。settings 同款(密钥名含 key/password/secret,递归嵌套 dict;
设置 key 本身命中也算)。库行/版本历史永远存原值。
- tenant_routes: 站点/域名四端点(POST sites/PATCH sites/POST domains/
DELETE domains)require_super → _tenant_guard;租户生命周期三端点
(POST /tenants/GET 列表/PATCH /tenants/{slug})保持超管专属。
- 测试: 掩码四处响应+落库分离断言(_current_snapshot 直查库)、哨兵
往返/新值/清空/嵌套路径/未设还原 None、settings 标量+嵌套+非密钥
字面量;租户管理员四端点自助(201/200)+他租户 403+生命周期仍 403。
Co-Authored-By: Claude Code <noreply@anthropic.com>
628 lines
29 KiB
Python
628 lines
29 KiB
Python
"""配置 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。
|
|
|
|
Phase 3 Task 1:密钥掩码哨兵——push.api_key / ai.api_key /
|
|
email.smtp.password 及平台设置的密钥名值,响应一律 "__SET__"(原值非空),
|
|
PUT 收到 "__SET__" 还原为库值(前端"不改就原样传回"的往返契约)。
|
|
"""
|
|
import json
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.platform.db import PlatformSetting
|
|
from app.platform.tenant_source import FileTenantSource
|
|
from tests.conftest import SEED_DIR
|
|
from tests.test_tenant_admin_api import (
|
|
_auth,
|
|
_bearer,
|
|
_create,
|
|
_current_snapshot,
|
|
_db_fetch,
|
|
_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_reset_push_prefix_collision(super_client):
|
|
"""终审 I-3:reset push 组补跨租户查重——本租户改过前缀后,出厂前缀被
|
|
别人取走,重置回出厂会撞车 → 400 提示改走 PUT 指定前缀。"""
|
|
_create(super_client, slug="jc1", profile="jiancai") # 出厂 ai-kefu-jc
|
|
_create(super_client, slug="au1", profile="auto") # 出厂 ai-kefu
|
|
# jc1 让出出厂前缀,au1 取走
|
|
assert _put_group(super_client, "jc1", "push",
|
|
{"prefix": "jc1-new"}).status_code == 200
|
|
assert _put_group(super_client, "au1", "push",
|
|
{"prefix": "ai-kefu-jc"}).status_code == 200
|
|
# 重置 jc1 的 push → 出厂 ai-kefu-jc 已被 au1 占用 → 400
|
|
r = super_client.post(f"{_cfg('jc1')}/push/reset",
|
|
headers=_auth(super_client))
|
|
assert r.status_code == 400
|
|
detail = r.json()["detail"]
|
|
assert "ai-kefu-jc" in detail and "占用" in detail
|
|
# 当前版本未被重置覆盖(仍是 jc1-new)
|
|
assert _get_config(super_client, "jc1")["config"]["push"]["prefix"] == "jc1-new"
|
|
# 撞车解除后(au1 改走)重置即可成功
|
|
assert _put_group(super_client, "au1", "push",
|
|
{"prefix": "au1-new"}).status_code == 200
|
|
r = super_client.post(f"{_cfg('jc1')}/push/reset",
|
|
headers=_auth(super_client))
|
|
assert r.status_code == 200
|
|
assert r.json()["config"]["prefix"] == "ai-kefu-jc"
|
|
|
|
|
|
# ---- 平台设置(仅超管;本阶段只落存储与接口) ----------------------------------------
|
|
|
|
|
|
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(与登录/改密之外的受保护端点同款门)
|
|
# (blank: 第二个租户不撞 jiancai 出厂 push.prefix——终审 I-3 创建查重)
|
|
_create(super_client, slug="t2", admin="t2-admin", password="t2-pass-1",
|
|
profile="blank")
|
|
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
|
|
|
|
|
|
# ---- 密钥掩码 / __SET__ 哨兵(Phase 3 Task 1) ---------------------------------------
|
|
|
|
|
|
def test_get_config_masks_secrets(super_client):
|
|
"""GET config:密钥字段非空 → "__SET__";空串/None 原样(区分"配了"与
|
|
"没配");原值绝不出现在响应。"""
|
|
_create(super_client, slug="demo")
|
|
_put_group(super_client, "demo", "push",
|
|
{"prefix": "demo-p", "api_key": "push-s3cret"})
|
|
_put_group(super_client, "demo", "ai", {"api_key": "ai-s3cret"})
|
|
_put_group(super_client, "demo", "email",
|
|
{"smtp": {"host": "smtp.example.com", "port": 465,
|
|
"password": "smtp-s3cret"}})
|
|
r = super_client.get(_cfg("demo"), headers=_auth(super_client))
|
|
assert r.status_code == 200
|
|
cfg = r.json()["config"]
|
|
assert cfg["push"]["api_key"] == "__SET__"
|
|
assert cfg["ai"]["api_key"] == "__SET__"
|
|
assert cfg["email"]["smtp"]["password"] == "__SET__"
|
|
for secret in ("push-s3cret", "ai-s3cret", "smtp-s3cret"):
|
|
assert secret not in r.text, f"响应泄露密钥: {secret}"
|
|
|
|
# 未设语义不受掩码遮蔽: 空串 → "";None → null
|
|
_put_group(super_client, "demo", "push",
|
|
{"prefix": "demo-p", "api_key": ""})
|
|
_put_group(super_client, "demo", "ai", {"api_key": None})
|
|
_put_group(super_client, "demo", "email", {"smtp": None})
|
|
cfg = _get_config(super_client, "demo")["config"]
|
|
assert cfg["push"]["api_key"] == ""
|
|
assert cfg["ai"]["api_key"] is None
|
|
assert cfg["email"]["smtp"] is None
|
|
|
|
|
|
def test_put_set_sentinel_keeps_stored_value(super_client, tmp_path):
|
|
"""PUT 哨兵还原:密钥字段传 "__SET__" = 保持库值(版本+1 照常);传新值
|
|
→ 覆盖;空串传回仍 = 清空(既有语义不因掩码改变)。"""
|
|
_create(super_client, slug="demo")
|
|
_put_group(super_client, "demo", "push",
|
|
{"prefix": "demo-p", "api_key": "orig-s3cret"}) # v2
|
|
r = _put_group(super_client, "demo", "push",
|
|
{"prefix": "demo-p", "api_key": "__SET__"}) # v3
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["version"] == 3
|
|
assert r.json()["config"]["api_key"] == "__SET__" # PUT 响应同样掩码
|
|
snapshot = _current_snapshot(tmp_path, "demo")
|
|
assert snapshot["push"]["api_key"] == "orig-s3cret" # 库值原样
|
|
assert snapshot["push"]["prefix"] == "demo-p" # 非密钥字段照常写
|
|
|
|
# 传新值 → 覆盖;空串 → 清空
|
|
_put_group(super_client, "demo", "push",
|
|
{"prefix": "demo-p", "api_key": "new-s3cret"}) # v4
|
|
assert _current_snapshot(tmp_path, "demo")["push"]["api_key"] == "new-s3cret"
|
|
_put_group(super_client, "demo", "push",
|
|
{"prefix": "demo-p", "api_key": ""}) # v5
|
|
assert _current_snapshot(tmp_path, "demo")["push"]["api_key"] == ""
|
|
|
|
# 嵌套路径(email.smtp.password)哨兵同款
|
|
_put_group(super_client, "demo", "email",
|
|
{"smtp": {"host": "smtp.example.com", "port": 465,
|
|
"password": "smtp-orig"}}) # v6
|
|
r = _put_group(super_client, "demo", "email",
|
|
{"smtp": {"host": "smtp2.example.com", "port": 587,
|
|
"password": "__SET__"}}) # v7
|
|
assert r.status_code == 200
|
|
assert r.json()["config"]["smtp"]["password"] == "__SET__"
|
|
assert _current_snapshot(tmp_path, "demo")["email"]["smtp"] == {
|
|
"host": "smtp2.example.com", "port": 587, "password": "smtp-orig"}
|
|
|
|
# 库值未设时哨兵还原为 None(保持"未设置"),而非落哨兵字面量
|
|
r = _put_group(super_client, "demo", "ai", {"api_key": "__SET__"})
|
|
assert r.status_code == 200
|
|
assert _current_snapshot(tmp_path, "demo")["ai"]["api_key"] is None
|
|
|
|
|
|
def test_version_detail_and_rollback_response_masked(super_client, tmp_path):
|
|
"""GET versions/{n} 与 rollback 响应的密钥字段也是 "__SET__";写入不受
|
|
影响——rollback 落库仍是原值(掩码只作用响应,不改历史与回滚写入)。"""
|
|
_create(super_client, slug="demo")
|
|
_put_group(super_client, "demo", "ai", {"api_key": "v2-s3cret"}) # v2
|
|
r = super_client.get(f"{_cfg('demo')}/versions/2",
|
|
headers=_auth(super_client))
|
|
assert r.status_code == 200
|
|
assert r.json()["config"]["ai"]["api_key"] == "__SET__"
|
|
assert "v2-s3cret" not in r.text
|
|
|
|
_put_group(super_client, "demo", "ai", {"api_key": "v3-s3cret"}) # v3
|
|
r = super_client.post(f"{_cfg('demo')}/rollback", headers=_auth(super_client),
|
|
json={"version": 2})
|
|
assert r.status_code == 200
|
|
assert r.json()["version"] == 4
|
|
assert r.json()["config"]["ai"]["api_key"] == "__SET__" # 响应掩码
|
|
assert "v2-s3cret" not in r.text
|
|
# 落库仍是原值(分离断言:响应掩码 ≠ 写入掩码)
|
|
assert _current_snapshot(tmp_path, "demo")["ai"]["api_key"] == "v2-s3cret"
|
|
# 当前生效视图(GET config)也掩码
|
|
assert _get_config(super_client, "demo")["config"]["ai"]["api_key"] == "__SET__"
|
|
|
|
|
|
def test_platform_settings_masked(super_client, tmp_path):
|
|
"""平台设置同款:key 名含 key/password/secret 的字符串值掩码(设置 key
|
|
本身与嵌套 dict 递归都算);PUT 哨兵还原;非密钥值原样。"""
|
|
assert super_client.put(SETTINGS, headers=_auth(super_client),
|
|
json={"key": "glm_api_key",
|
|
"value": "sk-plat-1"}).status_code == 200
|
|
assert super_client.put(SETTINGS, headers=_auth(super_client),
|
|
json={"key": "default_smtp", "value": {
|
|
"host": "smtp.example.com", "port": 465,
|
|
"password": "smtp-pw-1", "user": "bot"}}
|
|
).status_code == 200
|
|
assert super_client.put(SETTINGS, headers=_auth(super_client),
|
|
json={"key": "hmac_secret",
|
|
"value": "sec-plat-1"}).status_code == 200
|
|
assert super_client.put(SETTINGS, headers=_auth(super_client),
|
|
json={"key": "glm_timeout",
|
|
"value": 30}).status_code == 200
|
|
r = super_client.get(SETTINGS, headers=_auth(super_client))
|
|
assert r.status_code == 200
|
|
items = {i["key"]: i["value"] for i in r.json()["items"]}
|
|
assert items["glm_api_key"] == "__SET__" # 设置 key 本身命中
|
|
assert items["default_smtp"] == { # 嵌套 dict 递归
|
|
"host": "smtp.example.com", "port": 465,
|
|
"password": "__SET__", "user": "bot"}
|
|
assert items["hmac_secret"] == "__SET__"
|
|
assert items["glm_timeout"] == 30 # 非密钥原样
|
|
for secret in ("sk-plat-1", "smtp-pw-1", "sec-plat-1"):
|
|
assert secret not in r.text, f"响应泄露密钥: {secret}"
|
|
|
|
# PUT 哨兵还原: 嵌套密钥位置传 "__SET__" → 库值保持,其余字段更新
|
|
r = super_client.put(SETTINGS, headers=_auth(super_client),
|
|
json={"key": "default_smtp", "value": {
|
|
"host": "smtp2.example.com", "port": 587,
|
|
"password": "__SET__", "user": "bot2"}})
|
|
assert r.status_code == 200
|
|
assert r.json()["value"]["password"] == "__SET__" # 响应掩码
|
|
rows = _db_fetch(tmp_path, select(PlatformSetting)
|
|
.where(PlatformSetting.key == "default_smtp"))
|
|
assert json.loads(rows[0][0].value) == {
|
|
"host": "smtp2.example.com", "port": 587,
|
|
"password": "smtp-pw-1", "user": "bot2"}
|
|
# 标量密钥设置哨兵还原
|
|
r = super_client.put(SETTINGS, headers=_auth(super_client),
|
|
json={"key": "glm_api_key", "value": "__SET__"})
|
|
assert r.status_code == 200
|
|
rows = _db_fetch(tmp_path, select(PlatformSetting)
|
|
.where(PlatformSetting.key == "glm_api_key"))
|
|
assert json.loads(rows[0][0].value) == "sk-plat-1"
|
|
# 非密钥 key 的 "__SET__" 字面量 = 普通字符串(不还原也不掩码)
|
|
r = super_client.put(SETTINGS, headers=_auth(super_client),
|
|
json={"key": "plain_note", "value": "__SET__"})
|
|
assert r.status_code == 200
|
|
assert r.json()["value"] == "__SET__"
|
|
items = {i["key"]: i["value"] for i in super_client.get(
|
|
SETTINGS, headers=_auth(super_client)).json()["items"]}
|
|
assert items["plain_note"] == "__SET__"
|