166 lines
5.7 KiB
Python
166 lines
5.7 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
app_vx.py 接口测试脚本
|
||
用于验证小程序后端 API 是否正常工作
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
from datetime import datetime
|
||
|
||
# 配置
|
||
BASE_URL = "http://localhost:5002" # 服务器本地测试（推荐）
|
||
# BASE_URL = "http://110.42.32.207:5002" # 外网 HTTP 直连（需防火墙开放5002）
|
||
|
||
# 颜色输出
|
||
class Colors:
|
||
GREEN = '\033[92m'
|
||
RED = '\033[91m'
|
||
YELLOW = '\033[93m'
|
||
BLUE = '\033[94m'
|
||
END = '\033[0m'
|
||
|
||
def print_result(name, success, message="", data=None):
|
||
"""打印测试结果"""
|
||
status = f"{Colors.GREEN}✓ PASS{Colors.END}" if success else f"{Colors.RED}✗ FAIL{Colors.END}"
|
||
print(f"{status} | {name}")
|
||
if message:
|
||
print(f" {Colors.YELLOW}{message}{Colors.END}")
|
||
if data and not success:
|
||
print(f" Response: {json.dumps(data, ensure_ascii=False)[:200]}")
|
||
|
||
def test_api(name, method, endpoint, expected_code=200, token=None, data=None, params=None):
|
||
"""通用 API 测试函数"""
|
||
url = f"{BASE_URL}{endpoint}"
|
||
headers = {"Content-Type": "application/json"}
|
||
if token:
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
|
||
try:
|
||
if method == "GET":
|
||
resp = requests.get(url, headers=headers, params=params, timeout=10, verify=False)
|
||
elif method == "POST":
|
||
resp = requests.post(url, headers=headers, json=data, timeout=10, verify=False)
|
||
else:
|
||
print_result(name, False, f"不支持的方法: {method}")
|
||
return None
|
||
|
||
success = resp.status_code == expected_code
|
||
try:
|
||
resp_data = resp.json()
|
||
except:
|
||
resp_data = {"raw": resp.text[:100]}
|
||
|
||
msg = f"HTTP {resp.status_code}"
|
||
if "code" in resp_data:
|
||
msg += f", code={resp_data.get('code')}"
|
||
if "message" in resp_data:
|
||
msg += f", {resp_data.get('message', '')[:50]}"
|
||
|
||
print_result(name, success, msg, resp_data if not success else None)
|
||
return resp_data if success else None
|
||
|
||
except requests.exceptions.ConnectionError:
|
||
print_result(name, False, "连接失败 - 服务器可能未启动")
|
||
return None
|
||
except requests.exceptions.Timeout:
|
||
print_result(name, False, "请求超时")
|
||
return None
|
||
except Exception as e:
|
||
print_result(name, False, str(e))
|
||
return None
|
||
|
||
|
||
def main():
|
||
print(f"\n{Colors.BLUE}{'='*60}{Colors.END}")
|
||
print(f"{Colors.BLUE} app_vx.py 接口测试 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}{Colors.END}")
|
||
print(f"{Colors.BLUE} Base URL: {BASE_URL}{Colors.END}")
|
||
print(f"{Colors.BLUE}{'='*60}{Colors.END}\n")
|
||
|
||
# 禁用 SSL 警告
|
||
import urllib3
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
|
||
# ==================== 1. 公开接口测试 ====================
|
||
print(f"{Colors.BLUE}[1] 公开接口测试（无需登录）{Colors.END}")
|
||
print("-" * 40)
|
||
|
||
# 首页数据
|
||
test_api("首页数据 /api/home", "GET", "/api/home")
|
||
|
||
# 事件列表
|
||
test_api("事件列表 /api/events", "GET", "/api/events", params={"page": 1, "per_page": 5})
|
||
|
||
# 日历事件
|
||
test_api("日历事件 /api/calendar/events", "GET", "/api/calendar/events",
|
||
params={"start": "2025-12-01", "end": "2025-12-31"})
|
||
|
||
# 筛选选项
|
||
test_api("筛选选项 /api/filter/options", "GET", "/api/filter/options")
|
||
|
||
# 板块层级
|
||
test_api("板块层级 /api/sector/hierarchy", "GET", "/api/sector/hierarchy")
|
||
|
||
# 系统状态
|
||
test_api("ClickHouse连接池状态", "GET", "/api/system/clickhouse-pool-status")
|
||
|
||
# ==================== 2. 认证相关测试 ====================
|
||
print(f"\n{Colors.BLUE}[2] 认证相关接口{Colors.END}")
|
||
print("-" * 40)
|
||
|
||
# Token 验证（无效 token）
|
||
test_api("Token验证（无效token）", "POST", "/api/auth/verify-token",
|
||
expected_code=401, data={"token": "invalid_token_12345"})
|
||
|
||
# ==================== 3. 需要认证的接口（预期失败）====================
|
||
print(f"\n{Colors.BLUE}[3] 需要认证的接口（无token，预期401）{Colors.END}")
|
||
print("-" * 40)
|
||
|
||
# 用户信息
|
||
test_api("用户信息 /api/user/profile", "GET", "/api/user/profile", expected_code=401)
|
||
|
||
# 会员状态
|
||
test_api("会员状态 /api/membership/status", "GET", "/api/membership/status", expected_code=401)
|
||
|
||
# 订阅信息
|
||
test_api("订阅信息 /api/subscription/info", "GET", "/api/subscription/info", expected_code=401)
|
||
|
||
# 用户活动
|
||
test_api("用户活动 /api/user/activities", "GET", "/api/user/activities", expected_code=401)
|
||
|
||
# ==================== 4. 数据接口测试 ====================
|
||
print(f"\n{Colors.BLUE}[4] 数据接口测试{Colors.END}")
|
||
print("-" * 40)
|
||
|
||
# K线数据
|
||
test_api("K线数据 600519", "GET", "/api/stock/600519/kline",
|
||
params={"period": "daily", "limit": 30})
|
||
|
||
# 分钟图
|
||
test_api("分钟图 600519", "GET", "/api/stock/600519/minute-chart")
|
||
|
||
# 事件详情（假设事件ID=1存在）
|
||
test_api("事件相关股票详情", "GET", "/api/event/1/related-stocks-detail")
|
||
|
||
# 事件评论
|
||
test_api("事件评论", "GET", "/api/event/1/comments", params={"page": 1})
|
||
|
||
# ==================== 5. 协议接口 ====================
|
||
print(f"\n{Colors.BLUE}[5] 其他接口{Colors.END}")
|
||
print("-" * 40)
|
||
|
||
# 协议列表
|
||
test_api("协议列表 /api/agreements", "GET", "/api/agreements")
|
||
|
||
# 日历事件统计
|
||
test_api("日历事件统计", "GET", "/api/calendar-event-counts")
|
||
|
||
print(f"\n{Colors.BLUE}{'='*60}{Colors.END}")
|
||
print(f"{Colors.BLUE} 测试完成{Colors.END}")
|
||
print(f"{Colors.BLUE}{'='*60}{Colors.END}\n")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|