151 lines
4.4 KiB
Python
Executable File
151 lines
4.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
AIOps Integration Tests
|
|
python3 tests/integration_test.py [--host HOST] [--port PORT] [--username USERNAME] [--password PASSWORD]
|
|
"""
|
|
import urllib.request, urllib.error, json, os, sys, time, argparse
|
|
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--host", default="localhost")
|
|
parser.add_argument("--port", default="80")
|
|
parser.add_argument("--username", default=os.getenv("AIOPS_TEST_USERNAME"))
|
|
parser.add_argument("--password", default=os.getenv("AIOPS_TEST_PASSWORD"))
|
|
args = parser.parse_args()
|
|
|
|
# 集成测试必须显式提供测试账号，避免误用默认凭证访问非 CI 环境。 # Logan.Li
|
|
if not args.username or not args.password:
|
|
parser.error("请通过 --username/--password 或 AIOPS_TEST_USERNAME/AIOPS_TEST_PASSWORD 提供测试账号")
|
|
|
|
BASE = "http://{}:{}".format(args.host, args.port)
|
|
TOKEN = ""
|
|
PASSED = 0
|
|
FAILED = 0
|
|
|
|
def api(method, path, body=None, auth=True):
|
|
url = BASE + path
|
|
data = json.dumps(body).encode() if body else None
|
|
req = urllib.request.Request(url, data=data, method=method)
|
|
req.add_header("Content-Type", "application/json")
|
|
if auth and TOKEN:
|
|
req.add_header("Authorization", "Bearer " + TOKEN)
|
|
try:
|
|
resp = urllib.request.urlopen(req, timeout=10)
|
|
return resp.status, json.loads(resp.read())
|
|
except urllib.error.HTTPError as e:
|
|
try:
|
|
return e.code, json.loads(e.read().decode())
|
|
except:
|
|
return e.code, {}
|
|
except Exception as e:
|
|
return 0, {"error": str(e)[:100]}
|
|
|
|
def check(name, status, detail=""):
|
|
global PASSED, FAILED
|
|
ok = 200 <= status < 300
|
|
mark = "PASS" if ok else "FAIL"
|
|
if ok: PASSED += 1
|
|
else: FAILED += 1
|
|
d = str(detail)[:80] if detail else ""
|
|
print(" [{}] {}: HTTP {} {}".format(mark, name, status, d))
|
|
return ok
|
|
|
|
print("\n" + "=" * 60)
|
|
print("AIOps Integration Tests — " + BASE)
|
|
print("=" * 60)
|
|
|
|
# [1] Health
|
|
print("\n[1] Basic checks")
|
|
s, d = api("GET", "/health", auth=False)
|
|
check("Health check", s)
|
|
s, d = api("GET", "/version", auth=False)
|
|
check("Version API ", s)
|
|
|
|
# [2] Login
|
|
print("\n[2] Authentication")
|
|
s, d = api("POST", "/api/v1/auth/login",
|
|
body={"username": args.username, "password": args.password}, auth=False)
|
|
if check("Login ", s):
|
|
TOKEN = d.get("data", {}).get("token", "")
|
|
if not TOKEN:
|
|
print(" [WARN] No token in response")
|
|
FAILED += 1
|
|
|
|
if not TOKEN:
|
|
print("[SKIP] No auth token, skipping authenticated tests\n")
|
|
print("=" * 60)
|
|
print("Result: {} passed, {} failed".format(PASSED, FAILED))
|
|
print("=" * 60)
|
|
sys.exit(1)
|
|
|
|
# [3] User
|
|
print("\n[3] Users")
|
|
s, d = api("GET", "/api/v1/admin/users")
|
|
check("Profile ", s)
|
|
s, d = api("GET", "/api/v1/admin/users")
|
|
check("Admin users ", s)
|
|
|
|
# [4] Dashboard
|
|
print("\n[4] Dashboard")
|
|
s, d = api("GET", "/api/v1/dashboard/stats")
|
|
check("Stats ", s)
|
|
s, d = api("GET", "/api/v1/dashboard/recent-alerts")
|
|
check("Recent alert", s)
|
|
|
|
# [5] Servers
|
|
print("\n[5] Servers")
|
|
s, d = api("GET", "/api/v1/servers")
|
|
check("List ", s)
|
|
s, d = api("GET", "/api/v1/servers/stats")
|
|
check("Stats ", s)
|
|
|
|
# Create + Delete
|
|
name = "test-" + str(int(time.time()))
|
|
s, d = api("POST", "/api/v1/servers", body={
|
|
"name": name, "hostname": "test.local",
|
|
"ip": "10.0.0.99", "port": 22,
|
|
"os_type": "Ubuntu", "status": "offline"
|
|
})
|
|
if check("Create ", s):
|
|
sid = d.get("data", {}).get("id", 0)
|
|
if sid:
|
|
s, d = api("GET", "/api/v1/servers/" + str(sid))
|
|
check("Get by ID ", s)
|
|
s, d = api("DELETE", "/api/v1/servers/" + str(sid))
|
|
check("Delete ", s)
|
|
|
|
# [6] Navigation
|
|
print("\n[6] Navigation")
|
|
s, d = api("GET", "/api/v1/navigations/groups")
|
|
check("Tree ", s)
|
|
|
|
# [7] Credentials
|
|
print("\n[7] Credentials")
|
|
s, d = api("GET", "/api/v1/credentials")
|
|
check("List ", s)
|
|
|
|
# [8] Pipelines
|
|
print("\n[8] Pipelines")
|
|
s, d = api("GET", "/api/v1/pipelines")
|
|
check("List ", s)
|
|
|
|
# [9] Environments
|
|
print("\n[9] Deploy Environments")
|
|
s, d = api("GET", "/api/v1/deploy/environments")
|
|
check("List ", s)
|
|
|
|
# [10] Roles
|
|
print("\n[10] Roles")
|
|
s, d = api("GET", "/api/v1/admin/roles")
|
|
check("Roles ", s)
|
|
|
|
# Summary
|
|
total = PASSED + FAILED
|
|
print("\n" + "=" * 60)
|
|
print("Result: {}/{} passed, {} failed".format(PASSED, total, FAILED))
|
|
if FAILED == 0:
|
|
print("ALL TESTS PASSED!")
|
|
else:
|
|
print("WARNING: {} failures".format(FAILED))
|
|
print("=" * 60 + "\n")
|
|
sys.exit(0 if FAILED == 0 else 1)
|