207 lines
7.0 KiB
Python
207 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
测试FastAPI服务器功能的脚本
|
|
"""
|
|
|
|
import json
|
|
import time
|
|
import requests
|
|
import subprocess
|
|
import signal
|
|
import os
|
|
from pathlib import Path
|
|
|
|
def test_fastapi_server():
|
|
"""测试FastAPI服务器功能"""
|
|
|
|
print("=== FastAPI服务器功能测试 ===")
|
|
|
|
# 启动服务器
|
|
print("[信息] 启动FastAPI服务器...")
|
|
server_process = subprocess.Popen([
|
|
"python3", "fastapi_server.py",
|
|
"--host", "127.0.0.1",
|
|
"--port", "5051"
|
|
])
|
|
|
|
try:
|
|
# 等待服务器启动
|
|
print("[信息] 等待服务器启动...")
|
|
time.sleep(3)
|
|
|
|
base_url = "http://127.0.0.1:5051"
|
|
|
|
# 测试健康检查
|
|
print("\n1. 测试健康检查...")
|
|
try:
|
|
response = requests.get(f"{base_url}/", timeout=5)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
print(f" ✓ 健康检查成功: {data.get('status', 'unknown')}")
|
|
print(f" ✓ 服务版本: {data.get('version', 'unknown')}")
|
|
else:
|
|
print(f" ✗ 健康检查失败: {response.status_code}")
|
|
except Exception as e:
|
|
print(f" ✗ 健康检查异常: {e}")
|
|
|
|
# 测试服务信息
|
|
print("\n2. 测试服务信息...")
|
|
try:
|
|
response = requests.get(f"{base_url}/info", timeout=5)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
print(f" ✓ 服务信息获取成功")
|
|
print(f" ✓ 框架: {data.get('framework', 'unknown')}")
|
|
print(f" ✓ 功能数量: {len(data.get('features', []))}")
|
|
else:
|
|
print(f" ✗ 服务信息获取失败: {response.status_code}")
|
|
except Exception as e:
|
|
print(f" ✗ 服务信息获取异常: {e}")
|
|
|
|
# 测试API文档
|
|
print("\n3. 测试API文档...")
|
|
try:
|
|
# 测试Swagger UI
|
|
response = requests.get(f"{base_url}/docs", timeout=5)
|
|
if response.status_code == 200:
|
|
print(" ✓ Swagger UI 可访问")
|
|
else:
|
|
print(f" ✗ Swagger UI 访问失败: {response.status_code}")
|
|
|
|
# 测试ReDoc
|
|
response = requests.get(f"{base_url}/redoc", timeout=5)
|
|
if response.status_code == 200:
|
|
print(" ✓ ReDoc 可访问")
|
|
else:
|
|
print(f" ✗ ReDoc 访问失败: {response.status_code}")
|
|
|
|
# 测试OpenAPI规范
|
|
response = requests.get(f"{base_url}/openapi.json", timeout=5)
|
|
if response.status_code == 200:
|
|
openapi_spec = response.json()
|
|
print(" ✓ OpenAPI规范可获取")
|
|
print(f" ✓ API标题: {openapi_spec.get('info', {}).get('title', 'unknown')}")
|
|
print(f" ✓ 端点数量: {len(openapi_spec.get('paths', {}))}")
|
|
else:
|
|
print(f" ✗ OpenAPI规范获取失败: {response.status_code}")
|
|
except Exception as e:
|
|
print(f" ✗ API文档测试异常: {e}")
|
|
|
|
# 测试数据验证
|
|
print("\n4. 测试数据验证...")
|
|
try:
|
|
# 测试无效请求
|
|
invalid_config = {
|
|
"base_url": "invalid-url", # 无效URL
|
|
"page_size": 0 # 无效分页大小
|
|
}
|
|
response = requests.post(f"{base_url}/run", json=invalid_config, timeout=5)
|
|
if response.status_code == 422: # Validation Error
|
|
print(" ✓ 数据验证正常工作")
|
|
error_detail = response.json()
|
|
print(f" ✓ 验证错误数量: {len(error_detail.get('detail', []))}")
|
|
else:
|
|
print(f" ✗ 数据验证异常: {response.status_code}")
|
|
except Exception as e:
|
|
print(f" ✗ 数据验证测试异常: {e}")
|
|
|
|
# 测试报告列表
|
|
print("\n5. 测试报告列表...")
|
|
try:
|
|
response = requests.get(f"{base_url}/reports", timeout=5)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
print(" ✓ 报告列表获取成功")
|
|
print(f" ✓ 报告数量: {len(data.get('reports', []))}")
|
|
else:
|
|
print(f" ✗ 报告列表获取失败: {response.status_code}")
|
|
except Exception as e:
|
|
print(f" ✗ 报告列表测试异常: {e}")
|
|
|
|
print("\n=== 测试完成 ===")
|
|
print(f"FastAPI服务器运行在: {base_url}")
|
|
print(f"API文档地址: {base_url}/docs")
|
|
print(f"ReDoc地址: {base_url}/redoc")
|
|
|
|
finally:
|
|
# 停止服务器
|
|
print("\n[信息] 停止服务器...")
|
|
server_process.terminate()
|
|
try:
|
|
server_process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
server_process.kill()
|
|
server_process.wait()
|
|
print("[信息] 服务器已停止")
|
|
|
|
def test_pagination_parameters():
|
|
"""测试分页参数功能"""
|
|
|
|
print("\n=== 分页参数测试 ===")
|
|
|
|
# 测试配置
|
|
test_configs = [
|
|
{
|
|
"name": "基本配置",
|
|
"config": {
|
|
"dms": "./test.json",
|
|
"base_url": "https://api.example.com",
|
|
"page_size": 100,
|
|
"page_no": 1
|
|
}
|
|
},
|
|
{
|
|
"name": "大分页配置",
|
|
"config": {
|
|
"dms": "./test.json",
|
|
"base_url": "https://api.example.com",
|
|
"page_size": 5000,
|
|
"page_no": 1
|
|
}
|
|
},
|
|
{
|
|
"name": "跳页配置",
|
|
"config": {
|
|
"dms": "./test.json",
|
|
"base_url": "https://api.example.com",
|
|
"page_size": 500,
|
|
"page_no": 5
|
|
}
|
|
}
|
|
]
|
|
|
|
for test_case in test_configs:
|
|
print(f"\n测试: {test_case['name']}")
|
|
config = test_case['config']
|
|
|
|
# 验证配置格式
|
|
try:
|
|
from fastapi_server import TestConfig
|
|
validated_config = TestConfig(**config)
|
|
print(f" ✓ 配置验证通过")
|
|
print(f" ✓ 页面大小: {validated_config.page_size}")
|
|
print(f" ✓ 起始页码: {validated_config.page_no}")
|
|
except Exception as e:
|
|
print(f" ✗ 配置验证失败: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
print("开始FastAPI功能测试")
|
|
|
|
# 检查依赖
|
|
try:
|
|
import fastapi
|
|
import uvicorn
|
|
import pydantic
|
|
print(f"FastAPI版本: {fastapi.__version__}")
|
|
print(f"Pydantic版本: {pydantic.__version__}")
|
|
except ImportError as e:
|
|
print(f"依赖缺失: {e}")
|
|
print("请运行: pip install -r requirements_fastapi.txt")
|
|
exit(1)
|
|
|
|
# 运行测试
|
|
test_fastapi_server()
|
|
test_pagination_parameters()
|
|
|
|
print("\n测试完成！")
|