Standard Mocking Patterns for Native SQL query executions + Fast API with Async Calls

[sarathsund]

Hi community,

Framework - Fastapi
Version - python 3.11
Database - AWS RDS Postgresql

I would to understand how to mock the database layers with async methods with pytest-asyncio only. incase of executing direct sql queries, is it possible to mock the connection, cursor object and the mocked result sent for maximum unit test coverage.

Method Calling

  async def get_session(self, user_id: str) -> list[csession]:
    result = await self._db_service.run(SQL_SELECT_USER_SESSIONS, {'user_id': user_id})
    return result

Method Declaration

async def run(self, sql: str, params: dict[str,any] = None) -> any:
    async with self._connection_pool.connection() as connection:
      async with connection.cursor() as cursor:
        try:
          await cursor.execute(sql, params)
          try: 
            return await cursor.fetchall()
          except ProgrammingError as e:
            return cursor.statusmessage
        except psycopg.Error as e:
          raise HTTPException(500, str(e))

Thanks for the support

[sarathsund]

@seifertm review requested

[seifertm]

My understanding is that you want to test calls inside your run method, for example that cursor.execute is called with the correct sql argument.

Pytest-asyncio merely offers setup and teardown of asyncio event loops. It doesn't offer anything that will make mocking easier. Nevertheless, I'll first try to answer your question specifically and then answer more broadly.

Specific answer

It's possible to use mock.AsyncMock from the standard library to mock coroutines and the like.

In your case, you'd have to create a Mock that returns a mocked connection when __aenter__ is invoked (i.e. when the async with context is entered). The mock object returned by __aenter__ represents your connection variable. You can nest your mocks further like that. In pseudo code, it would look something like this:

connection = Mock()
connection_factory = AsyncMock()
connection_factory.__aenter__.return_value = connection
# More nesting goes here
connection_pool = Mock()
connection_pool.connection.return_value = connection_factory

If you set up this kind of mock structure, you can the do assertions on the mock objects.

Broad answer

Personally, I try to avoid mocking third-party code as much as possible. By mocking psycopg connections and cursors etc., the developer has to construct a bunch of mock objects. These mock objects make assumptions on how their psycopg counterparts behave. Most of the time, those assumptions are incomplete and/or wrong. This generally results in (lots of) test code that passes, even though the code under test is wrong.

That being said, I'd probably try to test the database interaction in a larger-than-unit test. For example, using _db_service to write user information to an actual database, then retrieve the user information from the db and see if the result is correct.

Once these tests get annoyingly slow, you can optimize from there. For example, you can provide two different implementations for _db_service, one backed by psycopg, another one using an in-memory store.

This gets easier if the user of _db_service doesn't have to pass SQL statements, but if _db_service offers more high-level functions, such as get_user_sessions. At that point, it doesn't matter if the underlying data store has an SQL engine or not.

Also: It's fine to mock _db_service in code that uses it, because you're the owner/implementor of _db_service.

This is my personal opinion on the matter. I'm aware that it's unpopular :) But in my experience it's highly effective.