dependency_overrides 在 FastAPI、Pytest 中不起作用

Question

对于我的测试，我想使用测试数据库，为此我使用依赖项覆盖。但是，我的端点也在访问生产数据库。

我的结构代码：

 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

Answer 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 根据不同的文件创建会话，具体取决于您运行的内容：应用程序或应用程序的测试。