115 lines
4.2 KiB
Python
115 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
測試管理後台 API
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
|
||
# Fix encoding for Windows console
|
||
if sys.stdout.encoding != 'utf-8':
|
||
sys.stdout.reconfigure(encoding='utf-8')
|
||
if sys.stderr.encoding != 'utf-8':
|
||
sys.stderr.reconfigure(encoding='utf-8')
|
||
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'app'))
|
||
|
||
import requests
|
||
import json
|
||
from app import create_app
|
||
from app.models.user import User
|
||
from flask_jwt_extended import create_access_token
|
||
|
||
def test_admin_api():
|
||
"""測試管理後台 API 認證"""
|
||
|
||
app = create_app()
|
||
|
||
with app.app_context():
|
||
# 找到管理員用戶
|
||
admin_user = User.query.filter_by(is_admin=True).first()
|
||
|
||
if not admin_user:
|
||
print("❌ 找不到管理員用戶")
|
||
return
|
||
|
||
print(f"✅ 找到管理員用戶: {admin_user.username} (ID: {admin_user.id})")
|
||
|
||
# 創建JWT token
|
||
token = create_access_token(
|
||
identity=admin_user.username,
|
||
additional_claims={
|
||
'user_id': admin_user.id,
|
||
'is_admin': admin_user.is_admin
|
||
}
|
||
)
|
||
|
||
print(f"✅ 創建JWT token: {token[:50]}...")
|
||
|
||
# 測試API調用
|
||
base_url = "http://127.0.0.1:5000/api/v1"
|
||
headers = {
|
||
'Authorization': f'Bearer {token}',
|
||
'Content-Type': 'application/json'
|
||
}
|
||
|
||
# 測試各個管理後台API端點
|
||
test_endpoints = [
|
||
('GET', '/admin/stats', '系統統計'),
|
||
('GET', '/admin/jobs', '任務列表'),
|
||
('GET', '/admin/users', '用戶列表'),
|
||
('GET', '/admin/api-usage', 'API使用統計'),
|
||
('GET', '/admin/cache/stats', '快取統計'),
|
||
('GET', '/admin/health', '系統健康狀態'),
|
||
('GET', '/admin/metrics', '系統指標'),
|
||
]
|
||
|
||
for method, endpoint, name in test_endpoints:
|
||
print(f"\n🧪 測試 {name}: {method} {endpoint}")
|
||
|
||
try:
|
||
if method == 'GET':
|
||
response = requests.get(f"{base_url}{endpoint}", headers=headers, timeout=10)
|
||
else:
|
||
response = requests.request(method, f"{base_url}{endpoint}", headers=headers, timeout=10)
|
||
|
||
print(f"📊 狀態碼: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
try:
|
||
data = response.json()
|
||
if data.get('success'):
|
||
print(f"✅ {name} API 測試成功")
|
||
# 顯示部分回傳數據
|
||
if 'data' in data:
|
||
data_keys = list(data['data'].keys()) if isinstance(data['data'], dict) else 'Array'
|
||
print(f" 數據鍵值: {data_keys}")
|
||
else:
|
||
print(f"❌ {name} API 返回失敗: {data.get('message', 'Unknown error')}")
|
||
except json.JSONDecodeError:
|
||
print(f"❌ {name} API 返回非JSON格式數據")
|
||
|
||
elif response.status_code == 401:
|
||
print(f"❌ {name} API 認證失敗 (401 Unauthorized)")
|
||
print(f" 錯誤信息: {response.text}")
|
||
|
||
elif response.status_code == 403:
|
||
print(f"❌ {name} API 權限不足 (403 Forbidden)")
|
||
print(f" 錯誤信息: {response.text}")
|
||
|
||
else:
|
||
print(f"❌ {name} API 測試失敗 ({response.status_code})")
|
||
print(f" 錯誤信息: {response.text}")
|
||
|
||
except requests.exceptions.ConnectionError:
|
||
print(f"❌ 無法連接到伺服器，請確認Flask應用正在運行")
|
||
except requests.exceptions.Timeout:
|
||
print(f"❌ 請求超時")
|
||
except Exception as e:
|
||
print(f"❌ 測試發生錯誤: {e}")
|
||
|
||
print(f"\n=== 測試完成 ===")
|
||
|
||
if __name__ == "__main__":
|
||
test_admin_api() |