novalon-manage-system/api_integration_tests/conftest.py

"""
Pytest配置和fixtures
"""

import asyncio
import pytest
from typing import AsyncGenerator, Generator
from playwright.async_api import async_playwright, Browser, BrowserContext, Page
from httpx import AsyncClient

from config.settings import settings
from utils.test_data_manager import TestDataManager


@pytest.fixture(scope="session")
def event_loop():
    """创建事件循环"""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    yield loop
    loop.close()
    asyncio.set_event_loop(None)


@pytest.fixture(scope="session")
async def browser() -> AsyncGenerator[Browser, None]:
    """浏览器fixture"""
    async with async_playwright() as p:
        browser = await p.launch(
            headless=settings.HEADLESS_BROWSER,
            browser_type=settings.BROWSER_TYPE
        )
        yield browser
        await browser.close()


@pytest.fixture
async def context(browser: Browser) -> AsyncGenerator[BrowserContext, None]:
    """浏览器上下文fixture"""
    context = await browser.new_context()
    yield context
    await context.close()


@pytest.fixture
async def page(context: BrowserContext) -> AsyncGenerator[Page, None]:
    """页面fixture"""
    page = await context.new_page()
    page.set_default_timeout(settings.REQUEST_TIMEOUT)
    yield page
    await page.close()


@pytest.fixture
async def http_client() -> AsyncGenerator[AsyncClient, None]:
    """HTTP客户端fixture"""
    async with AsyncClient(
        base_url=settings.API_BASE_URL,
        timeout=settings.REQUEST_TIMEOUT / 1000
    ) as client:
        yield client


@pytest.fixture
async def auth_token(http_client: AsyncClient) -> str:
    """获取认证token"""
    from config.settings import settings
    print(f"测试登录配置: username={settings.TEST_USERNAME}, password={settings.TEST_PASSWORD}")
    response = await http_client.post(
        "/api/auth/login",
        json={
            "username": settings.TEST_USERNAME,
            "password": settings.TEST_PASSWORD
        }
    )
    print(f"登录响应状态: {response.status_code}")
    if response.status_code != 200:
        print(f"登录响应内容: {response.text}")
    assert response.status_code == 200
    data = response.json()
    return data.get("token")


@pytest.fixture
async def authenticated_client(http_client: AsyncClient, auth_token: str) -> AsyncClient:
    """已认证的HTTP客户端"""
    http_client.headers.update({"Authorization": f"Bearer {auth_token}"})
    return http_client


@pytest.fixture
def test_user_data():
    """测试用户数据"""
    import time
    timestamp = int(time.time() * 1000)
    return {
        "username": f"testuser_{timestamp}",
        "password": "Password123!",
        "email": f"test_{timestamp}@example.com",
        "roleId": 2,
        "status": 1
    }


@pytest.fixture
def test_role_data():
    """测试角色数据"""
    import time
    timestamp = int(time.time() * 1000)
    return {
        "roleName": f"TEST_ROLE_{timestamp}",
        "roleKey": f"test_role_{timestamp}",
        "roleSort": 1,
        "status": 1
    }


@pytest.fixture
def test_dictionary_data():
    """测试字典数据"""
    return {
        "type": "USER_STATUS",
        "code": "ACTIVE",
        "name": "激活",
        "value": "1",
        "remark": "用户激活状态",
        "sort": 1
    }


@pytest.fixture
async def cleanup_user(authenticated_client: AsyncClient):
    """清理测试用户"""
    user_ids = []
    
    yield user_ids
    
    for user_id in user_ids:
        try:
            await authenticated_client.delete(f"/api/users/{user_id}")
        except Exception:
            pass


@pytest.fixture
async def cleanup_role(authenticated_client: AsyncClient):
    """清理测试角色"""
    role_ids = []
    
    yield role_ids
    
    for role_id in role_ids:
        try:
            await authenticated_client.delete(f"/api/roles/{role_id}")
        except Exception:
            pass


@pytest.fixture
async def cleanup_dictionary(authenticated_client: AsyncClient):
    """清理测试字典"""
    dict_ids = []
    
    yield dict_ids
    
    for dict_id in dict_ids:
        try:
            await authenticated_client.delete(f"/api/dictionaries/{dict_id}")
        except Exception:
            pass


@pytest.fixture
async def cleanup_dict_type(authenticated_client: AsyncClient):
    """清理字典类型"""
    dict_ids = []
    
    yield dict_ids
    
    for dict_id in dict_ids:
        try:
            await authenticated_client.delete(f"/api/dict/types/{dict_id}")
        except Exception:
            pass


@pytest.fixture
async def cleanup_config(authenticated_client: AsyncClient):
    """清理系统配置"""
    config_ids = []
    
    yield config_ids
    
    for config_id in config_ids:
        try:
            await authenticated_client.delete(f"/api/config/{config_id}")
        except Exception:
            pass


@pytest.fixture
async def cleanup_notice(authenticated_client: AsyncClient):
    """清理系统公告"""
    notice_ids = []
    
    yield notice_ids
    
    for notice_id in notice_ids:
        try:
            await authenticated_client.delete(f"/api/notices/{notice_id}")
        except Exception:
            pass


@pytest.fixture
async def cleanup_file(authenticated_client: AsyncClient):
    """清理文件"""
    file_ids = []
    
    yield file_ids
    
    for file_id in file_ids:
        try:
            await authenticated_client.delete(f"/api/files/{file_id}")
        except Exception:
            pass


@pytest.fixture
async def test_data_manager(authenticated_client: AsyncClient) -> AsyncGenerator[TestDataManager, None]:
    """测试数据管理器fixture"""
    manager = TestDataManager(authenticated_client)
    yield manager
    await manager.cleanup_all()