172 lines
8.2 KiB
Python
172 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
|
"""战略地图版本管理 生产 E2E 验收脚本 (project-20260908-map-versions-page)
|
|
|
|
覆盖 DoD: 列表摘要/详情/回滚(自动备份)/备注编辑/跨entity404/发布自动快照
|
|
安全: 用临时地图 (title 带标记) → 全程 API → 完成后 DELETE 地图清理 (FK 级联删版本行)
|
|
可复跑: venv/bin/python3 backend/scripts/verify_map_versions_e2e.py
|
|
"""
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
|
|
BASE = "http://127.0.0.1:8010/api/cma"
|
|
MARK = "【系统验证临时数据-验证后删除】"
|
|
PASSED = []
|
|
|
|
def log(ok, msg):
|
|
PASSED.append(ok)
|
|
print(("✅" if ok else "❌") + " " + msg)
|
|
|
|
def api(method, path, token=None, body=None):
|
|
url = BASE + path
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
req = urllib.request.Request(url, data=data, method=method)
|
|
req.add_header("Content-Type", "application/json")
|
|
if token:
|
|
req.add_header("Authorization", "Bearer " + token)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
return r.status, json.loads(r.read().decode())
|
|
except urllib.error.HTTPError as e:
|
|
raw = e.read().decode()
|
|
try:
|
|
return e.code, json.loads(raw)
|
|
except Exception:
|
|
return e.code, {"detail": raw[:300]}
|
|
|
|
def login(entity_id):
|
|
st, d = api("POST", "/auth/login", body={"username": "admin", "password": "admin123", "entity_id": entity_id})
|
|
assert st == 200, f"login entity{entity_id} fail: {d}"
|
|
return d["token"]
|
|
|
|
# ── 环境读取 ────────────────────────────────
|
|
def main():
|
|
# 1) 双账套 token (admin 授权 entity1+2)
|
|
tok2 = login(2)
|
|
tok1 = login(1)
|
|
log(True, "双账套登录 admin (entity1/2)")
|
|
|
|
# 2) map49 (博海 entity2) 版本列表含新摘要字段
|
|
st, d = api("GET", "/maps/49/versions", tok2)
|
|
if st == 200 and isinstance(d.get("data"), list) and len(d["data"]) >= 2:
|
|
row = d["data"][0]
|
|
need = {"id", "map_id", "entity_id", "version", "comment", "created_at",
|
|
"dimension_count", "objective_count", "kpi_count", "connection_count", "is_current"}
|
|
missing = need - set(row.keys())
|
|
log(not missing, f"map49 版本列表 {len(d['data'])}条, 字段完整 missing={missing or '无'}")
|
|
v4 = next((x for x in d["data"] if x["version"] == "v4.0"), None)
|
|
log(bool(v4 and v4["objective_count"] == 13 and v4["kpi_count"] >= 1),
|
|
f"map49 v4.0 摘要目标数={v4 and v4['objective_count']} KPI数={v4 and v4['kpi_count']}")
|
|
log(bool(d["data"][0].get("version") in ("v4.0",)),
|
|
"map49 列表按 created_at 倒序, 首条=" + str(d["data"][0]["version"]))
|
|
v1_id = next(x["id"] for x in d["data"] if x["version"] == "v1.0")
|
|
# 详情全量
|
|
st, det = api("GET", f"/maps/49/versions/{v1_id}", tok2)
|
|
log(st == 200 and isinstance(det.get("dimensions"), list) and isinstance(det.get("canvas_data"), dict),
|
|
"map49 v1.0 详情含全量 dimensions/canvas_data")
|
|
# 备注编辑 (恢复原样: 发布 v1.0)
|
|
st, cm = api("PUT", f"/maps/49/versions/{v1_id}", tok2, {"comment": "发布 v1.0"})
|
|
log(st == 200 and (cm.get("comment") == "发布 v1.0" or cm.get("version", {}).get("comment") == "发布 v1.0"),
|
|
f"map49 v1.0 备注PUT回原值 {st}")
|
|
else:
|
|
log(False, f"map49 列表异常 st={st} d={str(d)[:200]}")
|
|
|
|
# 3) 跨 entity: entity1 token 访问 map49(entity2) → 全部 404
|
|
cross_cases = [
|
|
("GET", "/maps/49/versions", None),
|
|
("GET", "/maps/49/versions/9", None),
|
|
("POST", "/maps/49/versions/9/rollback", None),
|
|
("PUT", "/maps/49/versions/9", {"comment": "x"}),
|
|
]
|
|
all404 = True
|
|
for m, p, b in cross_cases:
|
|
st, _ = api(m, p, tok1, b)
|
|
if st != 404:
|
|
all404 = False
|
|
print(f" cross-entity {m} {p} -> {st}")
|
|
log(all404, "跨账套 entity1 访问 map49 列表/详情/回滚/备注 全 404")
|
|
|
|
# 4) 临时地图全链路 (entity2): 建模板→发布(v1.0)→改内容→回滚(备份v1.1)→备注→详情
|
|
tmp_title = "E2E版本管理测试 " + MARK
|
|
st, m = api("POST", "/maps/create-with-template", tok2, {"title": tmp_title})
|
|
if st != 200:
|
|
log(False, f"临时地图创建失败 {st} {m}")
|
|
sys.exit(1)
|
|
mid = m["id"]
|
|
log(True, f"临时地图创建 id={mid} entity2")
|
|
|
|
st, _ = api("PUT", f"/maps/{mid}", tok2, {"status": "published"})
|
|
log(st == 200, "临时地图发布 (触发 _auto_snapshot)")
|
|
st, d = api("GET", f"/maps/{mid}/versions", tok2)
|
|
rows = d.get("data", [])
|
|
log(st == 200 and len(rows) == 1 and rows[0]["version"] == "v1.0" and rows[0]["is_current"] is True,
|
|
f"发布后版本列表 v1.0 is_current=True dims={rows[0]['dimension_count'] if rows else '?'}")
|
|
v1 = rows[0]
|
|
v1_detail = api("GET", f"/maps/{mid}/versions/{v1['id']}", tok2)[1]
|
|
|
|
# 改内容 (加新目标)
|
|
new_dims = json.loads(json.dumps(v1_detail["dimensions"]))
|
|
new_dims[0]["objectives"].append({"name": "E2E新目标" + MARK, "icon": "target",
|
|
"kpis": ["F_REVENUE"], "krs": [{"name": "KR-E2E", "kpi_code": "F_REVENUE"}]})
|
|
st, _ = api("PUT", f"/maps/{mid}", tok2, {"dimensions": new_dims,
|
|
"canvas_data": {"connections": [{"from": "finance-0", "to": "finance-1"}]}})
|
|
log(st == 200, "临时地图内容更新 (published 状态保存不自动快照)")
|
|
|
|
# 回滚 → 备份 v1.1
|
|
st, rb = api("POST", f"/maps/{mid}/versions/{v1['id']}/rollback", tok2)
|
|
log(st == 200 and rb.get("version") == "v1.0" and rb.get("backup_version") == "v1.1",
|
|
f"回滚成功 自动备份={rb.get('backup_version')} msg={rb.get('message','')[:40]}")
|
|
|
|
st, d = api("GET", f"/maps/{mid}/versions", tok2)
|
|
rows = d.get("data", [])
|
|
ver_set = {x["version"] for x in rows}
|
|
backup = next((x for x in rows if x["version"] == "v1.1"), None)
|
|
cur = next((x for x in rows if x["version"] == "v1.0"), None)
|
|
log(st == 200 and ver_set == {"v1.0", "v1.1"} and backup and backup["comment"] == "回滚前自动备份"
|
|
and cur and cur["is_current"] is True,
|
|
f"回滚后列表 2条={ver_set}, v1.1=自动备份, v1.0=当前")
|
|
|
|
# 内容还原验证 (业务字段)
|
|
maps_st, maps_d = api("GET", "/maps", tok2)
|
|
cur_map = next(x for x in maps_d["data"] if x["id"] == mid)
|
|
|
|
def norm(dims):
|
|
return [{"key": x.get("key"), "objectives": [
|
|
{k: o.get(k) for k in ("name", "kpis", "krs")} for o in x.get("objectives", [])]} for x in dims]
|
|
log(cur_map["version"] == "v1.0" and norm(cur_map["dimensions"]) == norm(v1_detail["dimensions"])
|
|
and cur_map["canvas_data"] == v1_detail["canvas_data"],
|
|
"回滚后地图内容语义还原为 v1.0 快照")
|
|
|
|
# 备注编辑
|
|
st, cm = api("PUT", f"/maps/{mid}/versions/{backup['id']}", tok2, {"comment": "E2E备份备注" + MARK})
|
|
log(st == 200, "回滚备份行备注编辑成功")
|
|
st, d = api("GET", f"/maps/{mid}/versions", tok2)
|
|
b2 = next(x for x in d["data"] if x["version"] == "v1.1")
|
|
log(("E2E备份备注" + MARK) in (b2.get("comment") or ""), "备注修改后列表可见")
|
|
|
|
# 详情全量
|
|
st, det = api("GET", f"/maps/{mid}/versions/{v1['id']}", tok2)
|
|
log(st == 200 and isinstance(det.get("dimensions"), list) and "canvas_data" in det, "版本详情含全量画布")
|
|
|
|
# 5) 清理: DELETE 临时地图 (FK 级联删版本行) + 回查
|
|
st, _ = api("DELETE", f"/maps/{mid}", tok2)
|
|
log(st == 200, "清理: 临时地图已删除")
|
|
st, d = api("GET", "/maps", tok2)
|
|
gone = not any(x["id"] == mid for x in d["data"])
|
|
log(gone, "清理后地图列表无临时地图")
|
|
st, d2 = api("GET", f"/maps/{mid}/versions", tok2)
|
|
log(st == 404, "清理后临时地图版本接口 404 (地图不存在)")
|
|
|
|
# 6) map49 未被污染
|
|
st, d = api("GET", "/maps/49/versions", tok2)
|
|
log(st == 200 and d["data"][0]["version"] == "v4.0" and d["data"][0].get("is_current") is True,
|
|
"map49 保持 v4.0 当前未受影响")
|
|
|
|
ok = sum(1 for x in PASSED if x)
|
|
print(f"\n结果: {ok}/{len(PASSED)} PASS")
|
|
sys.exit(0 if ok == len(PASSED) else 2)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|