""" 测试配置和Fixtures """ import pytest from httpx import AsyncClient, ASGITransport from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from sqlalchemy.pool import StaticPool from datetime import datetime from typing import AsyncGenerator from app.main import app from app.db.base import Base from app.models.user import User, Role, UserRole, Permission from app.models.device_type import DeviceType, DeviceTypeField from app.core.security import get_password_hash, security_manager # 创建测试数据库引擎 test_engine = create_async_engine( "sqlite+aiosqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) # 创建测试会话工厂 TestSessionLocal = async_sessionmaker( test_engine, class_=AsyncSession, expire_on_commit=False, autocommit=False, autoflush=False, ) @pytest.fixture(scope="function") async def db_session(): """创建测试数据库会话""" async with test_engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) async with TestSessionLocal() as session: yield session await session.rollback() async with test_engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) @pytest.fixture(scope="function") async def client(db_session): """创建测试客户端""" from app.core.deps import get_db async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test" ) as ac: yield ac app.dependency_overrides.clear() # ===== 用户相关Fixtures ===== @pytest.fixture async def test_password() -> str: """测试密码""" return "Test123456" @pytest.fixture async def test_user(db_session: AsyncSession, test_password: str) -> User: """创建测试用户""" user = User( username="testuser", password_hash=get_password_hash(test_password), real_name="测试用户", email="test@example.com", phone="13800138000", status="active", is_admin=False ) db_session.add(user) await db_session.flush() await db_session.refresh(user) return user @pytest.fixture async def test_admin(db_session: AsyncSession, test_password: str) -> User: """创建测试管理员""" admin = User( username="admin", password_hash=get_password_hash(test_password), real_name="系统管理员", email="admin@example.com", status="active", is_admin=True ) db_session.add(admin) await db_session.flush() await db_session.refresh(admin) return admin @pytest.fixture async def test_role(db_session: AsyncSession) -> Role: """创建测试角色""" role = Role( role_name="测试角色", role_code="TEST_ROLE", description="用于测试的角色", status="active", sort_order=1 ) db_session.add(role) await db_session.flush() await db_session.refresh(role) return role @pytest.fixture async def auth_headers(client: AsyncClient, test_user: User, test_password: str) -> dict: """获取认证头""" # 登录获取token response = await client.post( "/api/v1/auth/login", json={ "username": test_user.username, "password": test_password, "captcha": "1234", "captcha_key": "test-uuid" } ) if response.status_code == 200: token = response.json()["data"]["access_token"] return {"Authorization": f"Bearer {token}"} return {} @pytest.fixture async def admin_headers(client: AsyncClient, test_admin: User, test_password: str) -> dict: """获取管理员认证头""" response = await client.post( "/api/v1/auth/login", json={ "username": test_admin.username, "password": test_password, "captcha": "1234", "captcha_key": "test-uuid" } ) if response.status_code == 200: token = response.json()["data"]["access_token"] return {"Authorization": f"Bearer {token}"} return {} # ===== 设备类型相关Fixtures ===== @pytest.fixture async def test_device_type(db_session: AsyncSession, test_admin: User) -> DeviceType: """创建测试设备类型""" device_type = DeviceType( type_code="COMPUTER", type_name="计算机", category="IT设备", description="台式机、笔记本等", icon="computer", status="active", sort_order=1, created_by=test_admin.id ) db_session.add(device_type) await db_session.flush() await db_session.refresh(device_type) return device_type @pytest.fixture async def test_device_type_with_fields( db_session: AsyncSession, test_device_type: DeviceType, test_admin: User ) -> DeviceType: """创建带字段的测试设备类型""" fields = [ DeviceTypeField( device_type_id=test_device_type.id, field_code="cpu", field_name="CPU型号", field_type="text", is_required=True, placeholder="例如: Intel i5-10400", validation_rules={"max_length": 100}, sort_order=1, created_by=test_admin.id ), DeviceTypeField( device_type_id=test_device_type.id, field_code="memory", field_name="内存容量", field_type="select", is_required=True, options=[ {"label": "8GB", "value": "8"}, {"label": "16GB", "value": "16"}, {"label": "32GB", "value": "32"} ], sort_order=2, created_by=test_admin.id ), DeviceTypeField( device_type_id=test_device_type.id, field_code="disk", field_name="硬盘容量", field_type="text", is_required=False, placeholder="例如: 512GB SSD", sort_order=3, created_by=test_admin.id ) ] for field in fields: db_session.add(field) await db_session.flush() return test_device_type # ===== 辅助函数Fixtures ===== @pytest.fixture def sample_asset_data(test_device_type: DeviceType) -> dict: """示例资产数据""" return { "asset_name": "测试资产", "device_type_id": test_device_type.id, "organization_id": 1, "model": "测试型号", "serial_number": f"SN{datetime.now().strftime('%Y%m%d%H%M%S')}", "purchase_date": "2024-01-15", "purchase_price": 5000.00, "warranty_period": 24, "location": "测试位置", "dynamic_attributes": { "cpu": "Intel i5-10400", "memory": "16", "disk": "512GB SSD" } } @pytest.fixture def sample_device_type_data() -> dict: """示例设备类型数据""" return { "type_code": "LAPTOP", "type_name": "笔记本电脑", "category": "IT设备", "description": "笔记本电脑类", "icon": "laptop", "sort_order": 1 } @pytest.fixture def sample_field_data() -> dict: """示例字段数据""" return { "field_code": "gpu", "field_name": "显卡型号", "field_type": "text", "is_required": False, "placeholder": "例如: GTX 1660Ti", "validation_rules": {"max_length": 100}, "sort_order": 4 }