yumoqing/pipeline-app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(deploy): 冒泡测试套件 run_smoke.py+13关键用例 + ddiff类型差异不再误报缺列

Browse Source
This commit is contained in:
yumoqing 2026-09-01 16:57:15 +08:00
parent 704c3f6495
commit b86b39e051
3 changed files with 188 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
deploy/ddiff.py
View File

@ -87,27 +87,31 @@ def compare(base, target):
bc, tc = set(base["cols"]), set(target["cols"])
from collections import defaultdict
miss_cols = defaultdict(list)
for c in bc - tc:
p = c.split("|")
if len(p) == 3:
miss_cols[p[0]].append(p[1] + " " + p[2])
print("\n── 缺列 [%d 表] %s" % (len(miss_cols), "【需迁移】" if miss_cols else ""))
for t in sorted(miss_cols):
print(" %s: %s" % (t, ", ".join(sorted(miss_cols[t]))))
# 列宽/类型不同（同名不同型）
# 列类型不同（同名不同型）先算出来——这些不算"缺列"，避免重复报告
base_colmap = {}
for c in base["cols"]:
p = c.split("|")
if len(p) == 3:
base_colmap[(p[0], p[1])] = p[2]
type_diff = []
target_colmap = {}
for c in target["cols"]:
p = c.split("|")
if len(p) == 3 and (p[0], p[1]) in base_colmap:
bt2 = base_colmap[(p[0], p[1])]
if bt2 != p[2]:
type_diff.append("%s.%s: 对照=%s 本环境=%s" % (p[0], p[1], bt2, p[2]))
if len(p) == 3:
target_colmap[(p[0], p[1])] = p[2]
type_diff = []
for (t, col), bt2 in base_colmap.items():
if (t, col) in target_colmap and target_colmap[(t, col)] != bt2:
type_diff.append("%s.%s: 对照=%s 本环境=%s" % (t, col, bt2, target_colmap[(t, col)]))
type_diff_keys = set((d.split(".")[0], d.split(".")[1].split(":")[0]) for d in type_diff)
miss_cols = defaultdict(list)
for c in bc - tc:
p = c.split("|")
if len(p) == 3 and (p[0], p[1]) not in type_diff_keys:
miss_cols[p[0]].append(p[1] + " " + p[2])
print("\n── 缺列 [%d 表] %s" % (len(miss_cols), "【需迁移】" if miss_cols else ""))
for t in sorted(miss_cols):
print(" %s: %s" % (t, ", ".join(sorted(miss_cols[t]))))
print("\n── 列类型不同 [%d] %s" % (len(type_diff), "【需迁移】" if type_diff else ""))
for d in type_diff:
print(" ", d)

34
deploy/smoke/cases.json Normal file
View File

@ -0,0 +1,34 @@
{
"_说明": "冒泡测试用例——增量部署后关键功能点验收。runner: deploy/dsmoke.py。auth=login 的用例先登录拿会话。expect: status 状态码 / contains 响应含串 / not_contains 不含串 / json_field 存在字段。凭据从环境变量 SMOKE_USER/SMOKE_PASS 读（不入库）。",
"base": "",
"cases": [
{"id": "s01", "name": "健康检查", "url": "/", "auth": "none",
"expect": {"status": 200}},
{"id": "s02", "name": "登录", "url": "/rbac/user/up_login.dspy", "method": "POST",
"body": {"username": "${SMOKE_USER}", "password": "***"}, "auth": "none",
"expect": {"status": 200}},
{"id": "s03", "name": "主页壳", "url": "/", "auth": "login",
"expect": {"status": 200, "contains": "产线平台"}},
{"id": "s04", "name": "agent菜单", "url": "/pipeline_core/api/agent_menus.dspy", "auth": "login",
"expect": {"status": 200, "contains": "工作空间"}},
{"id": "s05", "name": "商机产线壳", "url": "/pipeline-opportunity/agent/index.ui", "auth": "login",
"expect": {"status": 200, "contains": "商机产线"}},
{"id": "s06", "name": "投标产线壳", "url": "/pipeline-bidding/agent/index.ui", "auth": "login",
"expect": {"status": 200, "contains": "投标产线"}},
{"id": "s07", "name": "开发产线壳", "url": "/pipeline-sdlc/agent/index.ui", "auth": "login",
"expect": {"status": 200}},
{"id": "s08", "name": "运营产线壳", "url": "/pipeline-ops/agent/index.ui", "auth": "login",
"expect": {"status": 200}},
{"id": "s09", "name": "营销产线壳", "url": "/pipeline-dist/agent/index.ui", "auth": "login",
"expect": {"status": 200}},
{"id": "s10", "name": "商机数据参考(爬虫平台连通)", "url": "/pipeline-opportunity/api/opp_references_popup.dspy", "auth": "login",
"expect": {"status": 200}},
{"id": "s11", "name": "项目管理列表", "url": "/pipeline_core/api/agent_menu_open.dspy", "method": "POST",
"body": {"menu": "workspace"}, "auth": "login",
"expect": {"status": 200}},
{"id": "s12", "name": "待办", "url": "/pipeline_core/api/human_todo_list.dspy", "auth": "login",
"expect": {"status": 200}},
{"id": "s13", "name": "i18n资源", "url": "/i18n/zh/i18n.json", "auth": "none",
"expect": {"status": 200, "contains": "产线"}}
]
}

