179 lines
7.9 KiB
Python
179 lines
7.9 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""远光智炼平台 - 高级功能测试（写操作、错误处理、权限、日志验证）"""
|
||
|
|
import json
|
||
|
|
import time
|
||
|
|
import requests
|
||
|
|
|
||
|
|
BASE_URL = "http://localhost:17861/modelTF"
|
||
|
|
results = []
|
||
|
|
|
||
|
|
def login(username="admin", password="admin123"):
|
||
|
|
resp = requests.post(f"{BASE_URL}/login", json={"username": username, "password": password})
|
||
|
|
data = resp.json()
|
||
|
|
if data.get("code") == 0:
|
||
|
|
return data["data"]["token"]
|
||
|
|
return None
|
||
|
|
|
||
|
|
def test_api(name, method="GET", url="", json_body=None, headers=None, timeout=30):
|
||
|
|
try:
|
||
|
|
h = {"Content-Type": "application/json"}
|
||
|
|
if headers:
|
||
|
|
h.update(headers)
|
||
|
|
resp = requests.request(method, f"{BASE_URL}{url}", json=json_body, headers=h, timeout=timeout)
|
||
|
|
data = resp.json()
|
||
|
|
code = data.get("code", -1)
|
||
|
|
msg = data.get("message", "")
|
||
|
|
d = data.get("data")
|
||
|
|
if isinstance(d, list):
|
||
|
|
data_desc = f"{len(d)} items"
|
||
|
|
elif isinstance(d, dict):
|
||
|
|
data_desc = f"obj({len(d)} keys)"
|
||
|
|
elif isinstance(d, str):
|
||
|
|
data_desc = f"str({len(d)})"
|
||
|
|
elif d is None:
|
||
|
|
data_desc = "null"
|
||
|
|
else:
|
||
|
|
data_desc = str(type(d).__name__)
|
||
|
|
status = "PASS" if code == 0 else f"FAIL(code={code})"
|
||
|
|
results.append({"name": name, "method": method, "url": url, "status": status, "message": msg, "data_desc": data_desc})
|
||
|
|
return data
|
||
|
|
except Exception as e:
|
||
|
|
results.append({"name": name, "method": method, "url": url, "status": "ERROR", "message": str(e)[:120], "data_desc": "-"})
|
||
|
|
return None
|
||
|
|
|
||
|
|
# 登录
|
||
|
|
token = login()
|
||
|
|
if not token:
|
||
|
|
print("FAILED: Cannot login")
|
||
|
|
exit(1)
|
||
|
|
auth_headers = {"Authorization": f"Bearer {token}"}
|
||
|
|
print(f"Login OK, token: {token}")
|
||
|
|
|
||
|
|
# ==================== 1. 写操作完整CRUD测试 ====================
|
||
|
|
print("\n--- CRUD Test: Model Manage ---")
|
||
|
|
# 创建模型
|
||
|
|
m_data = test_api("crud-model-create", method="POST", url="/model-manage",
|
||
|
|
json_body={"name": f"crud_test_{int(time.time())}", "source": "local", "model_path": "/tmp/crud_test", "description": "CRUD test model"},
|
||
|
|
headers=auth_headers)
|
||
|
|
model_id = None
|
||
|
|
if m_data and m_data.get("data") and isinstance(m_data["data"], dict):
|
||
|
|
model_id = m_data["data"].get("id", "")
|
||
|
|
# 查询创建的模型
|
||
|
|
if model_id:
|
||
|
|
test_api("crud-model-get-by-id", url=f"/model-manage/{model_id}", headers=auth_headers)
|
||
|
|
# 更新模型
|
||
|
|
if model_id:
|
||
|
|
test_api("crud-model-update", method="PUT", url=f"/model-manage/{model_id}",
|
||
|
|
json_body={"description": "updated description"},
|
||
|
|
headers=auth_headers)
|
||
|
|
# 更新用途
|
||
|
|
if model_id:
|
||
|
|
test_api("crud-model-purpose", method="PUT", url=f"/model-manage/{model_id}/purpose",
|
||
|
|
json_body={"purpose": "chat"},
|
||
|
|
headers=auth_headers)
|
||
|
|
# 删除模型
|
||
|
|
if model_id:
|
||
|
|
test_api("crud-model-delete", method="DELETE", url=f"/model-manage/{model_id}", headers=auth_headers)
|
||
|
|
|
||
|
|
print("\n--- CRUD Test: Dataset ---")
|
||
|
|
# 创建数据集
|
||
|
|
ds_data = test_api("crud-dataset-create", method="POST", url="/dataset-manage",
|
||
|
|
json_body={"name": f"crud_ds_{int(time.time())}", "description": "CRUD test ds"},
|
||
|
|
headers=auth_headers)
|
||
|
|
ds_id = None
|
||
|
|
if ds_data and ds_data.get("data") and isinstance(ds_data["data"], dict):
|
||
|
|
ds_id = ds_data["data"].get("id", "")
|
||
|
|
if ds_id:
|
||
|
|
test_api("crud-dataset-get-by-id", url=f"/dataset-manage/{ds_id}", headers=auth_headers)
|
||
|
|
test_api("crud-dataset-update", method="PUT", url=f"/dataset-manage/{ds_id}",
|
||
|
|
json_body={"description": "updated ds"},
|
||
|
|
headers=auth_headers)
|
||
|
|
test_api("crud-dataset-delete", method="DELETE", url=f"/dataset-manage/{ds_id}", headers=auth_headers)
|
||
|
|
|
||
|
|
# ==================== 2. 用户管理CRUD ====================
|
||
|
|
print("\n--- CRUD Test: Users ---")
|
||
|
|
username = f"crud_user_{int(time.time())}"
|
||
|
|
test_api("crud-user-create", method="POST", url="/users",
|
||
|
|
json_body={"username": username, "password": "Crud1234!", "display_name": "CRUD User", "role": "viewer"},
|
||
|
|
headers=auth_headers)
|
||
|
|
# 查找用户
|
||
|
|
users_data = test_api("crud-user-list", url="/users", headers=auth_headers)
|
||
|
|
user_id = None
|
||
|
|
if users_data and users_data.get("data"):
|
||
|
|
for u in users_data["data"]:
|
||
|
|
if u.get("username") == username:
|
||
|
|
user_id = u.get("id")
|
||
|
|
break
|
||
|
|
if user_id:
|
||
|
|
test_api("crud-user-update", method="PUT", url=f"/users/{user_id}",
|
||
|
|
json_body={"display_name": "Updated User", "role": "developer"},
|
||
|
|
headers=auth_headers)
|
||
|
|
test_api("crud-user-reset-pwd", method="POST", url=f"/users/{user_id}/reset-password",
|
||
|
|
json_body={"new_password": "NewPass123!"},
|
||
|
|
headers=auth_headers)
|
||
|
|
test_api("crud-user-delete", method="DELETE", url=f"/users/{user_id}", headers=auth_headers)
|
||
|
|
|
||
|
|
# ==================== 3. 错误处理测试 ====================
|
||
|
|
print("\n--- Error Handling ---")
|
||
|
|
test_api("error-invalid-model-id", url="/model-manage/nonexistent_id_12345", headers=auth_headers)
|
||
|
|
test_api("error-invalid-dataset-id", url="/dataset-manage/nonexistent_id_12345", headers=auth_headers)
|
||
|
|
test_api("error-invalid-finetune-id", url="/fine-tune/nonexistent_id_12345", headers=auth_headers)
|
||
|
|
test_api("error-invalid-eval-id", url="/model-eval/nonexistent_id_12345", headers=auth_headers)
|
||
|
|
test_api("error-duplicate-login", method="POST", url="/login",
|
||
|
|
json_body={"username": "admin", "password": "wrong_password"})
|
||
|
|
test_api("error-missing-fields", method="POST", url="/model-manage",
|
||
|
|
json_body={"name": ""},
|
||
|
|
headers=auth_headers)
|
||
|
|
|
||
|
|
# ==================== 4. 无token访问测试 ====================
|
||
|
|
print("\n--- Auth Tests ---")
|
||
|
|
test_api("auth-no-token-users", url="/users")
|
||
|
|
test_api("auth-no-token-finetune", url="/fine-tune")
|
||
|
|
test_api("auth-invalid-token", url="/users", headers={"Authorization": "Bearer invalid_token_12345"})
|
||
|
|
test_api("auth-empty-token", url="/users", headers={"Authorization": ""})
|
||
|
|
|
||
|
|
# ==================== 5. 评测维度CRUD ====================
|
||
|
|
print("\n--- CRUD Test: Dimension ---")
|
||
|
|
dim_data = test_api("crud-dimension-create", method="POST", url="/dimension",
|
||
|
|
json_body={"name": f"test_dim_{int(time.time())}", "description": "test dimension"},
|
||
|
|
headers=auth_headers)
|
||
|
|
dim_id = None
|
||
|
|
if dim_data and dim_data.get("data") and isinstance(dim_data["data"], dict):
|
||
|
|
dim_id = dim_data["data"].get("id", "")
|
||
|
|
if dim_id:
|
||
|
|
test_api("crud-dimension-get", url=f"/dimension/{dim_id}", headers=auth_headers)
|
||
|
|
test_api("crud-dimension-update", method="PUT", url=f"/dimension/{dim_id}",
|
||
|
|
json_body={"description": "updated dimension"},
|
||
|
|
headers=auth_headers)
|
||
|
|
test_api("crud-dimension-delete", method="DELETE", url=f"/dimension/{dim_id}", headers=auth_headers)
|
||
|
|
|
||
|
|
# ==================== 6. 算力节点测试 ====================
|
||
|
|
print("\n--- Compute Nodes ---")
|
||
|
|
test_api("compute-nodes-detail", url="/compute/nodes", headers=auth_headers)
|
||
|
|
nodes_data = test_api("compute-nodes-list2", url="/compute/nodes", headers=auth_headers)
|
||
|
|
if nodes_data and nodes_data.get("data"):
|
||
|
|
for node in nodes_data["data"][:1]:
|
||
|
|
node_id = node.get("id", "")
|
||
|
|
if node_id:
|
||
|
|
test_api("compute-node-replicas", url=f"/compute/nodes/{node_id}/replicas", headers=auth_headers)
|
||
|
|
test_api("compute-node-engines", url=f"/compute/nodes/{node_id}/engines", headers=auth_headers)
|
||
|
|
break
|
||
|
|
|
||
|
|
# ==================== 输出结果 ====================
|
||
|
|
pass_count = sum(1 for r in results if r["status"] == "PASS")
|
||
|
|
fail_count = sum(1 for r in results if "FAIL" in r["status"])
|
||
|
|
error_count = sum(1 for r in results if r["status"] == "ERROR")
|
||
|
|
total = len(results)
|
||
|
|
|
||
|
|
print(f"\n{'='*80}")
|
||
|
|
print(f"ADVANCED TEST SUMMARY: PASS={pass_count} FAIL={fail_count} ERROR={error_count} TOTAL={total}")
|
||
|
|
print(f"{'='*80}\n")
|
||
|
|
|
||
|
|
for r in results:
|
||
|
|
icon = "[OK]" if r["status"] == "PASS" else "[XX]"
|
||
|
|
print(f"{icon} {r['name']:45s} {r['method']:6s} {r['status']:20s} {r['data_desc']:20s} {r['message'][:60]}")
|
||
|
|
|
||
|
|
with open("test_results_advanced.json", "w", encoding="utf-8") as f:
|
||
|
|
json.dump(results, f, ensure_ascii=False, indent=2)
|
||
|
|
print(f"\nResults saved to test_results_advanced.json")
|