test: 添加 API 自动化测试脚本
- 测试前端页面可访问性（4 个页面） - 测试用户认证系统（登录/获取用户/错误处理/token 刷新） - 测试区域系统（列表/省级/详情/子区域） - 测试文章系统（列表/区域筛选） - 测试特色服务（列表/区域筛选） - 测试评论和评分系统 - 测试 Admin 后台 - 共 19 个测试用例，彩色输出
This commit is contained in:
maoshen
349
city-manual/test_api.py
Normal file
349
city-manual/test_api.py
Normal file
@@ -0,0 +1,349 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
城市手册 - API 自动化测试脚本
|
||||
测试所有已实现的功能
|
||||
"""
|
||||
|
||||
import requests
|
||||
import sys
|
||||
import json
|
||||
|
||||
BASE_URL = 'http://127.0.0.1:81/api'
|
||||
FRONTEND_URL = 'http://127.0.0.1:81'
|
||||
|
||||
# 颜色输出
|
||||
GREEN = '\033[92m'
|
||||
RED = '\033[91m'
|
||||
YELLOW = '\033[93m'
|
||||
RESET = '\033[0m'
|
||||
|
||||
def log_success(msg):
|
||||
print(f"{GREEN}✓{RESET} {msg}")
|
||||
|
||||
def log_error(msg):
|
||||
print(f"{RED}✗{RESET} {msg}")
|
||||
|
||||
def log_info(msg):
|
||||
print(f"{YELLOW}→{RESET} {msg}")
|
||||
|
||||
def test_frontend_pages():
|
||||
"""测试前端页面可访问性"""
|
||||
log_info("\n📄 测试前端页面...")
|
||||
|
||||
pages = [
|
||||
('/', '首页'),
|
||||
('/login/', '登录页'),
|
||||
('/register/', '注册页'),
|
||||
('/cities/', '城市列表页'),
|
||||
]
|
||||
|
||||
passed = 0
|
||||
for path, name in pages:
|
||||
try:
|
||||
resp = requests.get(f"{FRONTEND_URL}{path}", timeout=5)
|
||||
if resp.status_code == 200:
|
||||
log_success(f"{name} ({path}) - {resp.status_code}")
|
||||
passed += 1
|
||||
else:
|
||||
log_error(f"{name} ({path}) - {resp.status_code}")
|
||||
except Exception as e:
|
||||
log_error(f"{name} ({path}) - {str(e)}")
|
||||
|
||||
return passed, len(pages)
|
||||
|
||||
def test_user_auth():
|
||||
"""测试用户认证系统"""
|
||||
log_info("\n🔐 测试用户系统...")
|
||||
|
||||
passed = 0
|
||||
total = 4
|
||||
|
||||
# 1. 测试登录
|
||||
try:
|
||||
resp = requests.post(f"{BASE_URL}/token/", json={
|
||||
'username': 'demo',
|
||||
'password': 'demo123'
|
||||
}, timeout=5)
|
||||
|
||||
if resp.status_code == 200 and 'access' in resp.json():
|
||||
token = resp.json()['access']
|
||||
log_success(f"用户登录 - {resp.status_code}")
|
||||
passed += 1
|
||||
else:
|
||||
log_error(f"用户登录 - {resp.status_code}")
|
||||
except Exception as e:
|
||||
log_error(f"用户登录 - {str(e)}")
|
||||
token = None
|
||||
|
||||
# 2. 测试获取当前用户信息
|
||||
if token:
|
||||
try:
|
||||
resp = requests.get(f"{BASE_URL}/users/me/",
|
||||
headers={'Authorization': f'Bearer {token}'},
|
||||
timeout=5)
|
||||
if resp.status_code == 200:
|
||||
user = resp.json()
|
||||
log_success(f"获取用户信息 - {resp.status_code} (用户名：{user.get('username', 'N/A')})")
|
||||
passed += 1
|
||||
else:
|
||||
log_error(f"获取用户信息 - {resp.status_code}")
|
||||
except Exception as e:
|
||||
log_error(f"获取用户信息 - {str(e)}")
|
||||
|
||||
# 3. 测试错误密码登录
|
||||
try:
|
||||
resp = requests.post(f"{BASE_URL}/token/", json={
|
||||
'username': 'demo',
|
||||
'password': 'wrongpassword'
|
||||
}, timeout=5)
|
||||
|
||||
if resp.status_code == 401:
|
||||
log_success(f"错误密码拒绝 - {resp.status_code}")
|
||||
passed += 1
|
||||
else:
|
||||
log_error(f"错误密码拒绝 - {resp.status_code} (应该返回 401)")
|
||||
except Exception as e:
|
||||
log_error(f"错误密码拒绝 - {str(e)}")
|
||||
|
||||
# 4. 测试 token 刷新端点存在
|
||||
try:
|
||||
# 只需要测试端点存在，不实际刷新（因为需要有效的 refresh token）
|
||||
resp = requests.options(f"{BASE_URL}/token/refresh/", timeout=5)
|
||||
# OPTIONS 请求应该返回 200 或 401（需要认证）
|
||||
if resp.status_code in [200, 401, 403]:
|
||||
log_success(f"Token 刷新端点 - 可用")
|
||||
passed += 1
|
||||
else:
|
||||
log_error(f"Token 刷新端点 - {resp.status_code}")
|
||||
except Exception as e:
|
||||
log_error(f"Token 刷新端点 - {str(e)}")
|
||||
|
||||
return passed, total
|
||||
|
||||
def test_regions():
|
||||
"""测试区域系统"""
|
||||
log_info("\n🗺️ 测试区域系统...")
|
||||
|
||||
passed = 0
|
||||
total = 4
|
||||
|
||||
# 1. 获取区域列表
|
||||
try:
|
||||
resp = requests.get(f"{BASE_URL}/regions/", timeout=5)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
count = data.get('count', 0)
|
||||
log_success(f"区域列表 - {resp.status_code} (共{count}个区域)")
|
||||
passed += 1
|
||||
else:
|
||||
log_error(f"区域列表 - {resp.status_code}")
|
||||
except Exception as e:
|
||||
log_error(f"区域列表 - {str(e)}")
|
||||
|
||||
# 2. 获取省级区域
|
||||
try:
|
||||
resp = requests.get(f"{BASE_URL}/regions/provinces/", timeout=5)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
# API 可能返回 list 或 paginated response
|
||||
if isinstance(data, list):
|
||||
count = len(data)
|
||||
else:
|
||||
count = data.get('count', len(data))
|
||||
log_success(f"省级区域 - {resp.status_code} (共{count}个省)")
|
||||
passed += 1
|
||||
else:
|
||||
log_error(f"省级区域 - {resp.status_code}")
|
||||
except Exception as e:
|
||||
log_error(f"省级区域 - {str(e)}")
|
||||
|
||||
# 3. 获取区域详情（北京）
|
||||
try:
|
||||
resp = requests.get(f"{BASE_URL}/regions/1/", timeout=5)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
log_success(f"区域详情 - {resp.status_code} ({data.get('name', 'N/A')})")
|
||||
passed += 1
|
||||
else:
|
||||
log_error(f"区域详情 - {resp.status_code}")
|
||||
except Exception as e:
|
||||
log_error(f"区域详情 - {str(e)}")
|
||||
|
||||
# 4. 获取子区域
|
||||
try:
|
||||
resp = requests.get(f"{BASE_URL}/regions/3/children/", timeout=5)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
# API 可能返回 list 或 paginated response
|
||||
if isinstance(data, list):
|
||||
count = len(data)
|
||||
else:
|
||||
count = data.get('count', len(data))
|
||||
log_success(f"子区域 - {resp.status_code} (广东省有{count}个子区域)")
|
||||
passed += 1
|
||||
else:
|
||||
log_error(f"子区域 - {resp.status_code}")
|
||||
except Exception as e:
|
||||
log_error(f"子区域 - {str(e)}")
|
||||
|
||||
return passed, total
|
||||
|
||||
def test_articles():
|
||||
"""测试文章系统"""
|
||||
log_info("\n📝 测试文章系统...")
|
||||
|
||||
passed = 0
|
||||
total = 2
|
||||
|
||||
# 1. 获取文章列表
|
||||
try:
|
||||
resp = requests.get(f"{BASE_URL}/articles/", timeout=5)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
count = data.get('count', 0)
|
||||
log_success(f"文章列表 - {resp.status_code} (共{count}篇文章)")
|
||||
passed += 1
|
||||
else:
|
||||
log_error(f"文章列表 - {resp.status_code}")
|
||||
except Exception as e:
|
||||
log_error(f"文章列表 - {str(e)}")
|
||||
|
||||
# 2. 按区域筛选文章
|
||||
try:
|
||||
resp = requests.get(f"{BASE_URL}/articles/?region=1", timeout=5)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
count = data.get('count', 0)
|
||||
log_success(f"按区域筛选文章 - {resp.status_code} (北京有{count}篇文章)")
|
||||
passed += 1
|
||||
else:
|
||||
log_error(f"按区域筛选文章 - {resp.status_code}")
|
||||
except Exception as e:
|
||||
log_error(f"按区域筛选文章 - {str(e)}")
|
||||
|
||||
return passed, total
|
||||
|
||||
def test_services():
|
||||
"""测试特色服务系统"""
|
||||
log_info("\n🏪 测试特色服务...")
|
||||
|
||||
passed = 0
|
||||
total = 2
|
||||
|
||||
# 1. 获取服务列表
|
||||
try:
|
||||
resp = requests.get(f"{BASE_URL}/services/", timeout=5)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
count = data.get('count', 0)
|
||||
log_success(f"服务列表 - {resp.status_code} (共{count}个服务)")
|
||||
passed += 1
|
||||
else:
|
||||
log_error(f"服务列表 - {resp.status_code}")
|
||||
except Exception as e:
|
||||
log_error(f"服务列表 - {str(e)}")
|
||||
|
||||
# 2. 按区域筛选服务
|
||||
try:
|
||||
resp = requests.get(f"{BASE_URL}/services/?region=1", timeout=5)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
count = data.get('count', 0)
|
||||
log_success(f"按区域筛选服务 - {resp.status_code} (北京有{count}个服务)")
|
||||
passed += 1
|
||||
else:
|
||||
log_error(f"按区域筛选服务 - {resp.status_code}")
|
||||
except Exception as e:
|
||||
log_error(f"按区域筛选服务 - {str(e)}")
|
||||
|
||||
return passed, total
|
||||
|
||||
def test_comments_ratings():
|
||||
"""测试评论和评分系统"""
|
||||
log_info("\n💬 测试评论和评分...")
|
||||
|
||||
passed = 0
|
||||
total = 2
|
||||
|
||||
# 1. 获取评论列表
|
||||
try:
|
||||
resp = requests.get(f"{BASE_URL}/comments/", timeout=5)
|
||||
if resp.status_code == 200:
|
||||
log_success(f"评论列表 - {resp.status_code}")
|
||||
passed += 1
|
||||
else:
|
||||
log_error(f"评论列表 - {resp.status_code}")
|
||||
except Exception as e:
|
||||
log_error(f"评论列表 - {str(e)}")
|
||||
|
||||
# 2. 获取评分列表
|
||||
try:
|
||||
resp = requests.get(f"{BASE_URL}/ratings/", timeout=5)
|
||||
if resp.status_code == 200:
|
||||
log_success(f"评分列表 - {resp.status_code}")
|
||||
passed += 1
|
||||
else:
|
||||
log_error(f"评分列表 - {resp.status_code}")
|
||||
except Exception as e:
|
||||
log_error(f"评分列表 - {str(e)}")
|
||||
|
||||
return passed, total
|
||||
|
||||
def test_admin():
|
||||
"""测试 Admin 后台"""
|
||||
log_info("\n⚙️ 测试 Admin 后台...")
|
||||
|
||||
try:
|
||||
resp = requests.get(f"{FRONTEND_URL}/admin/", timeout=5, allow_redirects=False)
|
||||
if resp.status_code in [200, 302]:
|
||||
log_success(f"Admin 后台 - {resp.status_code}")
|
||||
return 1, 1
|
||||
else:
|
||||
log_error(f"Admin 后台 - {resp.status_code}")
|
||||
return 0, 1
|
||||
except Exception as e:
|
||||
log_error(f"Admin 后台 - {str(e)}")
|
||||
return 0, 1
|
||||
|
||||
def main():
|
||||
print("\n" + "="*60)
|
||||
print("📖 城市手册 - API 自动化测试")
|
||||
print("="*60)
|
||||
|
||||
total_passed = 0
|
||||
total_tests = 0
|
||||
|
||||
# 运行所有测试
|
||||
tests = [
|
||||
test_frontend_pages,
|
||||
test_user_auth,
|
||||
test_regions,
|
||||
test_articles,
|
||||
test_services,
|
||||
test_comments_ratings,
|
||||
test_admin,
|
||||
]
|
||||
|
||||
for test_func in tests:
|
||||
try:
|
||||
passed, total = test_func()
|
||||
total_passed += passed
|
||||
total_tests += total
|
||||
except Exception as e:
|
||||
log_error(f"{test_func.__name__} 异常：{str(e)}")
|
||||
|
||||
# 打印总结
|
||||
print("\n" + "="*60)
|
||||
print(f"📊 测试结果：{total_passed}/{total_tests} 通过")
|
||||
|
||||
if total_passed == total_tests:
|
||||
print(f"{GREEN}✅ 所有测试通过！{RESET}")
|
||||
else:
|
||||
failed = total_tests - total_passed
|
||||
print(f"{RED}⚠️ {failed} 个测试失败{RESET}")
|
||||
print("="*60 + "\n")
|
||||
|
||||
return 0 if total_passed == total_tests else 1
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user