110 lines
2.9 KiB
Python
110 lines
2.9 KiB
Python
import pytest
|
|
import os
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from app.main import app as fastapi_app
|
|
from app.database.database import Base, get_db
|
|
from app.config import settings
|
|
from app.models.user import User
|
|
|
|
# 设置测试环境变量
|
|
os.environ["TESTING"] = "True"
|
|
|
|
# 创建测试数据库引擎
|
|
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
|
|
engine = create_engine(
|
|
SQLALCHEMY_DATABASE_URL,
|
|
connect_args={"check_same_thread": False}
|
|
)
|
|
|
|
# 创建测试会话工厂
|
|
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
|
|
from app.common.utils import get_password_hash
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def test_db():
|
|
"""创建测试数据库"""
|
|
# 创建表
|
|
Base.metadata.create_all(bind=engine)
|
|
# 创建会话
|
|
db = TestingSessionLocal()
|
|
try:
|
|
# 添加测试用户
|
|
test_users = [
|
|
User(
|
|
username="test_admin",
|
|
password="$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW", # test_password
|
|
department="IT",
|
|
role="admin"
|
|
),
|
|
User(
|
|
username="test_marketing",
|
|
password="$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW", # test_password
|
|
department="Marketing",
|
|
role="marketing"
|
|
),
|
|
User(
|
|
username="test_other",
|
|
password="$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW", # test_password
|
|
department="Other",
|
|
role="other"
|
|
)
|
|
]
|
|
db.add_all(test_users)
|
|
db.commit()
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
# 删除表
|
|
Base.metadata.drop_all(bind=engine)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client(test_db):
|
|
"""创建测试客户端"""
|
|
def override_get_db():
|
|
try:
|
|
yield test_db
|
|
finally:
|
|
pass
|
|
|
|
# 覆盖依赖
|
|
fastapi_app.dependency_overrides[get_db] = override_get_db
|
|
|
|
with TestClient(fastapi_app) as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def admin_token(client):
|
|
"""获取管理员token"""
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
data={"username": "test_admin", "password": "test_password"}
|
|
)
|
|
return response.json()["access_token"]
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def marketing_token(client):
|
|
"""获取市场部token"""
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
data={"username": "test_marketing", "password": "test_password"}
|
|
)
|
|
return response.json()["access_token"]
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def other_token(client):
|
|
"""获取其他部门token"""
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
data={"username": "test_other", "password": "test_password"}
|
|
)
|
|
return response.json()["access_token"]
|