From cdbd85b27c121a510ad4089ef813eef2088d7de3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Fri, 11 Sep 2026 13:05:33 +0800 Subject: [PATCH] =?UTF-8?q?test(platform):=20=E7=AB=AF=E5=88=B0=E7=AB=AF?= =?UTF-8?q?=E9=AA=8C=E8=AF=81=E9=85=8D=E7=BD=AE=E9=A1=B9=E4=B8=A2=E5=A4=B1?= =?UTF-8?q?=E5=91=8A=E8=AD=A6=EF=BC=8C=E5=B9=B6=E6=8A=8A=E5=AE=83=E6=94=B6?= =?UTF-8?q?=E6=88=90=E5=8F=AF=E5=A4=8D=E7=94=A8=E7=9A=84=E5=9B=9E=E5=BD=92?= =?UTF-8?q?=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 上一轮加了"激活时点名将被丢掉的配置项"之后，我只用 caplog 验证了方法本身， **没有跑过真实激活**——这一步把缺口补上。 tools/verify_config_drop_warning.py 走完整的"创建 → 加配置项 → 提交复核 → 审核 → 激活" 状态机，发三个版本： - A：把当前生效配置项原样复制 → 一条不少，无告警（不产生噪音） - B：去掉 agent_tools/risk:general → 告警 1 条并**精确点名**该配置项 - C：把完整的那份再发一次 → 恢复原状，无告警 跑完校验生效配置项与起点一致，避免把环境留在"少一条"的状态。 实测结果：A 告警 0 条 / B 告警 1 条且点名 agent_tools/risk:general / C 告警 0 条； 起点与终点均为 9 条配置项，环境已复原。 --- tools/verify_config_drop_warning.py | 194 ++++++++++++++++++++++++++++ 1 file changed, 194 insertions(+) create mode 100644 tools/verify_config_drop_warning.py diff --git a/tools/verify_config_drop_warning.py b/tools/verify_config_drop_warning.py new file mode 100644 index 0000000..1e660cc --- /dev/null +++ b/tools/verify_config_drop_warning.py @@ -0,0 +1,194 @@ +"""端到端验证：配置发布激活时是否真的点名了"将被丢掉的配置项"。 + +背景见 `tests/unit/service/test_config_release_dropped_items.py` —— 那里锁的是方法本身， +这里验证**方法确实被 activate 调用**、且在真实流程里能打出日志。 + +三步，走完整的"创建 → 加配置项 → 提交复核 → 审核 → 激活"状态机： + +- A：把当前生效的配置项**原样复制**一份 → 一条不少，**不应**有告警 +- B：在 A 的基础上去掉一条 → **应**告警并点名那一条 +- C：把完整的那份再发一次 → 恢复原状，**不应**有告警 + +最后校验生效配置项数量回到起点，避免把环境留在"少一条"的状态。 +""" + +import asyncio +import datetime as dt +import json +import logging +import pathlib +import sys +import uuid +from typing import Any + +import asyncmy +import httpx +import jwt + +from app.core.config import get_settings +from app.main import create_app + +ADMIN = "9003" +DROPPED_KEY = "risk:general" # 故意丢掉这一条：它本身是空白名单，去掉对业务无影响 + +captured: list[str] = [] + + +class _Capture(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + if record.levelno >= logging.WARNING: + captured.append(record.getMessage()) + + +def token(subject: str) -> str: + settings = get_settings() + private_key = pathlib.Path(settings.jwt_private_key_path).read_text(encoding="utf-8") + now = dt.datetime.now(dt.UTC) + return jwt.encode( + { + "sub": subject, "iss": settings.jwt_issuer, "aud": settings.jwt_audience, + "exp": now + dt.timedelta(minutes=30), "nbf": now - dt.timedelta(seconds=5), + "jti": str(uuid.uuid4()), + }, + private_key, + algorithm="RS256", + ) + + +async def connect() -> Any: + settings = get_settings() + dsn = settings.mysql_dsn.split("://", 1)[1] + credentials, location = dsn.split("@", 1) + user, password = credentials.split(":", 1) + host_port, database = location.split("/", 1) + host, _, port = host_port.partition(":") + return await asyncmy.connect( + host=host, port=int(port or 3306), user=user, password=password, db=database + ) + + +async def active_items() -> list[dict[str, Any]]: + connection = await connect() + try: + cursor = connection.cursor() + await cursor.execute( + "SELECT i.namespace, i.config_key, i.value_json, i.schema_version" + " FROM platform_config_item i JOIN config_release r ON r.id = i.release_id" + " WHERE r.status = 'active'" + ) + rows = await cursor.fetchall() + finally: + connection.close() + return [ + { + "namespace": namespace, "item_key": key, + "value_json": json.loads(value) if isinstance(value, str) else value, + "schema_version": schema, + } + for namespace, key, value, schema in rows + ] + + +async def post(client: httpx.AsyncClient, path: str, *, auth: dict[str, str], + payload: Any = None, if_match: str | None = None) -> httpx.Response: + headers = {**auth, "Idempotency-Key": uuid.uuid4().hex} + if if_match: + headers["If-Match"] = if_match + return await client.post(path, json=payload, headers=headers) + + +async def etag(client: httpx.AsyncClient, path: str, auth: dict[str, str]) -> str | None: + return (await client.get(path, headers=auth)).headers.get("ETag") + + +async def publish(client: httpx.AsyncClient, auth: dict[str, str], + items: list[dict[str, Any]], label: str) -> str: + """走完整状态机发布一个版本，返回激活后的状态。""" + created = await post(client, "/api/v1/admin/config-releases", auth=auth, payload={ + "release_no": f"tz-verify-{uuid.uuid4().hex[:10]}", + "title": f"临时验证版本 {label}", + "change_summary": "端到端验证配置项丢失告警用，验证后恢复原状", + }) + if created.status_code != 201: + raise SystemExit(f"[{label}] 创建失败 {created.status_code} {created.text[:200]}") + release_id = int(created.json()["data"]["id"]) + + base = f"/api/v1/admin/config-releases/{release_id}/platform-config-items" + for item in items: + response = await post(client, base, auth=auth, payload=item) + if response.status_code != 201: + raise SystemExit(f"[{label}] 加配置项失败 {response.status_code} {response.text[:200]}") + + release_base = f"/api/v1/admin/config-releases/{release_id}" + steps = ( + ("validations", {}), + ("reviews", {"decision": "approved", "comment": "临时验证"}), + ("activations", {}), + ) + for action, payload in steps: + response = await post(client, f"{release_base}/{action}", auth=auth, payload=payload, + if_match=await etag(client, release_base, auth)) + if response.status_code not in (200, 201): + raise SystemExit( + f"[{label}] {action} 失败 {response.status_code} {response.text[:200]}" + ) + return str(response.json()["data"]["status"]) + + +async def main() -> int: + logging.getLogger("app.service.config_release_service").addHandler(_Capture()) + logging.getLogger("app.service.config_release_service").setLevel(logging.WARNING) + + app = create_app() + auth = {"Authorization": f"Bearer {token(ADMIN)}"} + out: list[str] = [] + + baseline = await active_items() + out.append(f"起点：生效配置项 {len(baseline)} 条") + for item in baseline: + out.append(f" {item['namespace']}/{item['item_key']}") + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://t", timeout=120 + ) as client: + # A：原样复制 → 一条不少，不应告警 + captured.clear() + status = await publish(client, auth, baseline, "A-完整复制") + out.append(f"\n[A] 完整复制 {len(baseline)} 条 → {status}") + out.append(f" 告警条数={len(captured)}（期望 0）") + for line in captured: + out.append(f" ! {line}") + + # B：去掉一条 → 应告警并点名 + reduced = [ + item for item in baseline + if not (item["namespace"] == "agent_tools" and item["item_key"] == DROPPED_KEY) + ] + captured.clear() + status = await publish(client, auth, reduced, "B-少一条") + out.append(f"\n[B] 去掉 agent_tools/{DROPPED_KEY} → 发 {len(reduced)} 条 → {status}") + out.append(f" 告警条数={len(captured)}（期望 ≥1）") + for line in captured: + out.append(f" ! {line}") + + # C：恢复完整 → 恢复原状，不应告警（C 比 B 只多不少） + captured.clear() + status = await publish(client, auth, baseline, "C-恢复") + out.append(f"\n[C] 恢复 {len(baseline)} 条 → {status}") + out.append(f" 告警条数={len(captured)}（期望 0）") + for line in captured: + out.append(f" ! {line}") + + final = await active_items() + out.append(f"\n终点：生效配置项 {len(final)} 条") + same = {(i["namespace"], i["item_key"]) for i in final} == { + (i["namespace"], i["item_key"]) for i in baseline + } + out.append(f"与起点一致：{'是' if same else '否 —— 需要人工恢复！'}") + + pathlib.Path("_dbg_verify.txt").write_text("\n".join(out), encoding="utf-8") + print("ok") + return 0 if same else 1 + + +sys.exit(asyncio.run(main()))