Litestar
How to mock dependencies? #1789
-
SummaryCan you please add a dependency override to the application. Something similar to fastapi's app.dependency_overrides. Basic Example@pytest.fixture
def client(app):
database = TestAsyncSessionLocal()
def test_get_db(app):
try:
yield database
finally:
database.close()
app.dependency_overrides['db'] = Provide(test_get_db) # nothing like dependency_overrides in starlite
...Drawbacks and ImpactHere's what I tried: def client(app):
database = TestAsyncSessionLocal()
def test_get_db(app):
try:
yield database
finally:
database.close()
app.dependencies['db'] = Provide(test_get_db)
...This doesn't work because handlers are already registered in main:app. So overriding dependencies like this after handlers have been registered changes nothing. Unresolved questionsNo response |
Beta Was this translation helpful? Give feedback.
Replies: 9 comments 3 replies
-
|
Hi. It's basically already in place. The For controllers you can use attribute mocking. What's missing? |
Beta Was this translation helpful? Give feedback.
-
Thanks for your reply. But the create_test_client was built upon TestClient, and I want to use AsyncTestClient. |
Beta Was this translation helpful? Give feedback.
-
|
Expanding on what @Goldziher said: It's unlikely that we'll support this for a number of reasons (which we actually cover here in our FastAPI migration guide). The gist is that it's not actually needed and often a crude workaround that has more drawbacks than benefits. Overriding dependencies in such a way:
The solution to your problem is this: @pytest.fixture
def database():
database = TestAsyncSessionLocal()
try:
yield database
finally:
database.close()
@pytest.fixture
def client(app, mocker, database):
mocker.patch("path.to.get_db", return_value=database)This actually has several benefits:
If you do need to actually swap out the whole dependency function because you're performing some additional logic in there you can easily adapt this pattern: def get_database(app):
database = TestAsyncSessionLocal()
try:
yield database
finally:
database.close()
@pytest.fixture
def client(app, mocker, database):
mocker.patch("path.to.get_db", new=get_database)Just to illustrate what the proper way to override dependencies like you want would look like in contrast: # add a separate fixture for the database to ensure it's always closed
# no matter where errors might occur
@pytest.fixture
def database()
database = TestAsyncSessionLocal()
try:
yield database
finally:
if database.is_open:
database.close()
@pytest.fixture
def client(app, database):
def test_get_db(app):
try:
yield database
finally:
database.close()
try:
app.dependency_overrides['db'] = Provide(test_get_db)
finally:
app.dependency_overrides = {} # ensure to reset overrides at the end |
Beta Was this translation helpful? Give feedback.
-
|
Beta Was this translation helpful? Give feedback.
-
|
Thanks for your reply. I tried your solution using mock, but it doesn't work. The test still seems to be using my original db instead of the test_db. Here's how my app looks in the main.py: Please help |
Beta Was this translation helpful? Give feedback.
-
|
Can you provide an example of your app, the |
Beta Was this translation helpful? Give feedback.
-
app.main.py app.api.tests.conftest.py app.core.database.py Thank you. |
Beta Was this translation helpful? Give feedback.
-
|
A couple of things here, but let me just say that this isn't something Litestar specific but rather about how scopes, imports and mocking works.
To fix it you can either:
The second approach is the preferred one and one of the reasons we don't provide dependency overrides: It encourages bad practices. As you've seen, the fact that Litestar does not provide you with such capabilities has brought to light a deeper issue in your code. ExamplesHere are two simplified examples that you can apply to your existing code: Importing the
|
Beta Was this translation helpful? Give feedback.
-
Thank you very much sir. With your help, I've finally been able to resolve my issue. |
Beta Was this translation helpful? Give feedback.
-
|
the solution here explains quite well the reasonning, still the test fails for mainly 2 reasons: the using a context manager to patch should be preferred. see the diff here from a non-passing test to a passing one: euri10/mock_dependency_litestar@9434736 |
Beta Was this translation helpful? Give feedback.
-
|
added an example on how to mock a httpx client dependency that was put in the state in the repo above ;) |
Beta Was this translation helpful? Give feedback.
-
|
The accepted answer has a challenge: you effectively lose the ability to test the actual dependency's internals (problematic if you care about test coverage). I'm running into this with a mocked DB connection—MRE below:
from litestar.datastructures import State
from litestar.di import Provide
from sqlalchemy.ext.asyncio import AsyncSession
async def get_db_session_dependency(state: State) -> AsyncSession:
"""Define a dependency injector to get a DB session.
Args:
----
state: The Litestar app state.
Yields:
------
The session.
"""
sessionmaker = state.db_sessionmaker
raised_exc = None
async with sessionmaker() as session:
try:
yield session
except Exception as exc: # noqa: BLE001
await session.rollback()
raised_exc = exc
finally:
await session.close()
# If we encountered an exception inside the yielded dependency, we raise here to
# ensure it properly propagates up to our handlers:
if raised_exc:
raise raised_exc
DBSessionDependency = Provide(get_db_session_dependency)
"""Define the API."""
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from litestar import Litestar, Router
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from .dependencies import DBSessionDependency
DB_CONNECTION_STRING: Final[str] = "postgresql+psycopg://postgres:postgres@logs-db:5432/my-db"
@asynccontextmanager
async def lifespan_db_sessionmaker(app: Litestar) -> AsyncGenerator[None, None]:
"""Define a lifespan context manager for creating a DB sessionmaker.
Args:
----
app: The Litestar app.
"""
engine = create_async_engine(DB_CONNECTION_STRING)
app.state.db_sessionmaker = async_sessionmaker(autocommit=False, bind=engine)
try:
yield
finally:
await engine.dispose()
def create_app() -> Litestar:
"""Create the app.
Returns
-------
The app.
"""
# Create the API router:
api_router = Router(
path="/api",
route_handlers=[
Router(
path="/v1",
route_handlers=[HEALTH_ROUTER],
)
],
)
# Create the app:
app = Litestar(
dependencies={
"db_session": DBSessionDependency,
},
lifespan=[lifespan_db_sessionmaker],
)
return app
from collections.abc import AsyncGenerator, Generator
from unittest.mock import AsyncMock, MagicMock, Mock, patch
from litestar.testing import AsyncTestClient
import pytest
import pytest_asyncio
@pytest.fixture(name="patch_sqlalchemy")
def _patch_sqlalchemy_fixture(
mock_sqlalchemy_sessionmaker: MagicMock,
) -> Generator[None, None, None]:
"""Define a patch for SQLAlchemy.
Args:
----
mock_sqlalchemy_sessionmaker: The mock SQLAlchemy session maker.
"""
with patch(
"src.app.create_async_engine", return_value=Mock(dispose=AsyncMock())
), patch("src.app.async_sessionmaker", return_value=mock_sqlalchemy_sessionmaker):
yield
@pytest_asyncio.fixture(name="http_client")
async def http_client_fixture(
patch_sqlalchemy: Generator[None, None, None],
) -> AsyncGenerator[AsyncTestClient, None]:
"""Define a test HTTP client.
Args:
----
patch_sqlalchemy: A patch for SQLAlchemy.
Yields:
------
The HTTP client.
"""
app = create_app()
async with AsyncTestClient(app=app) as client:
yield client
@pytest.fixture(name="mock_sqlalchemy_sessionmaker")
def mock_sqlalchemy_sessionmaker_fixture(mock_sqlalchemy_session: Mock) -> MagicMock:
"""Define a mock SQLAlchemy session maker.
Args:
----
mock_sqlalchemy_session: The mock SQLAlchemy session.
Returns:
-------
The mock SQLAlchemy session maker.
"""
return MagicMock(
return_value=MagicMock(
__aenter__=AsyncMock(return_value=mock_sqlalchemy_session),
__aexit__=AsyncMock(),
)
)
@pytest.fixture(name="mock_sqlalchemy_session")
def mock_sqlalchemy_session_fixture(
mock_sqlalchemy_session_commit: AsyncMock,
mock_sqlalchemy_session_execute: AsyncMock,
mock_sqlalchemy_session_rollback: AsyncMock,
) -> Mock:
"""Define a mock SQLAlchemy session.
Args:
----
mock_sqlalchemy_session_commit: The mock SQLAlchemy session commit coroutine.
mock_sqlalchemy_session_execute: The mock SQLAlchemy session execute coroutine.
mock_sqlalchemy_session_rollback: The mock SQLAlchemy session rollback
coroutine.
Returns:
-------
The mock SQLAlchemy session.
"""
return Mock(
add=Mock(),
close=AsyncMock(),
delete=AsyncMock(),
execute=mock_sqlalchemy_session_execute,
commit=mock_sqlalchemy_session_commit,
refresh=AsyncMock(),
rollback=mock_sqlalchemy_session_rollback,
)In production, my dependency returns a If I merely create a replacement dependency that returns a I know there are ways around this:
...but it would be great if |
Beta Was this translation helpful? Give feedback.
A couple of things here, but let me just say that this isn't something Litestar specific but rather about how scopes, imports and mocking works.
create_appfunction and then immediately calling it in the same file is functionally equivalent to simply creating the app in the file directly.app. This will also set the dependencies on that objectapp.core.database.get_db. Since the application object has already been created, this will not change theget_dbfunction you passed to itTo fix it you can either:
get_db…