141 lines
5.2 KiB
Python
141 lines
5.2 KiB
Python
#!/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)
|
||
# 手动解析 Set-Cookie（cookiejar 在 http:// 下会拒绝带 Secure 属性的 cookie）
|
||
names = []
|
||
if headers is not None:
|
||
for hv in headers.get_all("Set-Cookie", []):
|
||
pair = hv.split(";", 1)[0].strip()
|
||
if "=" in pair:
|
||
names.append(pair)
|
||
session_cookie = "; ".join(names) if names else None
|
||
if status != 200 or not session_cookie:
|
||
print("[LOGIN-FAIL] 登录用例返回 %s cookie=%s，后续需登录的用例将跳过: %s"
|
||
% (status, bool(session_cookie), (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()
|