chengshishouce/city-manual/test_api.py

#!/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())