test aiomock

test_aiomock.py

from unittest.mock import Mock
import asyncpg
import pytest
from async_lru import alru_cache


@alru_cache
async def run():
    conn = await asyncpg.connect(user='user', password='pass', database='db', host='127.0.0.1')
    await conn.close()


class AsyncMock(Mock):

    def __call__(self, *args, **kwargs):
        sup = super(AsyncMock, self)
        async def coro():
            return sup.__call__(*args, **kwargs)
        return coro()

    def __await__(self):
        return self().__await__()


@pytest.fixture
def mock_asyncpg(monkeypatch):
    """ Patch asyncpg.connect """
    monkeypatch.setattr(asyncpg, "connect", AsyncMock())


@pytest.fixture(autouse=True)
def clear_alru_cache():
    run.cache_clear()


@pytest.mark.asyncio
async def test_mocked(mock_asyncpg):
    await run()


@pytest.mark.asyncio
async def test_not_mocked():
    with pytest.raises(asyncpg.exceptions.InvalidAuthorizationSpecificationError):
        await run()

$ pytest -v
============================================= test session starts =============================================
platform linux -- Python 3.6.9, pytest-5.1.3, py-1.8.0, pluggy-0.13.0 -- /home/hoefling/projects/.venvs/stackoverflow/bin/python3.6
cachedir: .pytest_cache
rootdir: /home/hoefling/projects/private/stackoverflow/so-58205688
plugins: mock-1.11.0, cov-2.7.1, xdist-1.29.0, asyncio-0.10.0, forked-1.0.2, testinfra-3.2.0
collected 2 items                                                                                             

test_spam.py::test_mocked PASSED                                                                        [ 50%]
test_spam.py::test_not_mocked PASSED                                                                    [100%]

============================================== 2 passed in 0.07s ==============================================