49 lines
2.6 KiB
Python
49 lines
2.6 KiB
Python
"""只读核对原生账号与商务状态；写接口只发送缺失 ID 的无效请求。"""
|
|
|
|
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", required=True, type=Path)
|
|
parser.add_argument("--output", required=True, type=Path)
|
|
args = parser.parse_args()
|
|
token = args.token_file.read_text(encoding="utf-8").strip()
|
|
checks = []
|
|
|
|
def request(method, path, body=None, expected=0):
|
|
req = 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(req, timeout=30) as response:
|
|
result = json.load(response)
|
|
checks.append({"method": method, "path": path, "http": response.status,
|
|
"code": result.get("code"), "msg": result.get("msg"),
|
|
"passed": response.status == 200 and result.get("code") == expected})
|
|
return result.get("data", {})
|
|
|
|
native = request("POST", "/user/getUserList", {"page": 1, "pageSize": 500})
|
|
business = request("GET", "/sw/business/list?pageNum=1&pageSize=500")
|
|
native_rows = native.get("list", [])
|
|
by_id = {row["userId"]: row for row in business.get("rows", [])}
|
|
conflicts = sum((row.get("enable") == 1) != (row.get("businessStatus") == 0) for row in native_rows)
|
|
mismatches = sum(str(row.get("businessStatus")) != by_id.get(row["ID"], {}).get("status") for row in native_rows)
|
|
request("PUT", "/user/setUserInfo", {"ID": 0, "enable": 1}, expected=7)
|
|
report = {"scope": "真实 sw7 登录会话，只读检查；未启停任何真实账号。正向启停和审核状态保留由隔离 TCP HTTP 测试覆盖。",
|
|
"checks": checks, "native_total": native.get("total"), "business_total": business.get("total"),
|
|
"checked_users": len(native_rows), "inconsistent_states": conflicts, "response_mismatches": mismatches,
|
|
"passed": all(item["passed"] for item in checks) and conflicts == 0 and mismatches == 0
|
|
and len(native_rows) == native.get("total") == business.get("total")}
|
|
args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
print(json.dumps(report, ensure_ascii=False))
|
|
if not report["passed"]:
|
|
raise SystemExit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|