FastAPI 单元测试不覆盖 get_db

Question

我只是想让 FastAPI 单元测试与 SQLAlchemy 一起工作，但我在测试数据库中创建的对象时遇到了麻烦。根据docs，我有以下设置。

main.py

routes = [
    APIRoute('/games/', views.games_list, name='index', response_class=HTMLResponse),
]

settings = Settings()

app = FastAPI(debug=settings.debug, routes=routes)

views.py

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


async def games_list(request: Request, db: Session = Depends(get_db)):
    settings = Settings()
    games = db.query(Game).all()
    return settings.templates.TemplateResponse('games/list.jinja', {'request': request, 'games': games})

数据库.py

def prepare_database(settings):
    engine = create_engine(settings.database_url)
    Base.metadata.create_all(engine)
    return engine


engine = prepare_database(delete_existing=False, settings=Settings())
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

conftest.py

@pytest.fixture
def db_conn():
    session = SessionLocalTesting()
    try:
        yield session
    finally:
        session.close()

@pytest.fixture
def cli(db_conn):
    def override_get_db():
        session = SessionLocalTesting()
        try:
            yield session
        finally:
            session.close()

    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as client:
        yield client

测试文件

def test_games_list(cli, factory, db_conn):
    factory.create_game()

    # This will return the game I have created with my factory, definitely in the test db.
    print(db_conn.query(Game.name).all())

    r = cli.get('/games/')
    assert 'DnD Game' in r.content.decode()

我的问题是我无法从测试数据库中获取对象。如果我在视图中print(db.bind.database.url ) 没有使用测试数据库，那么它会尝试从真实数据库中获取项目，而不是测试数据库。

所以看起来get_db 没有被覆盖，尽管我不确定为什么。

提前致谢

Answer 1

我认为您应该在调用夹具时忘记指定夹具的范围

通过@pytest.fixture 创建一个数据库连接并覆盖数据库依赖项pytest.fixture(scope='session') 中有一种范围可用，module 级别也有，Type of scope 中还有其他范围之外的可用范围

我提到了一个可以帮助你的链接Use test DB intend of real

import pytest

from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy_utils import create_database, drop_database

from app.main import app
from db.base import Base
from db.settings import get_database

from settings.config import (
    TEST_DATABASE_URI as SQLALCHEMY_DATABASE_URL,
)


# SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL, pool_pre_ping=True)


def override_get_db():
    """" Override """
    try:
        db = Session(autocommit=False, autoflush=False, bind=engine)
        yield db
    finally:
        db.close()


@pytest.fixture(scope="session", autouse=True)
def create_db():
    """ creating db model in database """

    create_database(SQLALCHEMY_DATABASE_URL)
    print("\n" + "\x1b[6;30;42m" + "Creating test database." + "\x1b[0m")

    Base.metadata.create_all(bind=engine)
    app.dependency_overrides[get_database] = override_get_db

    yield 1

    drop_database(SQLALCHEMY_DATABASE_URL)
    print("\n" + "\x1b[6;30;42m" + "Delete test database." + "\x1b[0m")


@pytest.fixture()
def get_db_session():
    """ Getting session for db transaction """
    session = Session(autocommit=False, autoflush=False, bind=engine)
    yield session

    session.close()


@pytest.fixture()
def client():
    """ Getting testclient of app """
    with TestClient(app) as client:
        yield client

如果您仍然遇到问题或不告诉我