I am trying to setup testing on a FastAPI project using Pytest and also a database. I want it such that the database is created on every test and destroyed after every test.

My setup worked, but then, it fell apart when I introduced factories using the factory-boy package. I wanted to be able to use the factory-boy package to generate objects that will be persisted to the test database that I can use in my testcase.

The problem I have is defining the session that the factory class would use as factory-boy requires a session for the SQLAlchemy model.

NB: I use the dependency-injector library in my code to pass in dependencies to classes across my project.

In my db.py class, I have the database setup like this, a Database class for the actual application and a TestDatabase class that would be used in tests. The intention is that I’d override the database in the test environment.

class Database:
    def __init__(self, db_url: str) -> None:
        print("Initializing DB...")
        import pdb

        pdb.set_trace()
        engine_kwargs = {"echo": True if settings.DEBUG else False}
        self._engine = create_engine(db_url, **engine_kwargs)
        self._session_factory = scoped_session(
            sessionmaker(
                class_=Session,
                autocommit=False,
                autoflush=False,
                bind=self._engine,
            )
        )

    def session_factory(self):  # noqa
        return self._session_factory()

    @contextmanager
    def session(self) -> Callable[..., AbstractContextManager[Session]]:
        session: Session = self._session_factory()
        try:
            yield session
        except Exception as e:
            print(e)
            session.rollback()
            raise
        finally:
            session.close()


class TestDatabase(Database):
    def __init__(self, db_url: str) -> None:  # noqa
        import pdb

        pdb.set_trace()
        # super().__init__(db_url)
        print("Initializing DB...")
        engine_kwargs = {
            "echo": True,
            "connect_args": {
                "check_same_thread": False,
            },
            "poolclass": StaticPool,
        }
        self._engine = create_engine(db_url, **engine_kwargs)

        # Create database tables for for test environment.
        SQLModel.metadata.create_all(bind=self._engine)

        self._session_factory = scoped_session(
            sessionmaker(
                class_=Session,
                autocommit=False,
                autoflush=False,
                bind=self._engine,
            )
        )

I also have a container for the dependency injection setup:

class AppContainer(containers.DeclarativeContainer):
    """
    DI container for the project.
    """

    # Auto-wiring
    wiring_config = containers.WiringConfiguration(packages=["app.api"])

    # Singletons
    db = providers.Singleton(Database, db_url=settings.DATABASE_URI)

    # Repositories
    book_repository = providers.Factory(
        BookRepository,
        session_factory=db.provided.session_factory,
    )

    # Services
    book_service = providers.Factory(
        BookService,
        book_repository=book_repository,
    )

In my conftest.py, I have these fixtures setup, one to override the db before returning the container, one to return a session, and another for the test client:

@pytest.fixture()
def app_container():
    container = AppContainer()

    container.db.override(
        providers.Singleton(
            TestDatabase,
            db_url="sqlite:///:memory:",
        )
    )

    return container


@pytest.fixture()
def db_session(app_container):
    return app_container.db().session_factory()


@pytest.fixture()
def test_client():
    with TestClient(application) as test_client:
        yield test_client

In my factories.py, I have a factory class setup:

app_container = AppContainer()
app_container.db.override(
    providers.Singleton(
        TestDatabase,
        db_url="sqlite:///:memory:",
    )
)
session = app_container.db().session_factory()


class BookFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = Book
        sqlalchemy_session = session

    name = "Test Book"
    author = factory.Faker("name")

Then in a sample test function, I write a test like this:

def test_store_book(test_client, app_container, db_session):
    BookFactory(name="Sample Book Name 2")

    payload = {
        "name": "Sample Book Name",
        "author": "John Doe",
    }

    response = test_client.post("/books", json=payload)

    books = db_session.query(Book).all()

    assert response.status_code == 200
    assert len(books) == 2

My expectation from this test is that the book factory would create a book in the database and the endpoint will create another book in the database so that when I fetch the number of books from the database, the length of the books list should be 2. But that’s not the case here, it returns 1. My guess is that the database is being destroyed and recreated. So, the database where the factory stores the initial book created from the factory is not the same as the one the endpoint will store the book.

What I want is to have a single database through out the lifecycle of a single test case so that I can use factories to store items in the database and also store item via the endpoint and I’d see both of them.