Is there anyway to test the query string generated from repository functions?

[Al3xDo]

Hi everyone,

I want to test the query string generated by the repository function only. My goal is to make the unit test as independent as possible and avoid creating a dedicated database instance during testing.

Here's what my code looks like:

tests/app/user/adapter/output/persistence/sqlalchemy/test_user.py

@pytest.mark.asyncio
async def test_get_users():
    """
    Test the get_users method of UserSQLAlchemyRepo.
    """
    mock_session = AsyncMock()
    mock_session.execute.return_value.scalars.return_value.all.return_value = []

    with patch('app.user.adapter.output.persistence.sqlalchemy.user.session_factory', return_value=mock_session):
        await user_repo.get_users(limit=5, prev=10)

    # Assert the query
    query = select(User).where(User.id < 10).limit(5)
    mock_session.execute.assert_called_once_with(query)

However, I encountered this error:

tests/app/user/adapter/output/persistence/sqlalchemy/test_user.py F                                                                                                               [100%]

======================================================================================= FAILURES ========================================================================================
____________________________________________________________________________________ test_get_users _____________________________________________________________________________________

    @pytest.mark.asyncio
    async def test_get_users():
        """
        Test the get_users method of UserSQLAlchemyRepo.
        """
        mock_session = AsyncMock()
        mock_session.execute.return_value.scalars.return_value.all.return_value = []
    
        with patch('app.user.adapter.output.persistence.sqlalchemy.user.session_factory', return_value=mock_session):
>           await user_repo.get_users(limit=5, prev=10)

tests/app/user/adapter/output/persistence/sqlalchemy/test_user.py:22: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <app.user.adapter.output.persistence.sqlalchemy.user.UserSQLAlchemyRepo object at 0x10e7ae950>

    async def get_users(
        self,
        *,
        limit: int = 12,
        prev: int | None = None,
    ) -> list[User]:
        query = select(User)
    
        if prev:
            query = query.where(User.id < prev)
    
        if limit > 12:
            limit = 12
    
        query = query.limit(limit)
        async with session_factory() as read_session:
            result = await read_session.execute(query)
>       return result.scalars().all()
E       AttributeError: 'coroutine' object has no attribute 'all'

app/user/adapter/output/persistence/sqlalchemy/user.py:26: AttributeError
================================================================================ short test summary info ================================================================================
FAILED tests/app/user/adapter/output/persistence/sqlalchemy/test_user.py::test_get_users - AttributeError: 'coroutine' object has no attribute 'all'
=================================================================================== 1 failed in 0.27s ===================================================================================
sys:1: RuntimeWarning: coroutine 'AsyncMockMixin._execute_mock_call' was never awaited

I've tried many ways but could not resolve the issue.

Here is the function that I want to test:

fastapi-boilerplate/app/user/adapter/output/persistence/sqlalchemy/user.py

Lines 8 to 27 in ca3bb3f

	class UserSQLAlchemyRepo(UserRepo):
	async def get_users(
	self,
	*,
	limit: int = 12,
	prev: int | None = None,
	) -> list[User]:
	query = select(User)
	if prev:
	query = query.where(User.id < prev)
	if limit > 12:
	limit = 12
	query = query.limit(limit)
	async with session_factory() as read_session:
	result = await read_session.execute(query)
	return result.scalars().all()

@Al3xDo Hmm, why did you create a test for the repository? I believe you should focus on creating unit tests at the service layer, where you can mock all dependencies of the service.

However, if you need integration tests, you can consider using Testcontainers. It's simple to use and performs well.

[Al3xDo]

Hi @vuxmai ,
I think the repository should be tested as well if that function includes some logic/conditions.
In the example code above, you can see it has the condition inside the repository function:

if limit > 12:
   limit = 12

you may think that we can have the integration test to cover that but for me, the integration test has some drawbacks:

• it consumes more resources and time
• needs to spend time setting up mock data, especially worse if we need to set-up data for many tables and also deal with many foreign keys constraints
• there are also some features that are only in PostgreSQL but not in SQL Lite (if we choose to create a small testing db like in-mem db) like full-text-search

I think testing the SQL query generated by the repository function is not only covering all the logic but also fast, and does not require external running service (mock db instance).