Yuzhiran Dev
9c37c9a574
Phase 4: org_id 注入 JWT/API 过滤/组织管理 CRUD/前端组织列 测试: tests/test_phase_upgrades.py 97项全覆盖 CSS: theme-modern.css 共享 mobile-card-list/status-dot/search-bar 等模式 修复: initial_data.py LLM配置 NOT NULL 约束, TopicResponse 含 org_id
259 lines
12 KiB
Python
259 lines
12 KiB
Python
"""四阶段升级全面测试"""
|
|
import uvicorn, time, threading, requests, json, os, sys
|
|
from pathlib import Path
|
|
|
|
os.environ.setdefault("USE_POSTGRES", "false")
|
|
backend_dir = str(Path(__file__).parent.parent / "platform" / "backend")
|
|
sys.path.insert(0, backend_dir)
|
|
os.chdir(backend_dir)
|
|
|
|
results = {"pass": 0, "fail": 0}
|
|
def test(name, ok, detail=""):
|
|
status = chr(10003) if ok else chr(10007)
|
|
results["pass" if ok else "fail"] += 1
|
|
print(f" {status} {name}")
|
|
if detail: print(f" {detail}")
|
|
|
|
# ===== 0. 编译检查 =====
|
|
print("\n====== 阶段 0: 编译检查 ======")
|
|
root = Path(__file__).parent.parent
|
|
files = [
|
|
"platform/backend/app/models.py", "platform/backend/app/api/auth.py",
|
|
"platform/backend/app/api/topics.py", "platform/backend/app/api/articles.py",
|
|
"platform/backend/app/api/calendar.py", "platform/backend/app/api/metrics.py",
|
|
"platform/backend/app/api/publishing.py", "platform/backend/app/api/system.py",
|
|
"platform/backend/app/api/admin.py", "platform/backend/app/api/tasks.py",
|
|
"platform/backend/app/schemas.py", "platform/backend/app/database.py",
|
|
"platform/backend/app/initial_data.py", "platform/backend/app/main.py",
|
|
"platform/backend/app/core/nvidia_client.py", "platform/backend/app/core/scheduler.py",
|
|
"scripts/collector.py", "scripts/sync_metrics.py", "scripts/creator.py",
|
|
]
|
|
for f in files:
|
|
try:
|
|
src = (root / f).read_text()
|
|
compile(src, f, "exec")
|
|
test(f"编译 {Path(f).name}", True)
|
|
except SyntaxError as e:
|
|
test(f"编译 {Path(f).name}", False, str(e))
|
|
|
|
# ===== 1. 启动服务 =====
|
|
print("\n====== 阶段 0: API 服务启动 ======")
|
|
def start_server():
|
|
uvicorn.run("app.main:app", host="127.0.0.1", port=18504, log_level="error")
|
|
t = threading.Thread(target=start_server, daemon=True)
|
|
t.start()
|
|
time.sleep(4)
|
|
|
|
base = "http://127.0.0.1:18504"
|
|
try:
|
|
r = requests.get(f"{base}/api/system/status", timeout=5)
|
|
test("服务可访问", r.status_code == 200)
|
|
except Exception as e:
|
|
test("服务可访问", False, str(e))
|
|
exit(1)
|
|
|
|
r = requests.post(f"{base}/api/auth/login", json={"username": "admin", "password": "admin123"}, timeout=5)
|
|
token = r.json().get("token", "") if r.status_code == 200 else ""
|
|
test("管理员登录", r.status_code == 200 and bool(token))
|
|
if not token:
|
|
exit(1)
|
|
auth = {"Authorization": f"Bearer {token}"}
|
|
|
|
# Decode JWT to check org_id
|
|
import jwt
|
|
payload = jwt.decode(token, options={"verify_signature": False})
|
|
test("JWT含org_id", payload.get("org_id") == "default", f"org_id={payload.get('org_id')}")
|
|
test("JWT含username", payload.get("username") == "admin")
|
|
test("JWT含role", payload.get("role") == "admin")
|
|
|
|
# ===== Phase 1.1: 配图集成 =====
|
|
print("\n====== Phase 1.1: 配图集成 ======")
|
|
r = requests.get(f"{base}/api/topics", headers=auth, timeout=5)
|
|
test("选题列表可访问", r.status_code == 200)
|
|
topics = r.json()
|
|
test("有选题数据", len(topics) > 0, f"共{len(topics)}个")
|
|
if topics:
|
|
tid = topics[0]["id"]
|
|
r = requests.get(f"{base}/api/articles/{tid}/images", headers=auth, timeout=5)
|
|
test("配图API可访问", r.status_code == 200)
|
|
test("配图返回dict", isinstance(r.json(), dict))
|
|
r = requests.get(f"{base}/api/articles/{tid}/preview", headers=auth, timeout=5)
|
|
test("预览API可访问", r.status_code == 200 or r.status_code == 404)
|
|
|
|
# ===== Phase 1.3: 前端三态 =====
|
|
print("\n====== Phase 1.3: 前端三态UX ======")
|
|
pages = ["index.html", "topics.html", "metrics.html", "calendar.html",
|
|
"articles.html", "tasks.html", "admin.html", "users.html"]
|
|
for page in pages:
|
|
p = root / "platform" / "frontend" / page
|
|
html = p.read_text(encoding="utf-8")
|
|
has_loading = "loading" in html.lower()
|
|
has_empty = "empty" in html.lower() or "暂无" in html
|
|
has_error = "error" in html.lower() or "失败" in html or "catch" in html
|
|
ok = has_loading and has_empty and has_error
|
|
test(f"{page}: 三态 ({'✓' if ok else '✗'})", ok,
|
|
f"loading={has_loading} empty={has_empty} error={has_error}")
|
|
|
|
# ===== Phase 2.1: 多平台发布 =====
|
|
print("\n====== Phase 2.1: 多平台发布 ======")
|
|
r = requests.post(f"{base}/api/topics", headers=auth, json={
|
|
"title": f"测试选题_{int(time.time())}", "tags": ["test"]
|
|
}, timeout=5)
|
|
test("创建测试选题", r.status_code == 200)
|
|
new_tid = r.json().get("id", "")
|
|
if new_tid:
|
|
r = requests.put(f"{base}/api/topics/{new_tid}", headers=auth,
|
|
json={"status": "ready"}, timeout=5)
|
|
test("设置待发布状态", r.status_code == 200)
|
|
r = requests.post(f"{base}/api/publishing/create", headers=auth, json={
|
|
"topic_id": new_tid,
|
|
"platforms": ["zhihu", "wechat"]
|
|
}, timeout=5)
|
|
test("多平台发布API", r.status_code == 200)
|
|
data = r.json()
|
|
test("发布返回ok", data.get("ok") == True)
|
|
test("发布结果含平台列表", len(data.get("results", [])) > 0)
|
|
if data.get("results"):
|
|
test("第一平台发布成功", data["results"][0]["status"] == "success")
|
|
|
|
# ===== Phase 2.2: 日历关联 =====
|
|
print("\n====== Phase 2.2: 日历选题关联 ======")
|
|
r = requests.get(f"{base}/api/calendar", headers=auth, params={"year": 2026, "month": 5}, timeout=5)
|
|
test("日历API可访问", r.status_code == 200)
|
|
entries = r.json()
|
|
if entries:
|
|
e = entries[0]
|
|
test("含topic_status字段", "topic_status" in e, f"值={e.get('topic_status')}")
|
|
test("含platform_icon字段", "platform_icon" in e, f"值={e.get('platform_icon')}")
|
|
|
|
# ===== Phase 2.3: SVG图标 + H5 =====
|
|
print("\n====== Phase 2.3: SVG图标 + H5响应式 ======")
|
|
icon_js = (root / "platform" / "frontend" / "icon-components.js").read_text(encoding="utf-8")
|
|
import re
|
|
icon_names = set(re.findall(r'Icon[A-Z][a-zA-Z]+', icon_js))
|
|
test("icon-components.js注册图标", len(icon_names) >= 20, f"共{len(icon_names)}个图标: {sorted(icon_names)[:10]}...")
|
|
test("含IconCheck组件", "IconCheck" in icon_js)
|
|
test("含IconClose组件", "IconClose" in icon_js)
|
|
test("含IconSetting组件", "IconSetting" in icon_js)
|
|
test("含IconDashboard组件", "IconDashboard" in icon_js)
|
|
|
|
mobile_pages = ["topics.html", "metrics.html", "admin.html", "users.html", "articles.html"]
|
|
for page in mobile_pages:
|
|
html = (root / "platform" / "frontend" / page).read_text()
|
|
has_card_list = "card-list" in html or "card_list" in html
|
|
has_media = "@media" in html
|
|
test(f"{page}: H5卡片+响应式", has_card_list and has_media)
|
|
|
|
# ===== Phase 3.1: 指标同步 =====
|
|
print("\n====== Phase 3.1: 指标同步 ======")
|
|
sync_py = (root / "scripts" / "sync_metrics.py").read_text()
|
|
test("sync_metrics.py 存在", True)
|
|
test("含估计算法", "compliance_score" in sync_py or "estimate" in sync_py or "platform_multiplier" in sync_py)
|
|
test("含upsert逻辑", "upsert" in sync_py.lower() or "on_conflict" in sync_py or "existing" in sync_py)
|
|
|
|
scheduler_py = (root / "platform" / "backend" / "app" / "core" / "scheduler.py").read_text()
|
|
test("调度含scheduled_metrics_sync", "scheduled_metrics_sync" in scheduler_py)
|
|
|
|
r = requests.get(f"{base}/api/metrics/entries", headers=auth, timeout=5)
|
|
test("指标API可访问", r.status_code == 200)
|
|
test("指标列表是list", isinstance(r.json(), list))
|
|
|
|
# ===== Phase 3.2: Chart.js 看板 =====
|
|
print("\n====== Phase 3.2: Chart.js 数据看板 ======")
|
|
chart_js = root / "platform" / "frontend" / "chart.umd.min.js"
|
|
test("chart.umd.min.js 存在", chart_js.exists())
|
|
if chart_js.exists():
|
|
test("chart文件非空", chart_js.stat().st_size > 10000, f"大小={chart_js.stat().st_size}B")
|
|
|
|
r = requests.get(f"{base}/api/metrics/dashboard", headers=auth, timeout=5)
|
|
test("数据看板API", r.status_code == 200)
|
|
db_data = r.json()
|
|
test("含total_topics", "total_topics" in db_data, f"值={db_data.get('total_topics')}")
|
|
test("含topics_by_status", "topics_by_status" in db_data)
|
|
test("含total_views", "total_views" in db_data)
|
|
test("含total_likes", "total_likes" in db_data)
|
|
test("含avg_engagement_rate", "avg_engagement_rate" in db_data)
|
|
test("含top_topics", "top_topics" in db_data)
|
|
test("含recent_metrics", "recent_metrics" in db_data)
|
|
|
|
r = requests.get(f"{base}/api/metrics/trend", headers=auth, timeout=5)
|
|
test("趋势API", r.status_code == 200, f"返回{len(r.json())}条")
|
|
r = requests.get(f"{base}/api/metrics/by-platform", headers=auth, timeout=5)
|
|
test("平台对比API", r.status_code == 200, f"返回{len(r.json())}个平台")
|
|
|
|
metrics_html = (root / "platform" / "frontend" / "metrics.html").read_text()
|
|
test("metrics.html含Chart.js引用", "chart.umd.min.js" in metrics_html or "chart.js" in metrics_html.lower())
|
|
test("含折线图初始化", "new Chart" in metrics_html or "Chart(" in metrics_html)
|
|
test("含环形图初始化", "doughnut" in metrics_html or "doughnut" in metrics_html.lower())
|
|
test("含柱状图初始化", "bar" in metrics_html)
|
|
|
|
# ===== Phase 4: 多租户 =====
|
|
print("\n====== Phase 4: 多租户隔离 ======")
|
|
|
|
# 4a. 模型有org_id
|
|
models_py = (root / "platform" / "backend" / "app" / "models.py").read_text()
|
|
test("User模型含org_id", "org_id" in models_py)
|
|
test("Topic模型含org_id", "org_id" in models_py)
|
|
|
|
# 4b. JWT含org_id (已在前面验证)
|
|
# 4c. 选题创建继承org
|
|
if new_tid:
|
|
r = requests.get(f"{base}/api/topics/{new_tid}", headers=auth, timeout=5)
|
|
test("新选题含org_id", r.json().get("org_id") == "default" or "org_id" in r.json())
|
|
|
|
# 4d. 组织管理CRUD
|
|
r = requests.get(f"{base}/api/admin/orgs", headers=auth, timeout=5)
|
|
test("组织列表API", r.status_code == 200)
|
|
orgs = r.json()
|
|
test("默认组织存在", any(o.get("org_id") == "default" for o in orgs))
|
|
|
|
org_id = f"test_org_{int(time.time())}"
|
|
r = requests.post(f"{base}/api/admin/orgs", headers=auth, json={
|
|
"org_id": org_id, "name": "测试组织", "description": "自动化测试"
|
|
}, timeout=5)
|
|
test("创建组织", r.status_code == 200, f"org_id={org_id}")
|
|
|
|
r = requests.put(f"{base}/api/admin/orgs/{org_id}", headers=auth,
|
|
json={"name": "测试组织(已更新)"}, timeout=5)
|
|
test("更新组织", r.status_code == 200)
|
|
|
|
r = requests.delete(f"{base}/api/admin/orgs/{org_id}", headers=auth, timeout=5)
|
|
test("删除组织", r.status_code == 200)
|
|
|
|
# 4e. users.html org_id列
|
|
users_html = (root / "platform" / "frontend" / "users.html").read_text()
|
|
test("users.html含组织列", "org_id" in users_html)
|
|
test("users.html表含组织头", "组织" in users_html)
|
|
|
|
# 4f. admin.html组织管理标签
|
|
admin_html = (root / "platform" / "frontend" / "admin.html").read_text()
|
|
test("admin.html含组织标签", "组织管理" in admin_html)
|
|
test("admin.html含组织CRUD函数", "loadOrgs" in admin_html)
|
|
|
|
# 4g. 数据库迁移
|
|
db_py = (root / "platform" / "backend" / "app" / "database.py").read_text()
|
|
test("database.py含org_id迁移", "users.*org_id" in db_py.replace(" ", "").replace("\n", "") or
|
|
("org_id" in db_py and "ALTER TABLE" in db_py))
|
|
|
|
# ===== 统一CSS检查 =====
|
|
print("\n====== 统一CSS检查 ======")
|
|
theme_css = (root / "platform" / "frontend" / "theme-modern.css").read_text()
|
|
test("theme-modern.css存在", True)
|
|
test("含响应式@media 768px", "@media (max-width: 768px)" in theme_css)
|
|
test("含统一卡片类.card", ".card" in theme_css)
|
|
test("含统一布局.main-content", ".main-content" in theme_css)
|
|
test("含统一.page-header", ".page-header" in theme_css)
|
|
test("含空状态.empty-state", ".empty-state" in theme_css)
|
|
test("含加载状态.loading-state", ".loading-state" in theme_css)
|
|
|
|
mobile_patterns = ["mobile-card", "padding-bottom: 80px", "el-dialog", "el-button"]
|
|
present = sum(1 for p in mobile_patterns if p in theme_css)
|
|
test(f"H5触控优化 ({present}/{len(mobile_patterns)})", present >= 2, f"含: mobile-card/padding-bottom/el-dialog/el-button")
|
|
|
|
# ===== 总结 =====
|
|
total = results["pass"] + results["fail"]
|
|
print(f"\n{'='*50}")
|
|
print(f"阶段测试完成: {results['pass']}/{total} 通过, {results['fail']} 失败")
|
|
if results["fail"] > 0:
|
|
exit(1)
|