123 lines
3.3 KiB
Python
123 lines
3.3 KiB
Python
|
# tests/conftest.py - pytest fixtures for 甲辰藏品系统
|
|||
|
|
|
|||
|
|
import os
|
|||
|
|
import sys
|
|||
|
|
from datetime import datetime
|
|||
|
|
|
|||
|
|
# 设置项目路径
|
|||
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|||
|
|
|
|||
|
|
# 设置环境变量（测试环境使用SQLite内存数据库）
|
|||
|
|
os.environ["SECRET_KEY"] = "test-secret-key-for-testing-only"
|
|||
|
|
os.environ["DATABASE_URL"] = "sqlite:///./test.db"
|
|||
|
|
os.environ["ALGORITHM"] = "HS256"
|
|||
|
|
os.environ["ACCESS_TOKEN_EXPIRE_MINUTES"] = "60"
|
|||
|
|
|
|||
|
|
import pytest
|
|||
|
|
from fastapi.testclient import TestClient
|
|||
|
|
from sqlalchemy import create_engine, event
|
|||
|
|
from sqlalchemy.orm import sessionmaker
|
|||
|
|
from sqlalchemy.pool import StaticPool
|
|||
|
|
|
|||
|
|
from app.core.database import Base, get_db
|
|||
|
|
from app.main import app
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ========== 数据库 Fixture ==========
|
|||
|
|
|
|||
|
|
# 使用SQLite内存数据库进行测试
|
|||
|
|
TEST_DATABASE_URL = "sqlite:///:memory:"
|
|||
|
|
|
|||
|
|
engine = create_engine(
|
|||
|
|
TEST_DATABASE_URL,
|
|||
|
|
connect_args={"check_same_thread": False},
|
|||
|
|
poolclass=StaticPool,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# 启用外键约束（SQLite需要）
|
|||
|
|
@event.listens_for(engine, "connect")
|
|||
|
|
def set_sqlite_pragma(dbapi_connection, connection_record):
|
|||
|
|
cursor = dbapi_connection.cursor()
|
|||
|
|
cursor.execute("PRAGMA foreign_keys=ON")
|
|||
|
|
cursor.close()
|
|||
|
|
|
|||
|
|
|
|||
|
|
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture(scope="function")
|
|||
|
|
def db_session():
|
|||
|
|
"""每次测试创建新的数据库表，测试结束后清理"""
|
|||
|
|
Base.metadata.create_all(bind=engine)
|
|||
|
|
session = TestingSessionLocal()
|
|||
|
|
try:
|
|||
|
|
yield session
|
|||
|
|
finally:
|
|||
|
|
session.close()
|
|||
|
|
Base.metadata.drop_all(bind=engine)
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture(scope="function")
|
|||
|
|
def client(db_session):
|
|||
|
|
"""FastAPI测试客户端，使用测试数据库"""
|
|||
|
|
def override_get_db():
|
|||
|
|
try:
|
|||
|
|
yield db_session
|
|||
|
|
finally:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
app.dependency_overrides[get_db] = override_get_db
|
|||
|
|
with TestClient(app) as test_client:
|
|||
|
|
yield test_client
|
|||
|
|
app.dependency_overrides.clear()
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture(scope="function")
|
|||
|
|
def sample_user_data():
|
|||
|
|
"""示例用户注册数据"""
|
|||
|
|
return {
|
|||
|
|
"f01_01_name": "testuser",
|
|||
|
|
"password": "testpass123",
|
|||
|
|
"email": "test@example.com",
|
|||
|
|
"phone": "13800138000",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture(scope="function")
|
|||
|
|
def registered_user(db_session, sample_user_data):
|
|||
|
|
"""创建一个已注册的用户（带密码哈希）"""
|
|||
|
|
from app.core.auth import get_password_hash
|
|||
|
|
from app.models.models import User
|
|||
|
|
import uuid
|
|||
|
|
|
|||
|
|
hashed_password = get_password_hash(sample_user_data["password"])
|
|||
|
|
user = User(
|
|||
|
|
f99_90_id=str(uuid.uuid4()),
|
|||
|
|
f99_91_user_id=str(uuid.uuid4()),
|
|||
|
|
user_code="201",
|
|||
|
|
f01_01_name=sample_user_data["f01_01_name"],
|
|||
|
|
email=sample_user_data["email"],
|
|||
|
|
phone=sample_user_data["phone"],
|
|||
|
|
password=hashed_password,
|
|||
|
|
role="user",
|
|||
|
|
)
|
|||
|
|
db_session.add(user)
|
|||
|
|
db_session.commit()
|
|||
|
|
db_session.refresh(user)
|
|||
|
|
return user
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture(scope="function")
|
|||
|
|
def auth_token(registered_user):
|
|||
|
|
"""生成已注册用户的访问令牌"""
|
|||
|
|
from app.core.auth import create_access_token
|
|||
|
|
|
|||
|
|
token = create_access_token(data={"sub": registered_user.f99_90_id})
|
|||
|
|
return token
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture(scope="function")
|
|||
|
|
def auth_headers(auth_token):
|
|||
|
|
"""带Bearer令牌的请求头"""
|
|||
|
|
return {"Authorization": f"Bearer {auth_token}"}
|