136
deploy/smoke/run_smoke.py Normal file
View File

@ -0,0 +1,136 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""run_smoke.py — 增量部署冒泡测试执行器。
用法（在应用根目录，凭据走环境变量或参数，不落库）：
SMOKE_USER=admin SMOKE_PASS=*** ./py3/bin/python deploy/smoke/run_smoke.py
./py3/bin/python deploy/smoke/run_smoke.py --base http://127.0.0.1:9090 --user admin --pass ***
./py3/bin/python deploy/smoke/run_smoke.py --https https://doit.opencomputing.cn # 走外网域名
退出码：0=全部通过；1=有用例失败。增量部署第3步验证必跑。
"""
import json
import os
import sys
import ssl
import urllib.request
import urllib.error
import urllib.parse
HERE = os.path.dirname(os.path.abspath(__file__))
CASES = os.path.join(HERE, "cases.json")
def req(opener, base, case, session_cookie, verify_ssl):
url = base.rstrip("/") + case["url"]
method = case.get("method", "GET").upper()
body = case.get("body")
data = None
headers = {}
if body is not None:
# 变量替换
b = {}
for k, v in body.items():
if isinstance(v, str):
v = v.replace("${SMOKE_USER}", os.environ.get("SMOKE_USER", ""))
v = v.replace("${SMOKE_PASS}", os.environ.get("SMOKE_PASS", ""))
b[k] = v
data = json.dumps(b).encode("utf-8")
headers["Content-Type"] = "application/json"
if case.get("auth") == "login" and session_cookie:
headers["Cookie"] = session_cookie
r = urllib.request.Request(url, data=data, method=method, headers=headers)
ctx = None
if url.startswith("https"):
ctx = ssl.create_default_context()
if not verify_ssl:
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
try:
resp = opener.open(r, timeout=20, context=ctx) if ctx else opener.open(r, timeout=20)
status = resp.getcode()
text = resp.read().decode("utf-8", errors="replace")
return status, text, resp.headers, None
except urllib.error.HTTPError as e:
return e.code, (e.read() or b"").decode("utf-8", errors="replace"), e.headers, None
except Exception as e:
return 0, "", None, str(e)
def main():
base = "http://127.0.0.1:9090"
verify_ssl = True
args = sys.argv[1:]
if "--https" in args:
base = args[args.index("--https") + 1]
verify_ssl = False # 自签/内网域名常见
elif "--base" in args:
base = args[args.index("--base") + 1]
if "--user" in args:
os.environ["SMOKE_USER"] = args[args.index("--user") + 1]
if "--pass" in args:
os.environ["SMOKE_PASS"] = args[args.index("--pass") + 1]
cfg = json.load(open(CASES))
cases = cfg["cases"]
import http.cookiejar
cj = http.cookiejar.CookieJar()
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
# 第一步：登录拿会话
session_cookie = None
login_case = None
for c in cases:
if c.get("auth") == "none" and c.get("method", "GET") == "POST" and "login" in c["url"]:
login_case = c
break
if login_case and os.environ.get("SMOKE_PASS"):
status, text, headers, err = req(opener, base, login_case, None, verify_ssl)
# 从 cookiejar 提取会话名
names = []
for ck in cj:
names.append("%s=%s" % (ck.name, ck.value))
session_cookie = "; ".join(names)
if status != 200:
print("[LOGIN-FAIL] 登录用例返回 %s，后续需登录的用例将跳过: %s" % (status, (err or text[:100])))
passed, failed, skipped = [], [], []
for c in cases:
cid, name = c["id"], c["name"]
if c.get("auth") == "login" and not session_cookie:
skipped.append(cid)
print(" SKIP %-4s %s（无登录态）" % (cid, name))
continue
status, text, headers, err = req(opener, base, c, session_cookie, verify_ssl)
exp = c.get("expect", {})
ok = True
reason = []
if err:
ok = False
reason.append("请求异常:" + err[:80])
if "status" in exp and status != exp["status"]:
ok = False
reason.append("状态 %s≠%s" % (status, exp["status"]))
if "contains" in exp and exp["contains"] not in text:
ok = False
reason.append("缺内容:%s" % exp["contains"][:20])
if "not_contains" in exp and exp["not_contains"] in text:
ok = False
reason.append("含禁止内容:%s" % exp["not_contains"][:20])
if ok:
passed.append(cid)
print(" PASS %-4s %s" % (cid, name))
else:
failed.append(cid)
print(" FAIL %-4s %s [%s]" % (cid, name, "; ".join(reason)))
print("\n══ 冒泡结果: %d 通过 / %d 失败 / %d 跳过（共 %d）══"
% (len(passed), len(failed), len(skipped), len(cases)))
sys.exit(0 if not failed else 1)
if __name__ == "__main__":
main()