为什么 alembic 升级在 pytest 中的测试数据库上不起作用

Question

总的来说，我对测试还很陌生。我使用 postgreSQL 和 SQLAlchemy 作为数据库，使用 FastAPI 后端和 Alembic 进行迁移。我想建立一个测试数据库进行测试。我有要在测试数据库上执行的自定义迁移脚本。这是一个示例test_users文件。

import pytest
import alembic
from alembic.config import Config

from fastapi.testclient import TestClient
import sqlalchemy as sa
from sqlalchemy import create_engine
from sqlalchemy_utils import create_database, database_exists, drop_database
from sqlalchemy.orm import sessionmaker
`
TEST_DB_URL = f"{DATABASE_URL}_test"

if database_exists(TEST_DB_URL):
    drop_database(TEST_DB_URL)

create_database(TEST_DB_URL)

alembic_cfg = Config("alembic.ini")
alembic_cfg.set_main_option('sqlalchemy.url', str(TEST_DB_URL))
alembic.command.upgrade(alembic_cfg, "head")

engine = create_engine(TEST_DB_URL)
TestingSessionLocal = sessionmaker()


@pytest.fixture(scope='session')
def connection():
    connection = engine.connect()
    yield connection
    connection.close()

@pytest.fixture
def session(connection):
    transaction = connection.begin()
    session = TestingSessionLocal(bind=connection)
    yield session
    session.close()
    transaction.rollback()
    # drop_database(TEST_DB_URL)

@pytest.fixture()
def client(session):
    def override_get_db():
        yield session

    app.dependency_overrides[get_db] = override_get_db
    client = TestClient(app)
    yield client
    del app.dependency_overrides[get_db]

def test_user_register(client):
    response = client.post("/register", json={"email": "def@def.com", "master_pwd": "pass"})
    print(response)
    assert response.status_code == 200

但当我检查 pgadmin 时，测试数据库已创建，但未应用迁移，并且运行 pytest 会出现 404 错误。我究竟做错了什么？

Answer 1

我知道这是一个较旧的线程，但我遇到了同样的问题，并且这个问题出现在我的谷歌搜索中。

就我而言，问题是env.py我有以下行：

config.set_main_option("sqlalchemy.url", settings.DATABASE_URL)

其中settings.DATABASE_URL是包含从环境变量读取的数据库连接的变量（以避免在alembic.ini文件中对其进行硬编码）。

现在，当您调用时，也会隐式alembic.command.upgrade(alembic_cfg, "head")运行env.py并覆盖您的alembic_cfg.set_main_option('sqlalchemy.url', str(TEST_DB_URL))真实 DATABASE_URL。

我settings通过模拟我的模块解决了这个问题

from core import settings as mocked_settings
mocked_settings.DATABASE_URL = TEST_DB_URL
mocker.patch.dict("sys.modules", {"core.settings": mocked_settings})

其中mocker指的是模块中的夹具（不过pytest-mock您也可以使用标准库）。mock

如果您直接从环境变量中读取数据库 URL，env.py可能会更容易：

mocker.patch.dict("os.environ", DATABASE_URL=TEST_DB_URL)

希望对其他人有帮助！:)