浅念
6839cb94d2
- I-1 站点停用全链接线: Site 数据类加 status 字段; DB 源 _load_one 与 文件源 read_tenant 均只输出 active 站点(停用站点域名不进 registry 索引 → 该域名 404;全停用=合法运营状态输出空 sites;文件源无 status 键默认 active 兼容)。E2E: PATCH 站点 disabled → 其域名 404、复启用 恢复;源级直测两源过滤。 - I-3 prefix 跨租户查重补两口: db.other_push_prefixes 共用助手 (MAX(version) 逐租户当前前缀,排除自身);创建向导(出厂档案前缀被占 → 400 提示改 blank/改占用方)与 reset push(重置回出厂撞车 → 400 提示 走 PUT)接入;PUT 原有查重改用同一助手。rollback 维持豁免(历史不可变 语义优先,取舍清单记档)。既有测试第二租户改 blank profile 避撞。 - I-4 默认凭据警示: lifespan 启动检测 ADMIN_PASSWORD 仍为 Settings 字段 默认值(非硬编码第二份,防双源漂移)→ logger.warning 醒目提示; CLAUDE.md Phase 4 部署记档补 ADMIN_PASSWORD 生产要求。 - I-2/M-2 取舍补录: 密钥掩码未实现(GET config/settings 明文返回 api_key/smtp 密码,Phase 3 前须定 __SET__ 哨兵方案);站点管理权限现为 仅超管(Phase 3 对接时决策放开或改 spec);取舍清单第 2 条同步刷新 (rollback 现为唯一豁免口)。 全量回归: 571 passed(基线 566 + 新增 5)。 Co-Authored-By: Claude Code <noreply@anthropic.com>
461 lines
21 KiB
Python
461 lines
21 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。
|
|
"""
|
|
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_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
|