对于我的测试,我想使用测试数据库,为此我使用依赖项覆盖。但是,我的端点也在访问生产数据库。
我的结构代码:
src \
. . .|auth \
. . .| . . . |router.py\
. . .|main.py\
tests\
. . .|conftest.py\
. . .|test_auth.py
main.py
app = FastAPI(
title="Grid Bot Backend"
)
routers = APIRouter(prefix="/api/v1")
routers.include_router(auth_router)
auth.router.py
@router.post(
"/register",
summary="Create User",
response_model=UserOut,
dependencies=[Depends(check_user)],
)
async def register_user(data: UserIn, session=Depends(get_async_session)):
data.password = get_hashed_password(data.password)
new_user = User(**data.dict())
session.add(new_user)
await session.flush()
token = create_token()
new_token = Token(user_id=new_user.id, token=token)
session.add(new_token)
await session.commit()
return UserOut(token=token, username=data.username, email=data.email)
tests.conftest.py
DATABASE_URL_TEST = f"postgresql+asyncpg://{TEST_DB_USER}..."
engine = create_async_engine(DATABASE_URL_TEST)
test_async_session_maker = async_sessionmaker(engine, expire_on_commit=False)
async def override_get_async_session() -> AsyncGenerator[AsyncSession, None]:
async with test_async_session_maker() as session:
yield session
app.dependency_overrides[get_async_session] = override_get_async_session
@pytest.fixture(scope="session")
async def ac() -> AsyncGenerator[AsyncClient, None]:
async with AsyncClient(app=app, base_url="http://0.0.0.0:8000/api/v1/") as ac:
yield ac
测试.test_auth.py
@pytest.mark.asyncio
async def test_router_register_user(ac: AsyncClient, test_db: AsyncSession):
data: UserIn = UserIn(
email="[email protected]",
username="test",
password="test1234"
)
body = data.dict()
response = await ac.post("/auth/register", json=body)
assert response.status_code == 422 or 1 == 1
我花了很多时间来解决这个问题,并找到了另一种解决方案。它适用于我的情况,希望对您也有用。
首先,我将第一个数据库连接参数移至文件 .env:
DB_USER = test
DB_PASS = 1234
DB_HOST = localhost
DB_NAME = fastapiproj
DB_PORT = 3306
其次,我将第二个数据库(测试)连接参数移至文件 .test.env:
DB_USER = test
DB_PASS = 1234
DB_HOST = localhost
DB_NAME = fastapiproj_test
DB_PORT = 3306
之后我使用以下代码创建了 config.py 文件:
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
DB_USER: str
DB_PASS: str
DB_HOST: str
DB_NAME: str
DB_PORT: int
MODE: str
SECRETa: str
@property
def DB_URL(self):
return f"mysql+aiomysql://{self.DB_USER}:{self.DB_PASS}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
model_config = SettingsConfigDict(env_file=".env", extra='allow')
settings = Settings()
之后我安装了 pytest-dotenv 并创建了文件 pytest.ini:
[pytest]
pythonpath = . src
env_files =
.test.env
asyncio_mode = auto
这允许我覆盖原始文件 .env 进行测试,因此 sessionmaker 根据不同的文件创建会话,具体取决于您运行的内容:应用程序或应用程序的测试。