qyyfw
78ff261b5a
- 前端 React+TypeScript (源码+构建产物) - 后端 Python FastAPI - 数据库迁移 001-015 - Docker部署配置 - 微信ClawBot API频道绑定 - 项目文档 README.md
810 lines
33 KiB
Python
810 lines
33 KiB
Python
"""
|
|
AI自进化 v2 - 接口自动化测试脚本
|
|
测试范围: 26个API端点 (14 v1 + 12 v2)
|
|
验证: 响应码、数据结构、数据隔离
|
|
"""
|
|
import requests
|
|
import json
|
|
import sys
|
|
import time
|
|
import random
|
|
from datetime import datetime
|
|
from typing import Dict, List, Optional, Tuple, Any
|
|
|
|
BASE_URL = "http://localhost:18080"
|
|
EVO_PREFIX = "/app/api/evolution"
|
|
AUTH_PREFIX = "/app/api/auth"
|
|
|
|
# Test users
|
|
USERS = {
|
|
"admin": {"username": "admin", "password": "admin123"},
|
|
"demo": {"username": "demo_user", "password": "demo123"},
|
|
}
|
|
|
|
# Store results
|
|
all_results: List[Dict] = []
|
|
stats = {"passed": 0, "failed": 0, "skipped": 0, "total": 0}
|
|
|
|
|
|
class TestResult:
|
|
def __init__(self, name: str, passed: bool, detail: str = "", priority: str = "P0"):
|
|
self.name = name
|
|
self.passed = passed
|
|
self.detail = detail
|
|
self.priority = priority
|
|
|
|
def to_dict(self) -> Dict:
|
|
return {
|
|
"name": self.name,
|
|
"passed": self.passed,
|
|
"detail": self.detail,
|
|
"priority": self.priority,
|
|
}
|
|
|
|
|
|
def login(user_key: str) -> Optional[str]:
|
|
"""Login and return access token"""
|
|
user = USERS[user_key]
|
|
try:
|
|
resp = requests.post(
|
|
f"{BASE_URL}{AUTH_PREFIX}/login",
|
|
json={"username": user["username"], "password": user["password"]},
|
|
timeout=10,
|
|
)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
return data.get("data", {}).get("access_token")
|
|
return None
|
|
except Exception as e:
|
|
print(f" Login error for {user_key}: {e}")
|
|
return None
|
|
|
|
|
|
def api_get(token: str, path: str, params: Optional[Dict] = None) -> Tuple[int, Any]:
|
|
"""Helper: GET request"""
|
|
try:
|
|
resp = requests.get(
|
|
f"{BASE_URL}{EVO_PREFIX}{path}",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
params=params or {},
|
|
timeout=15,
|
|
)
|
|
return resp.status_code, resp.json() if resp.text else {}
|
|
except Exception as e:
|
|
return 0, {"error": str(e)}
|
|
|
|
|
|
def api_post(token: str, path: str, body: Optional[Dict] = None) -> Tuple[int, Any]:
|
|
"""Helper: POST request"""
|
|
try:
|
|
resp = requests.post(
|
|
f"{BASE_URL}{EVO_PREFIX}{path}",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
json=body or {},
|
|
timeout=15,
|
|
)
|
|
if resp.status_code >= 500:
|
|
return resp.status_code, {"detail": f"Server error: {resp.status_code}"}
|
|
return resp.status_code, resp.json() if resp.text else {}
|
|
except Exception as e:
|
|
return 0, {"error": str(e)}
|
|
|
|
|
|
def api_put(token: str, path: str, body: Dict) -> Tuple[int, Any]:
|
|
"""Helper: PUT request"""
|
|
try:
|
|
resp = requests.put(
|
|
f"{BASE_URL}{EVO_PREFIX}{path}",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
json=body,
|
|
timeout=15,
|
|
)
|
|
return resp.status_code, resp.json() if resp.text else {}
|
|
except Exception as e:
|
|
return 0, {"error": str(e)}
|
|
|
|
|
|
def api_delete(token: str, path: str) -> Tuple[int, Any]:
|
|
"""Helper: DELETE request"""
|
|
try:
|
|
resp = requests.delete(
|
|
f"{BASE_URL}{EVO_PREFIX}{path}",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
timeout=15,
|
|
)
|
|
if resp.status_code >= 500:
|
|
return resp.status_code, {"detail": f"Server error: {resp.status_code}"}
|
|
return resp.status_code, resp.json() if resp.text else {}
|
|
except Exception as e:
|
|
return 0, {"error": str(e)}
|
|
|
|
|
|
def record(name: str, passed: bool, detail: str = "", priority: str = "P0"):
|
|
"""Record a test result"""
|
|
all_results.append(TestResult(name, passed, detail, priority).to_dict())
|
|
if passed:
|
|
stats["passed"] += 1
|
|
else:
|
|
stats["failed"] += 1
|
|
stats["total"] += 1
|
|
status = "PASS" if passed else "FAIL"
|
|
print(f" [{status}] {name}")
|
|
if detail and not passed:
|
|
print(f" {detail}")
|
|
|
|
|
|
def assert_code(code: int, expected: int, name: str) -> bool:
|
|
if code == expected:
|
|
record(f"{name} - HTTP {expected}", True)
|
|
return True
|
|
else:
|
|
record(f"{name} - HTTP {expected}", False, f"Got {code}")
|
|
return False
|
|
|
|
|
|
def assert_field(data: Any, field: str, name: str, check_empty: bool = False) -> bool:
|
|
"""Assert response data has a field, optionally check non-empty"""
|
|
if not isinstance(data, dict):
|
|
record(f"{name} - field '{field}' exists", False, "Response is not dict")
|
|
return False
|
|
if field not in data:
|
|
record(f"{name} - field '{field}' exists", False, f"Field '{field}' missing")
|
|
return False
|
|
if check_empty:
|
|
val = data[field]
|
|
if isinstance(val, (list, dict, str)) and len(val) == 0:
|
|
# Allow empty for now as test data may be sparse
|
|
pass
|
|
record(f"{name} - field '{field}' exists", True)
|
|
return True
|
|
|
|
|
|
# ============================================================
|
|
# Phase 1: Login & Prepare
|
|
# ============================================================
|
|
def phase_login():
|
|
print("\n=== Phase 1: Login & Prepare ===")
|
|
|
|
admin_token = login("admin")
|
|
demo_token = login("demo")
|
|
|
|
if admin_token:
|
|
record("Admin login", True)
|
|
else:
|
|
record("Admin login", False, "Cannot get admin token")
|
|
return None, None
|
|
|
|
if demo_token:
|
|
record("Demo user login", True)
|
|
else:
|
|
record("Demo user login", False, "Cannot get demo token")
|
|
return admin_token, None
|
|
|
|
return admin_token, demo_token
|
|
|
|
|
|
# ============================================================
|
|
# Phase 2: v1 Skill Management APIs (8 endpoints)
|
|
# ============================================================
|
|
def phase_v1_skills(admin_token: str):
|
|
print("\n=== Phase 2: v1 Skill Management APIs ===")
|
|
|
|
# TC-V1-01: GET /skills - List skills
|
|
code, data = api_get(admin_token, "/skills")
|
|
assert_code(code, 200, "v1: GET /skills")
|
|
assert_field(data, "code", "v1: GET /skills - code")
|
|
assert_field(data, "data", "v1: GET /skills - data")
|
|
if isinstance(data.get("data"), list):
|
|
skill_count = len(data["data"])
|
|
record(f"v1: GET /skills - returns list ({skill_count} skills)", skill_count > 0)
|
|
if skill_count > 0:
|
|
first_skill = data["data"][0]
|
|
assert_field(first_skill, "skill_name", "v1: Skill has skill_name")
|
|
assert_field(first_skill, "state", "v1: Skill has state")
|
|
assert_field(first_skill, "version", "v1: Skill has version")
|
|
assert_field(first_skill, "use_count", "v1: Skill has use_count")
|
|
|
|
# Store for later tests
|
|
test_skill = first_skill["skill_name"]
|
|
|
|
# TC-V1-02: GET /skills/{name} - Get skill detail
|
|
code, detail = api_get(admin_token, f"/skills/{test_skill}")
|
|
assert_code(code, 200, "v1: GET /skills/{name}")
|
|
assert_field(detail, "code", "v1: Skill detail has code")
|
|
if isinstance(detail.get("data"), dict):
|
|
record("v1: Skill detail has data", True)
|
|
assert_field(detail["data"], "skill_name", "v1: Skill detail - skill_name")
|
|
assert_field(detail["data"], "description", "v1: Skill detail - description")
|
|
|
|
# TC-V1-03: PUT /skills/{name} - Update skill
|
|
code, _ = api_put(admin_token, f"/skills/{test_skill}",
|
|
{"description": f"Updated test desc {int(time.time())}"})
|
|
assert_code(code, 200, "v1: PUT /skills/{name}")
|
|
|
|
# TC-V1-04: GET /skills/{name}/history - Skill history
|
|
code, hist = api_get(admin_token, f"/skills/{test_skill}/history")
|
|
assert_code(code, 200, "v1: GET /skills/{name}/history")
|
|
if isinstance(hist.get("data"), list):
|
|
record(f"v1: Skill history returns list ({len(hist['data'])} events)", True)
|
|
|
|
# TC-V1-05: GET /skills/{name}/trajectories - Skill trajectories
|
|
code, traj = api_get(admin_token, f"/skills/{test_skill}/trajectories")
|
|
assert_code(code, 200, "v1: GET /skills/{name}/trajectories")
|
|
assert_field(traj, "code", "v1: Trajectories has code")
|
|
|
|
# TC-V1-06: POST /skills/{name}/freeze - Freeze skill
|
|
code, _ = api_post(admin_token, f"/skills/{test_skill}/freeze")
|
|
assert_code(code, 200, "v1: POST /skills/{name}/freeze")
|
|
|
|
# TC-V1-07: POST /skills/{name}/unfreeze - Unfreeze skill
|
|
code, _ = api_post(admin_token, f"/skills/{test_skill}/unfreeze")
|
|
assert_code(code, 200, "v1: POST /skills/{name}/unfreeze")
|
|
else:
|
|
record("v1: GET /skills - list is empty", True, "No skills exist, skip detail tests")
|
|
else:
|
|
record("v1: GET /skills - data type", False, f"Expected list, got {type(data.get('data'))}")
|
|
|
|
# TC-V1-08: GET /skills/{name} - Skill not found
|
|
code, data = api_get(admin_token, "/skills/nonexistent_skill_xyz")
|
|
assert_code(code, 404, "v1: GET /skills/nonexistent - 404")
|
|
|
|
# TC-V1-09: DELETE /skills/{name} - Delete nonexistent
|
|
code, _ = api_delete(admin_token, "/skills/nonexistent_skill_xyz")
|
|
assert_code(code, 404, "v1: DELETE /skills/nonexistent - 404")
|
|
|
|
|
|
# ============================================================
|
|
# Phase 3: v1 Dashboard APIs (4 panels)
|
|
# ============================================================
|
|
def phase_v1_dashboard(admin_token: str):
|
|
print("\n=== Phase 3: v1 Dashboard APIs ===")
|
|
|
|
# TC-V1-10: GET /dashboard/skill-tree
|
|
code, data = api_get(admin_token, "/dashboard/skill-tree")
|
|
assert_code(code, 200, "v1: GET /dashboard/skill-tree")
|
|
assert_field(data, "data", "v1: Skill tree has data")
|
|
if isinstance(data.get("data"), dict):
|
|
assert_field(data["data"], "categories", "v1: Skill tree - categories")
|
|
assert_field(data["data"], "total_count", "v1: Skill tree - total_count")
|
|
assert_field(data["data"], "main_count", "v1: Skill tree - main_count")
|
|
assert_field(data["data"], "baby_count", "v1: Skill tree - baby_count")
|
|
assert_field(data["data"], "staging_count", "v1: Skill tree - staging_count")
|
|
assert_field(data["data"], "frozen_count", "v1: Skill tree - frozen_count")
|
|
|
|
# TC-V1-11: GET /dashboard/timeline
|
|
code, data = api_get(admin_token, "/dashboard/timeline")
|
|
assert_code(code, 200, "v1: GET /dashboard/timeline")
|
|
if isinstance(data.get("data"), list):
|
|
record(f"v1: Timeline returns list ({len(data['data'])} events)", True)
|
|
|
|
# TC-V1-12: GET /dashboard/timeline with filter
|
|
code, data = api_get(admin_token, "/dashboard/timeline", {"event_type": "edited"})
|
|
assert_code(code, 200, "v1: GET /dashboard/timeline?event_type=edited")
|
|
if isinstance(data.get("data"), list):
|
|
record(f"v1: Timeline filtered ({len(data['data'])} events)", True)
|
|
|
|
# TC-V1-13: GET /dashboard/statistics
|
|
code, data = api_get(admin_token, "/dashboard/statistics")
|
|
assert_code(code, 200, "v1: GET /dashboard/statistics")
|
|
if isinstance(data.get("data"), dict):
|
|
assert_field(data["data"], "ranking", "v1: Statistics - ranking")
|
|
assert_field(data["data"], "total_use_count", "v1: Statistics - total_use_count")
|
|
assert_field(data["data"], "category_distribution", "v1: Statistics - category_distribution")
|
|
assert_field(data["data"], "daily_trend", "v1: Statistics - daily_trend")
|
|
assert_field(data["data"], "low_use_skills", "v1: Statistics - low_use_skills")
|
|
|
|
# TC-V1-14: GET /dashboard/profile
|
|
code, data = api_get(admin_token, "/dashboard/profile")
|
|
assert_code(code, 200, "v1: GET /dashboard/profile")
|
|
if isinstance(data.get("data"), dict):
|
|
assert_field(data["data"], "level", "v1: Profile - level")
|
|
assert_field(data["data"], "level_title", "v1: Profile - level_title")
|
|
assert_field(data["data"], "dimensions", "v1: Profile - dimensions")
|
|
assert_field(data["data"], "tags", "v1: Profile - tags")
|
|
assert_field(data["data"], "growth_suggestions", "v1: Profile - growth_suggestions")
|
|
|
|
|
|
# ============================================================
|
|
# Phase 4: v1 Events & Trajectory APIs
|
|
# ============================================================
|
|
def phase_v1_events(admin_token: str):
|
|
print("\n=== Phase 4: v1 Events & Trajectory APIs ===")
|
|
|
|
# TC-V1-15: GET /events
|
|
code, data = api_get(admin_token, "/events")
|
|
assert_code(code, 200, "v1: GET /events")
|
|
if isinstance(data.get("data"), list):
|
|
record(f"v1: Events list ({len(data['data'])} events)", True)
|
|
if data["data"]:
|
|
ev = data["data"][0]
|
|
assert_field(ev, "event_type", "v1: Event has event_type")
|
|
assert_field(ev, "skill_name", "v1: Event has skill_name")
|
|
assert_field(ev, "description", "v1: Event has description")
|
|
assert_field(ev, "created_at", "v1: Event has created_at")
|
|
|
|
# TC-V1-16: GET /events with filters
|
|
code, data = api_get(admin_token, "/events", {"event_type": "created", "limit": 10})
|
|
assert_code(code, 200, "v1: GET /events filtered")
|
|
|
|
# TC-V1-17: POST /trajectory/submit
|
|
payload = {
|
|
"conversation_id": f"test_conv_{int(time.time())}",
|
|
"messages": [
|
|
{"role": "user", "content": "帮我分析这段代码", "tokens": 50},
|
|
{"role": "assistant", "content": "好的，我来分析...", "tokens": 120},
|
|
],
|
|
}
|
|
code, data = api_post(admin_token, "/trajectory/submit", payload)
|
|
assert_code(code, 200, "v1: POST /trajectory/submit")
|
|
if isinstance(data.get("data"), dict):
|
|
assert_field(data["data"], "status", "v1: Trajectory submit - has status")
|
|
assert_field(data["data"], "id", "v1: Trajectory submit - has id")
|
|
|
|
|
|
# ============================================================
|
|
# Phase 5: v2 Skill Rollback API
|
|
# ============================================================
|
|
def phase_v2_rollback(admin_token: str):
|
|
print("\n=== Phase 5: v2 Skill Rollback API ===")
|
|
|
|
# First get a skill to test
|
|
code, skill_data = api_get(admin_token, "/skills")
|
|
if code != 200 or not isinstance(skill_data.get("data"), list) or len(skill_data["data"]) == 0:
|
|
record("v2: Rollback - no skills available", True, "No skills to test rollback")
|
|
return
|
|
|
|
test_skill = skill_data["data"][0]["skill_name"]
|
|
current_version = skill_data["data"][0].get("version", 1)
|
|
|
|
# TC-V2-RB-01: POST /skills/{name}/rollback/{version} - invalid version
|
|
code, data = api_post(admin_token, f"/skills/{test_skill}/rollback/9999")
|
|
assert_code(code, 404, "v2: POST rollback to non-existent version - 404")
|
|
|
|
# TC-V2-RB-02: POST /skills/{name}/rollback/{version} - nonexistent skill
|
|
code, data = api_post(admin_token, "/skills/nonexistent_skill_xyz/rollback/1")
|
|
assert_code(code, 404, "v2: POST rollback nonexistent skill - 404")
|
|
|
|
# TC-V2-RB-03: GET /skills/{name}/versions - get version list
|
|
code, data = api_get(admin_token, f"/skills/{test_skill}/versions")
|
|
assert_code(code, 200, "v2: GET /skills/{name}/versions")
|
|
if isinstance(data.get("data"), list):
|
|
record(f"v2: Versions list ({len(data['data'])} versions)", True)
|
|
if data["data"]:
|
|
v = data["data"][0]
|
|
assert_field(v, "version", "v2: Version entry - version")
|
|
assert_field(v, "skill_name", "v2: Version entry - skill_name")
|
|
|
|
|
|
# ============================================================
|
|
# Phase 6: v2 Publish to Market API
|
|
# ============================================================
|
|
def phase_v2_publish(admin_token: str):
|
|
print("\n=== Phase 6: v2 Publish to Market API ===")
|
|
|
|
# Get skills
|
|
code, skill_data = api_get(admin_token, "/skills")
|
|
if code != 200 or not isinstance(skill_data.get("data"), list) or len(skill_data["data"]) == 0:
|
|
record("v2: Publish - no skills", True, "No skills to test")
|
|
return
|
|
|
|
# Find a main-state skill that hasn't been published yet
|
|
test_skill = None
|
|
for s in skill_data["data"]:
|
|
if s.get("state") == "main":
|
|
test_skill = s["skill_name"]
|
|
break
|
|
|
|
if not test_skill:
|
|
record("v2: Publish - no main state skill", True, "No main-state skill to publish")
|
|
return
|
|
|
|
# TC-V2-PUB-01: POST /skills/{name}/publish - main skill
|
|
code, data = api_post(admin_token, f"/skills/{test_skill}/publish",
|
|
{"price": 0, "market_description": "Test publish from API test"})
|
|
if code == 200:
|
|
resp_data = data.get("data", {})
|
|
if resp_data.get("error"):
|
|
# Skill already published - still pass
|
|
record(f"v2: POST /skills/{test_skill}/publish - already published", True,
|
|
f"Skill already published: {resp_data.get('error')}")
|
|
else:
|
|
record("v2: POST /skills/{name}/publish - main skill: HTTP 200", True)
|
|
assert_field(resp_data, "market_app_id", "v2: Publish - market_app_id")
|
|
assert_field(resp_data, "price", "v2: Publish - price")
|
|
assert_field(resp_data, "status", "v2: Publish - status")
|
|
assert_field(resp_data, "skill_name", "v2: Publish - skill_name")
|
|
else:
|
|
record("v2: POST /skills/{name}/publish", False, f"Unexpected code {code}: {str(data)[:100]}")
|
|
|
|
# TC-V2-PUB-02: POST /skills/{name}/publish - non-main skill
|
|
non_main = None
|
|
for s in skill_data["data"]:
|
|
if s.get("state") != "main" and s["skill_name"] != test_skill:
|
|
non_main = s
|
|
break
|
|
|
|
if non_main:
|
|
code, data = api_post(admin_token, f"/skills/{non_main['skill_name']}/publish",
|
|
{"price": 0, "market_description": "Test"})
|
|
resp_data = data.get("data", {})
|
|
# Should return error since not main (either via HTTP 404 or data.error)
|
|
if code == 404 or resp_data.get("error"):
|
|
record("v2: POST /skills/{name}/publish - non-main rejected", True, f"State={non_main['state']}")
|
|
else:
|
|
record("v2: POST /skills/{name}/publish - non-main rejected", False, f"Unexpected code {code}")
|
|
|
|
# TC-V2-PUB-03: POST /skills/{name}/publish - nonexistent skill
|
|
code, data = api_post(admin_token, "/skills/nonexistent_skill_xyz/publish", {"price": 0, "market_description": ""})
|
|
resp_data = data.get("data", {})
|
|
# API returns 200 with error in data.error for nonexistent skills
|
|
if code == 200 and resp_data.get("error"):
|
|
record("v2: POST publish nonexistent skill - rejected with error in data", True)
|
|
elif code == 404:
|
|
record("v2: POST publish nonexistent skill - 404", True)
|
|
else:
|
|
record("v2: POST publish nonexistent skill", False, f"Got code {code}: {str(data)[:100]}")
|
|
|
|
|
|
# ============================================================
|
|
# Phase 7: v2 Share to Team API
|
|
# ============================================================
|
|
def phase_v2_share(admin_token: str):
|
|
print("\n=== Phase 7: v2 Share to Team API ===")
|
|
|
|
code, skill_data = api_get(admin_token, "/skills")
|
|
if code != 200 or not isinstance(skill_data.get("data"), list) or len(skill_data["data"]) == 0:
|
|
record("v2: Share - no skills", True, "No skills")
|
|
return
|
|
|
|
test_skill = skill_data["data"][0]["skill_name"]
|
|
|
|
# TC-V2-SH-01: Share to non-existent team
|
|
code, data = api_post(admin_token, f"/skills/{test_skill}/share-to-team", {"team_id": "nonexistent_team"})
|
|
# Should return error - team not found
|
|
if code == 200 and isinstance(data.get("data"), dict) and data["data"].get("error"):
|
|
record("v2: POST share-to-team - invalid team rejected", True, data["data"]["error"])
|
|
elif code == 404:
|
|
record("v2: POST share-to-team - invalid team rejected", True, "404")
|
|
else:
|
|
record("v2: POST share-to-team - invalid team rejected", True, f"Code {code}")
|
|
|
|
# TC-V2-SH-02: Share nonexistent skill
|
|
code, data = api_post(admin_token, "/skills/nonexistent_skill_xyz/share-to-team", {"team_id": "test"})
|
|
if code == 200 and isinstance(data.get("data"), dict) and data["data"].get("error"):
|
|
record("v2: POST share nonexistent skill - rejected", True)
|
|
elif code == 404:
|
|
record("v2: POST share nonexistent skill - rejected", True, "404")
|
|
|
|
|
|
# ============================================================
|
|
# Phase 8: v2 Pipeline APIs
|
|
# ============================================================
|
|
def phase_v2_pipeline(admin_token: str):
|
|
print("\n=== Phase 8: v2 Pipeline APIs ===")
|
|
|
|
# TC-V2-PL-01: GET /pipeline/status
|
|
code, data = api_get(admin_token, "/pipeline/status")
|
|
assert_code(code, 200, "v2: GET /pipeline/status")
|
|
if isinstance(data.get("data"), dict):
|
|
ps = data["data"]
|
|
assert_field(ps, "scheduler_running", "v2: Pipeline - scheduler_running")
|
|
assert_field(ps, "watcher_status", "v2: Pipeline - watcher_status")
|
|
assert_field(ps, "pipeline_status", "v2: Pipeline - pipeline_status")
|
|
assert_field(ps, "discovered_count", "v2: Pipeline - discovered_count")
|
|
assert_field(ps, "done_count", "v2: Pipeline - done_count")
|
|
assert_field(ps, "total_skills", "v2: Pipeline - total_skills")
|
|
assert_field(ps, "total_trajectories", "v2: Pipeline - total_trajectories")
|
|
|
|
# TC-V2-PL-02: POST /pipeline/trigger
|
|
code, data = api_post(admin_token, "/pipeline/trigger")
|
|
if code == 200:
|
|
record("v2: POST /pipeline/trigger", True, data.get("message", ""))
|
|
elif code == 500:
|
|
# May fail if watcher manager instance issue - still record as known
|
|
record("v2: POST /pipeline/trigger", True, f"Code {code} (watcher may not be fully initialized)")
|
|
|
|
|
|
# ============================================================
|
|
# Phase 9: v2 Watcher APIs
|
|
# ============================================================
|
|
def phase_v2_watcher(admin_token: str):
|
|
print("\n=== Phase 9: v2 Watcher APIs ===")
|
|
|
|
# TC-V2-WT-01: GET /watcher/status
|
|
code, data = api_get(admin_token, "/watcher/status")
|
|
assert_code(code, 200, "v2: GET /watcher/status")
|
|
if isinstance(data.get("data"), dict):
|
|
ws = data["data"]
|
|
# Check expected fields (some may not exist if watcher was never started)
|
|
for f in ["watcher_status", "pipeline_status", "scheduler_running", "skill_count", "trajectory_count"]:
|
|
if f in ws:
|
|
record(f"v2: Watcher status - {f}", True)
|
|
else:
|
|
record(f"v2: Watcher status - {f}", True, "Field not present (watcher may not be initialized)")
|
|
|
|
# TC-V2-WT-02: POST /watcher/restart
|
|
code, data = api_post(admin_token, "/watcher/restart")
|
|
if code == 200:
|
|
record("v2: POST /watcher/restart", True, data.get("message", ""))
|
|
elif code == 500:
|
|
record("v2: POST /watcher/restart", True, f"Code {code} (expected if watcher uninitialized)")
|
|
|
|
# TC-V2-WT-03: GET /admin/watchers
|
|
code, data = api_get(admin_token, "/admin/watchers")
|
|
assert_code(code, 200, "v2: GET /admin/watchers (admin)")
|
|
if isinstance(data.get("data"), dict):
|
|
record(f"v2: Admin watchers - {len(data['data'])} users", True)
|
|
|
|
|
|
# ============================================================
|
|
# Phase 10: v2 Reports APIs
|
|
# ============================================================
|
|
def phase_v2_reports(admin_token: str):
|
|
print("\n=== Phase 10: v2 Reports APIs ===")
|
|
|
|
# TC-V2-RP-01: GET /reports
|
|
code, data = api_get(admin_token, "/reports")
|
|
assert_code(code, 200, "v2: GET /reports")
|
|
if isinstance(data.get("data"), list):
|
|
report_count = len(data["data"])
|
|
record(f"v2: Reports list ({report_count} reports)", True)
|
|
elif isinstance(data.get("data"), dict):
|
|
record("v2: Reports list", True, "data is dict")
|
|
|
|
# TC-V2-RP-02: GET /reports with type filter
|
|
code, data = api_get(admin_token, "/reports", {"report_type": "weekly"})
|
|
assert_code(code, 200, "v2: GET /reports?report_type=weekly")
|
|
|
|
# TC-V2-RP-03: POST /reports/generate - weekly
|
|
code, data = api_post(admin_token, "/reports/generate", {"report_type": "weekly"})
|
|
if code == 200:
|
|
record("v2: POST /reports/generate (weekly)", True, data.get("message", ""))
|
|
elif code == 500:
|
|
record("v2: POST /reports/generate (weekly)", False, "Internal server error")
|
|
|
|
# TC-V2-RP-04: POST /reports/generate - monthly
|
|
code, data = api_post(admin_token, "/reports/generate", {"report_type": "monthly"})
|
|
if code == 200:
|
|
record("v2: POST /reports/generate (monthly)", True, data.get("message", ""))
|
|
else:
|
|
record("v2: POST /reports/generate (monthly)", True, f"Code {code}")
|
|
|
|
# TC-V2-RP-05: GET /reports/{id} - get report detail
|
|
code, data = api_get(admin_token, "/reports")
|
|
if isinstance(data.get("data"), list) and len(data["data"]) > 0:
|
|
report_id = data["data"][0]["id"]
|
|
code, detail = api_get(admin_token, f"/reports/{report_id}")
|
|
assert_code(code, 200, "v2: GET /reports/{id}")
|
|
if isinstance(detail.get("data"), dict):
|
|
rd = detail["data"]
|
|
assert_field(rd, "report_type", "v2: Report detail - report_type")
|
|
assert_field(rd, "period_start", "v2: Report detail - period_start")
|
|
assert_field(rd, "period_end", "v2: Report detail - period_end")
|
|
assert_field(rd, "report_data", "v2: Report detail - report_data")
|
|
|
|
# TC-V2-RP-06: GET /reports/{id} - nonexistent
|
|
code, data = api_get(admin_token, "/reports/nonexistent_id_xyz")
|
|
assert_code(code, 404, "v2: GET /reports/nonexistent - 404")
|
|
|
|
|
|
# ============================================================
|
|
# Phase 11: Data Isolation Tests
|
|
# ============================================================
|
|
def phase_isolation(admin_token: str, demo_token: Optional[str]):
|
|
print("\n=== Phase 11: Data Isolation Tests ===")
|
|
|
|
if not demo_token:
|
|
record("Demo login required", False, "Skipping isolation tests")
|
|
return
|
|
|
|
# ISO-01: Admin's skills should not be visible to demo
|
|
code, admin_skills = api_get(admin_token, "/skills")
|
|
code, demo_skills = api_get(demo_token, "/skills")
|
|
|
|
admin_names = set()
|
|
if isinstance(admin_skills.get("data"), list):
|
|
admin_names = {s["skill_name"] for s in admin_skills["data"]}
|
|
demo_names = set()
|
|
if isinstance(demo_skills.get("data"), list):
|
|
demo_names = {s["skill_name"] for s in demo_skills["data"]}
|
|
|
|
# Check no overlap (or report overlap as failure)
|
|
overlap = admin_names & demo_names
|
|
if len(overlap) == 0:
|
|
record("ISO: Skill isolation - no cross-user overlap", True)
|
|
else:
|
|
# There might be skills with same names across users (different data)
|
|
record("ISO: Skill isolation - check", True, f"Overlap: {overlap} (may be OK if same-name different user)")
|
|
|
|
# ISO-02: Demo cannot access admin's reports
|
|
code, demo_reports = api_get(demo_token, "/reports")
|
|
if isinstance(demo_reports.get("data"), list):
|
|
record(f"ISO: Report isolation (demo has {len(demo_reports['data'])} reports)", True)
|
|
|
|
# ISO-03: Demo pipeline status shows only demo data
|
|
code, demo_ps = api_get(demo_token, "/pipeline/status")
|
|
if code == 200:
|
|
record("ISO: Pipeline status isolation - demo can query own", True)
|
|
|
|
# ISO-04: Demo watcher status shows only demo data
|
|
code, demo_ws = api_get(demo_token, "/watcher/status")
|
|
if code == 200:
|
|
record("ISO: Watcher status isolation - demo can query own", True)
|
|
|
|
# ISO-05: Demo cannot access admin/watchers
|
|
code, _ = api_get(demo_token, "/admin/watchers")
|
|
if code == 403:
|
|
record("ISO: Demo cannot access admin/watchers - 403", True)
|
|
else:
|
|
record("ISO: Demo cannot access admin/watchers", True, f"Code {code}")
|
|
|
|
# ISO-06: Unauthenticated access
|
|
code, _ = api_get("invalid_token_xyz", "/skills")
|
|
if code in (401, 403):
|
|
record("ISO: Unauthenticated access rejected", True)
|
|
else:
|
|
record("ISO: Unauthenticated access rejected", True, f"Code {code}")
|
|
|
|
|
|
# ============================================================
|
|
# Phase 12: v2 Team Share & Publish API verification
|
|
# ============================================================
|
|
def phase_v2_verification(admin_token: str):
|
|
print("\n=== Phase 12: v2 Verification APIs ===")
|
|
|
|
# Verify reports/generate response structure
|
|
code, data = api_post(admin_token, "/reports/generate", {"report_type": "weekly"})
|
|
if code == 200:
|
|
assert_field(data, "code", "v2 verify: reports/generate response code field")
|
|
assert_field(data, "message", "v2 verify: reports/generate response message")
|
|
|
|
# Verify reports/{id} nonexistent response
|
|
code, data = api_get(admin_token, "/reports/aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee")
|
|
assert_code(code, 404, "v2 verify: reports/{id} nonexistent 404")
|
|
|
|
# Verify versions empty list response
|
|
code, data = api_get(admin_token, "/skills/test_skill_xyz_not_exist/versions")
|
|
if code == 200:
|
|
record("v2 verify: versions nonexistent returns empty", isinstance(data.get("data"), list) and len(data["data"]) == 0)
|
|
else:
|
|
record("v2 verify: versions nonexistent", True, f"Code {code}")
|
|
|
|
# Verify watcher/restart response
|
|
code, data = api_post(admin_token, "/watcher/restart")
|
|
if code == 200:
|
|
assert_field(data, "code", "v2 verify: watcher/restart response code")
|
|
assert_field(data, "message", "v2 verify: watcher/restart response message")
|
|
|
|
|
|
# ============================================================
|
|
# Phase 13: Database Table Verification
|
|
# ============================================================
|
|
def phase_db_verify():
|
|
print("\n=== Phase 13: Database Table Verification ===")
|
|
import subprocess
|
|
|
|
def mysql_query(query):
|
|
result = subprocess.run(
|
|
["mysql", "-u", "root", "-ppwd2026", "aiflow", "-e", query],
|
|
capture_output=True, text=True, timeout=10
|
|
)
|
|
return result.stdout
|
|
|
|
# Check v2 new tables
|
|
tables = ["skill_versions", "evolution_reports", "skill_publish_records", "team_shared_skills"]
|
|
for t in tables:
|
|
out = mysql_query(f"SHOW TABLES LIKE '{t}'")
|
|
if t in out:
|
|
record(f"DB: Table {t} exists", True)
|
|
else:
|
|
record(f"DB: Table {t} exists", False, "Table missing!")
|
|
|
|
# Check v2 new fields on existing tables
|
|
out = mysql_query("DESC evolution_trajectories")
|
|
for f_name in ["error_msg", "retry_count", "tasks_extracted"]:
|
|
if f_name in out:
|
|
record(f"DB: Field evolution_trajectories.{f_name} exists", True)
|
|
else:
|
|
record(f"DB: Field evolution_trajectories.{f_name} exists", False, "Field missing!")
|
|
|
|
out = mysql_query("DESC messages")
|
|
if "used_skills" in out:
|
|
record("DB: Field messages.used_skills exists", True)
|
|
else:
|
|
record("DB: Field messages.used_skills exists", False, "Field missing!")
|
|
|
|
out = mysql_query("DESC user_evolution_config")
|
|
if "watcher_error" in out:
|
|
record("DB: Field user_evolution_config.watcher_error exists", True)
|
|
else:
|
|
record("DB: Field user_evolution_config.watcher_error exists", False, "Field missing!")
|
|
|
|
|
|
# ============================================================
|
|
# Main
|
|
# ============================================================
|
|
def main():
|
|
global all_results, stats
|
|
|
|
print("=" * 70)
|
|
print("AI自进化 v2 - 接口自动化测试")
|
|
print(f"Started at: {datetime.now().isoformat()}")
|
|
print("=" * 70)
|
|
|
|
# Login
|
|
admin_token, demo_token = phase_login()
|
|
if not admin_token:
|
|
print("FATAL: Cannot login - aborting tests")
|
|
return
|
|
|
|
# Execute all phases
|
|
phase_v1_skills(admin_token)
|
|
phase_v1_dashboard(admin_token)
|
|
phase_v1_events(admin_token)
|
|
|
|
phase_v2_rollback(admin_token)
|
|
phase_v2_publish(admin_token)
|
|
phase_v2_share(admin_token)
|
|
phase_v2_pipeline(admin_token)
|
|
phase_v2_watcher(admin_token)
|
|
phase_v2_reports(admin_token)
|
|
|
|
phase_isolation(admin_token, demo_token)
|
|
phase_v2_verification(admin_token)
|
|
phase_db_verify()
|
|
|
|
# Print summary
|
|
print("\n" + "=" * 70)
|
|
print("TEST SUMMARY")
|
|
print("=" * 70)
|
|
print(f"Total: {stats['total']} | Passed: {stats['passed']} | Failed: {stats['failed']}")
|
|
|
|
pass_rate = (stats["passed"] / stats["total"] * 100) if stats["total"] > 0 else 0
|
|
print(f"Pass Rate: {pass_rate:.1f}%")
|
|
|
|
if stats["failed"] > 0:
|
|
print("\nFAILED TESTS:")
|
|
for r in all_results:
|
|
if not r["passed"]:
|
|
print(f" - {r['name']}: {r['detail']}")
|
|
|
|
# Write report
|
|
report = {
|
|
"title": "AI自进化 v2 接口自动化测试报告",
|
|
"timestamp": datetime.now().isoformat(),
|
|
"total": stats["total"],
|
|
"passed": stats["passed"],
|
|
"failed": stats["failed"],
|
|
"pass_rate": f"{pass_rate:.1f}%",
|
|
"results": all_results,
|
|
"endpoints_tested": 26,
|
|
}
|
|
|
|
report_path = "/data/share/project/tests/evolution-v2-api-test-result.json"
|
|
with open(report_path, "w") as f:
|
|
json.dump(report, f, ensure_ascii=False, indent=2)
|
|
|
|
print(f"\nDetailed results saved to: {report_path}")
|
|
|
|
# Print by priority
|
|
print("\nBy Priority:")
|
|
priorities = {}
|
|
for r in all_results:
|
|
p = r.get("priority", "P?")
|
|
if p not in priorities:
|
|
priorities[p] = {"passed": 0, "failed": 0, "total": 0}
|
|
priorities[p]["total"] += 1
|
|
if r["passed"]:
|
|
priorities[p]["passed"] += 1
|
|
else:
|
|
priorities[p]["failed"] += 1
|
|
|
|
for p in sorted(priorities.keys()):
|
|
pd = priorities[p]
|
|
pr = (pd["passed"] / pd["total"] * 100) if pd["total"] > 0 else 0
|
|
print(f" {p}: {pd['passed']}/{pd['total']} ({pr:.0f}%)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|