73 lines
3.6 KiB
Python
73 lines
3.6 KiB
Python
"""请求运行中的商务接口并生成脱敏结果；令牌从仓库外文件读取。"""
|
|
|
|
import argparse
|
|
import json
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
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()
|
|
results = []
|
|
|
|
def request(method, path, payload=None, expected=0):
|
|
started = time.monotonic()
|
|
req = urllib.request.Request(
|
|
args.base_url.rstrip("/") + path,
|
|
data=json.dumps(payload).encode() if payload is not None else None,
|
|
headers={"x-token": token, "Content-Type": "application/json"}, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as response:
|
|
body = json.load(response)
|
|
item = {"method": method, "path": path, "http": response.status,
|
|
"code": body.get("code"), "msg": body.get("msg"),
|
|
"elapsed_ms": round((time.monotonic() - started) * 1000)}
|
|
item["passed"] = response.status == 200 and body.get("code") == expected
|
|
data = body.get("data")
|
|
if isinstance(data, dict) and "rows" in data:
|
|
item.update(total=data.get("total"), rows=len(data["rows"]))
|
|
if "available" in data:
|
|
item["available"] = data["available"]
|
|
results.append(item)
|
|
print(json.dumps(item, ensure_ascii=False), flush=True)
|
|
return data if item["passed"] else None
|
|
except (urllib.error.URLError, TimeoutError, ValueError) as error:
|
|
item = {"method": method, "path": path, "passed": False, "error": str(error)}
|
|
results.append(item)
|
|
print(json.dumps(item, ensure_ascii=False), flush=True)
|
|
return None
|
|
|
|
first = request("GET", "/sw/business/list?pageNum=1&pageSize=500")
|
|
request("GET", "/sw/business/list?pageNum=2&pageSize=10")
|
|
if first and first["rows"]:
|
|
account = first["rows"][0]
|
|
request("GET", f"/sw/business/{account['userId']}")
|
|
request("GET", "/sw/business/list?" + urllib.parse.urlencode({"userName": account["userName"]}))
|
|
bound = next((row for row in first["rows"] if row["bindCount"] > 0), account)
|
|
request("GET", f"/sw/business/{bound['userId']}/bindPlatform")
|
|
request("GET", f"/sw/business/{bound['userId']}/promotes")
|
|
request("GET", "/sw/user/preference")
|
|
request("GET", "/sw/business/bindPlatform?pageNum=1&pageSize=10")
|
|
request("GET", "/sw/business/bindPlatform?sourceDb=lx,hw&status=0&pageSize=2")
|
|
# 使用校验必然拒绝的请求体检查写接口，不新增或修改测试库业务记录。
|
|
for method, path in [("PUT", "/sw/user/preference"), ("POST", "/sw/business"),
|
|
("PUT", "/sw/business"), ("POST", "/sw/business/bindPlatform"),
|
|
("DELETE", "/sw/business/bindPlatform")]:
|
|
request(method, path, {}, expected=7)
|
|
args.output.write_text(json.dumps({"base_url": args.base_url, "checks": results,
|
|
"passed": all(row["passed"] for row in results)}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
if not all(row["passed"] for row in results):
|
|
raise SystemExit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|