Fastapi: [Question] cannot override `app.dependency_override`

[koralle]

I'm using fastapi to create a backend app connect to MySQL 5.7 (on Google Cloud SQL) with sqlalchemy.

When testing (with pytest), I want to switch the connected DB to SQLite3.

However, when I execute pytest, the test fails because DB has not been switched to SQLite3.
How can I fix it?

conftest.py as follows:

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy_utils import database_exists, drop_database
from src.app.dependencies.database import Base

class User(Base):
     __tablename__ = "User"
     id = Column("id", Integer, primary_key=True, autoincrement=True)
     name = Column("name", String)

@pytest.fixture(scope="function")
def SessionLocal():
    # settings of test database
    TEST_SQLALCHEMY_DATABASE_URL = "sqlite:///./test_temp.db"
    engine = create_engine(
        TEST_SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
    )

    assert not database_exists(
        TEST_SQLALCHEMY_DATABASE_URL
    ), "Test database already exists. Aborting tests."

    Base.metadata.create_all(engine)
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

    yield SessionLocal

    drop_database(TEST_SQLALCHEMY_DATABASE_URL)

tests/test_user.py :

def temp_db(f):
    def func(SessionLocal, *args, **kwargs):

        def override_get_db():
            try:
                db = SessionLocal()
                yield db
            finally:
                db.close()

        # cannot override app's dependency...
        app.dependency_overrides[get_db] = override_get_db
        # Run tests
        f(*args, **kwargs)

        app.dependency_overrides[get_db] = get_db
    return func

client = TestClient(app)


@temp_db
def test_fetch_all_users():
    response = client.get("/api/users")
    assert response.status_code == 200
    assert len(response.json()["data"]) == 0
    # Expect 0 (on SQLite3), but actual 23 (on MySQL).

My dependency and router:

def get_db():
    db = SessionLocal()
    try:
        yield db
    except Exception:
        db.close()

@router.get("/api/users", response_model=FetchAllUsersResponse)
async def get_all_users(db: Session = Depends(get_db)):
    all_users = db.query(db_User).all()

    return FetchAllUsersResponse(
        status=UserStatus.ok,
        data=[
            User(user_id=user.user_id, name=user.name)
            for user in all_users
        ],
    )

app.include_router(
    router, responses={404: {"description": "Not found"}}
)

Test result:

tests/test_user.py:21: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
    @temp_db
    def test_fetch_all_users():
        response = client.get("/api/users")
        assert response.status_code == 200
>       assert len(response.json()["data"]) == 0
E       assert 23 == 0
E         +23
E         -0

tests/test_user.py:35: AssertionError

[Mause]

I can't see any glaring issues in your code - when you run it in a debugger, does override_get_db get run?

[koralle]

@Mause

I can't see any glaring issues in your code - when you run it in a debugger, does override_get_db get run?

No.
To support this, I added import pdb; pdb.set_trace() in override_get_db().
When I executed pytest, I could not start pdb.

def override_get_db():
+   import pdb; pdb.set_trace()
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()

Next, I added the same statements in original dependency, get_db() instead of override_get_db().
When I executed pytest, pdb was started.

def get_db():
+   import pdb; pdb.set_trace()
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()

And, I confirmed that the SQLite3 DB for testing was created as expected.
I can't understand why app dependency, get_db() override failes...

# conftest.py
import pytest
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy_utils import database_exists, drop_database
from src.app.dependencies.database import Base

class User(Base):
     __tablename__ = "User"
     id = Column("id", Integer, primary_key=True, autoincrement=True)
     name = Column("name", String)

@pytest.fixture(scope="function")
def SessionLocal():
    # settings of test database
    TEST_SQLALCHEMY_DATABASE_URL = "sqlite:///./test_temp.db"
    engine = create_engine(
        TEST_SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
    )

    assert not database_exists(
        TEST_SQLALCHEMY_DATABASE_URL
    ), "Test database already exists. Aborting tests."

    Base.metadata.create_all(engine)
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

+   db = SessionLocal()
+   records = db.query(User).all()
+   assert len(records) == 0 # PASSED. (I only created Table 'User'.)

    yield SessionLocal

    drop_database(TEST_SQLALCHEMY_DATABASE_URL)

[Mause]

Are you able to provide a small reproduction of the issue I can use to debug and help?

[koralle]

@Mause

Sorry for the delay in contacting you.
After reviewing the project structure from the beginning while creating a small reproduction,
the problem was solved.
I still don't know what the root cause is, but I can now run the tests with a clean SQLite3 DB.

Thank you!!

[Mause]

If your problem is solved, can you please close this issue?