59 lines
2.8 KiB
Python
59 lines
2.8 KiB
Python
"""实测运行实例的预警查询和无效写请求，不修改真实预警或上传文件。"""
|
|
|
|
import argparse
|
|
import json
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--base-url", default="http://127.0.0.1:18000")
|
|
parser.add_argument("--token-file", type=Path, required=True)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
token = args.token_file.read_text(encoding="utf-8").strip()
|
|
checks = []
|
|
|
|
def call(method, path, body=None, expected=0):
|
|
request = urllib.request.Request(
|
|
args.base_url + path, method=method,
|
|
headers={"x-token": token, "Content-Type": "application/json"},
|
|
data=json.dumps(body).encode() if body is not None else None,
|
|
)
|
|
with urllib.request.urlopen(request, timeout=30) as response:
|
|
value = json.load(response)
|
|
result = {"method": method, "path": path, "http": response.status,
|
|
"code": value.get("code"), "msg": value.get("msg"),
|
|
"passed": response.status == 200 and value.get("code") == expected}
|
|
data = value.get("data", {})
|
|
if isinstance(data, dict) and "pageData" in data:
|
|
result.update(total=data["pageData"]["total"], available=data.get("available"),
|
|
status_count=data["statusCount"])
|
|
checks.append(result)
|
|
print(json.dumps(result, ensure_ascii=False), flush=True)
|
|
return data
|
|
|
|
call("GET", "/sw/alertRecord/list?pageNum=1&pageSize=10")
|
|
call("GET", "/sw/alertRecord/list?platformCodes=lx,hw&status=5")
|
|
call("GET", "/sw/alertRule/ruleTypes")
|
|
accounts = call("GET", "/sw/business/list?pageNum=1&pageSize=500")
|
|
bound = next((row for row in accounts.get("rows", []) if row["bindCount"] > 0), None)
|
|
unbound = next((row for row in accounts.get("rows", []) if row["bindCount"] == 0), None)
|
|
for account in [bound, unbound]:
|
|
if account:
|
|
call("GET", f"/sw/alertRecord/list?businessId={account['userId']}")
|
|
call("GET", "/sw/alertRecord/detail/0", expected=7)
|
|
for action in ["startProcess", "submitAudit", "approve", "reject", "reProcess", "addNote"]:
|
|
call("POST", f"/sw/alertRecord/{action}", {"alertId": 0}, expected=7)
|
|
report = {"scope": "既有 sw7，只读查询及必然拒绝的无效写请求；正向流转另用隔离 SQLite 和 TCP HTTP 测试",
|
|
"base_url": args.base_url, "checks": checks,
|
|
"passed": all(row["passed"] for row in checks)}
|
|
args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
if not report["passed"]:
|
|
raise SystemExit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|