novalon-manage-system/test-suite/api/base_api.py

# API 集成测试 - 基础API客户端
import pytest
import requests
import time
import os
from typing import Optional, Dict, Any, Union
from requests.adapters import HTTPAdapter, Retry
from dotenv import load_dotenv
import httpx

# 加载环境变量
load_dotenv()


class BaseAPIClient:
    """基础 API 客户端，提供通用的 HTTP 请求方法"""
    
    def __init__(self, base_url: Optional[str] = None, timeout: int = 30):
        self.base_url = base_url or os.getenv('BASE_URL', 'http://localhost:8084')
        self.timeout = timeout
        self.session = requests.Session()
        self.token: Optional[str] = None
        self.user_id: Optional[int] = None
        
        # 配置重试策略
        retries = Retry(
            total=3,
            backoff_factor=0.1,
            status_forcelist=[500, 502, 503, 504]
        )
        self.session.mount('http', HTTPAdapter(max_retries=retries))
        self.session.mount('https', HTTPAdapter(max_retries=retries))
    
    def login(self, username: str, password: str) -> bool:
        """登录并获取 Token"""
        response = self.post(
            '/api/auth/login',
            json={'username': username, 'password': password},
            include_auth=False
        )
        
        if response.status_code == 200:
            data = response.json()
            self.token = data.get('token')
            self.user_id = data.get('userId')
            print(f"✅ 登录成功: {username} (User ID: {self.user_id})")
            return True
        else:
            print(f"❌ 登录失败: {response.status_code}")
            return False
    
    def _build_url(self, path: str) -> str:
        """构建完整 URL"""
        if path.startswith('http'):
            return path
        return f"{self.base_url}{path}"
    
    def _get_headers(self, include_auth: bool = True) -> Dict[str, str]:
        """获取请求头"""
        headers = {
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        }
        
        if include_auth and self.token:
            headers['Authorization'] = f"Bearer {self.token}"
        
        return headers
    
    def get(self, path: str, params: Optional[Dict] = None, include_auth: bool = True) -> requests.Response:
        """GET 请求"""
        url = self._build_url(path)
        headers = self._get_headers(include_auth)
        
        response = self.session.get(
            url,
            headers=headers,
            params=params,
            timeout=self.timeout
        )
        
        return response
    
    def post(self, path: str, data: Optional[Dict] = None, json: Optional[Dict] = None, 
             include_auth: bool = True) -> requests.Response:
        """POST 请求"""
        url = self._build_url(path)
        headers = self._get_headers(include_auth)
        
        response = self.session.post(
            url,
            headers=headers,
            data=data,
            json=json,
            timeout=self.timeout
        )
        
        return response
    
    def put(self, path: str, data: Optional[Dict] = None, json: Optional[Dict] = None,
            include_auth: bool = True) -> requests.Response:
        """PUT 请求"""
        url = self._build_url(path)
        headers = self._get_headers(include_auth)
        
        response = self.session.put(
            url,
            headers=headers,
            data=data,
            json=json,
            timeout=self.timeout
        )
        
        return response
    
    def delete(self, path: str, include_auth: bool = True) -> requests.Response:
        """DELETE 请求"""
        url = self._build_url(path)
        headers = self._get_headers(include_auth)
        
        response = self.session.delete(
            url,
            headers=headers,
            timeout=self.timeout
        )
        
        return response
    
    def cleanup_resource(self, resource_type: str, resource_id: int) -> bool:
        """清理测试资源"""
        try:
            response = self.delete(f'/api/{resource_type}/{resource_id}')
            return response.status_code == 200
        except Exception as e:
            print(f"清理资源失败: {e}")
            return False


class APIFixture:
    """API 测试固定装置，提供测试数据管理"""
    
    def __init__(self, api_client: BaseAPIClient):
        self.api_client = api_client
        self.created_users = []
        self.created_roles = []
        self.created_menus = []
        self.created_configs = []
        self.created_dicts = []
    
    def cleanup(self):
        """清理所有创建的测试数据"""
        print("\n🧹 清理测试数据...")
        
        # 清理用户
        for user_id in self.created_users:
            self.api_client.cleanup_resource('users', user_id)
        self.created_users.clear()
        
        # 清理角色
        for role_id in self.created_roles:
            self.api_client.cleanup_resource('roles', role_id)
        self.created_roles.clear()
        
        # 清理菜单
        for menu_id in self.created_menus:
            self.api_client.cleanup_resource('menus', menu_id)
        self.created_menus.clear()
        
        # 清理配置
        for config_id in self.created_configs:
            self.api_client.cleanup_resource('config', config_id)
        self.created_configs.clear()
        
        # 清理字典
        for dict_id in self.created_dicts:
            self.api_client.cleanup_resource('dict', dict_id)
        self.created_dicts.clear()
        
        print("✅ 测试数据清理完成")


class AsyncAPIClient:
    """异步 API 客户端，使用 httpx"""
    
    def __init__(self, client: httpx.AsyncClient):
        self.client = client
        self.token: Optional[str] = None
        self.user_id: Optional[int] = None
    
    def set_auth(self, token: str, user_id: int = None):
        """设置认证信息"""
        self.token = token
        self.user_id = user_id
        self.client.headers.update({'Authorization': f'Bearer {token}'})
    
    async def login(self, username: str, password: str) -> httpx.Response:
        """登录并获取 Token"""
        response = await self.client.post(
            '/api/auth/login',
            json={'username': username, 'password': password}
        )
        
        if response.status_code == 200:
            data = response.json()
            self.token = data.get('token')
            self.user_id = data.get('userId')
            print(f"✅ 登录成功: {username} (User ID: {self.user_id})")
        
        return response
    
    async def get(self, path: str, params: Optional[Dict] = None) -> httpx.Response:
        """GET 请求"""
        return await self.client.get(path, params=params)
    
    async def post(self, path: str, json: Optional[Dict] = None) -> httpx.Response:
        """POST 请求"""
        return await self.client.post(path, json=json)
    
    async def put(self, path: str, json: Optional[Dict] = None) -> httpx.Response:
        """PUT 请求"""
        return await self.client.put(path, json=json)
    
    async def delete(self, path: str) -> httpx.Response:
        """DELETE 请求"""
        return await self.client.delete(path)