25. Mocking en código asíncrono con AsyncMock

25.1 Objetivo del tema

Cuando el código usa async y await, los mocks comunes no siempre alcanzan. Para simular funciones asíncronas, Python ofrece AsyncMock.

En este tema veremos cómo probar servicios asíncronos, configurar respuestas, simular errores y verificar que una dependencia fue esperada con await.

25.2 Importar AsyncMock

AsyncMock se importa desde unittest.mock:

from unittest.mock import AsyncMock

Un AsyncMock puede ser esperado con await y devuelve el valor configurado en return_value.

25.3 Instalar soporte async para pytest

Para escribir pruebas asíncronas con pytest, normalmente se usa el plugin pytest-asyncio:

pip install pytest-asyncio

Luego marcamos la prueba con @pytest.mark.asyncio.

25.4 Función asíncrona de ejemplo

Supongamos esta función:

async def obtener_nombre_usuario(usuario_id, cliente_api):
    usuario = await cliente_api.obtener_usuario(usuario_id)

    if usuario is None:
        return "Usuario no encontrado"

    return usuario["nombre"]

cliente_api.obtener_usuario es una dependencia asíncrona.

25.5 Primera prueba con AsyncMock

Prueba:

import pytest
from unittest.mock import AsyncMock


@pytest.mark.asyncio
async def test_obtener_nombre_usuario():
    cliente_api = AsyncMock()
    cliente_api.obtener_usuario.return_value = {
        "id": 1,
        "nombre": "Ana",
    }

    nombre = await obtener_nombre_usuario(1, cliente_api)

    assert nombre == "Ana"
    cliente_api.obtener_usuario.assert_awaited_once_with(1)

La verificación correcta es assert_awaited_once_with, no assert_called_once_with.

25.6 Diferencia entre llamado y esperado

En código asíncrono, una coroutine puede ser llamada pero no esperada. Por eso AsyncMock ofrece aserciones específicas:

• assert_awaited()
• assert_awaited_once()
• assert_awaited_with(...)
• assert_awaited_once_with(...)
• assert_not_awaited()

25.7 Caso donde devuelve None

Podemos probar el caso de usuario inexistente:

@pytest.mark.asyncio
async def test_obtener_nombre_usuario_inexistente():
    cliente_api = AsyncMock()
    cliente_api.obtener_usuario.return_value = None

    nombre = await obtener_nombre_usuario(99, cliente_api)

    assert nombre == "Usuario no encontrado"
    cliente_api.obtener_usuario.assert_awaited_once_with(99)

25.8 AsyncMock como función

Si la dependencia inyectada es una función asíncrona:

async def crear_codigo_async(generar_token):
    token = await generar_token()
    return f"COD-{token}"

Prueba:

@pytest.mark.asyncio
async def test_crear_codigo_async():
    generar_token = AsyncMock(return_value="ABC123")

    codigo = await crear_codigo_async(generar_token)

    assert codigo == "COD-ABC123"
    generar_token.assert_awaited_once_with()

25.9 side_effect con AsyncMock

side_effect también funciona con AsyncMock. Podemos simular errores:

async def consultar_con_manejo(cliente_api):
    try:
        return await cliente_api.consultar()
    except TimeoutError:
        return {"error": "timeout"}

Prueba:

@pytest.mark.asyncio
async def test_consultar_con_manejo_timeout():
    cliente_api = AsyncMock()
    cliente_api.consultar.side_effect = TimeoutError("Tiempo agotado")

    resultado = await consultar_con_manejo(cliente_api)

    assert resultado == {"error": "timeout"}
    cliente_api.consultar.assert_awaited_once_with()

25.10 Respuestas secuenciales

También podemos devolver valores distintos en cada await:

async def obtener_dos_paginas(cliente_api):
    pagina_1 = await cliente_api.obtener_pagina(1)
    pagina_2 = await cliente_api.obtener_pagina(2)
    return pagina_1 + pagina_2

Prueba:

@pytest.mark.asyncio
async def test_obtener_dos_paginas():
    cliente_api = AsyncMock()
    cliente_api.obtener_pagina.side_effect = [
        ["Ana"],
        ["Luis"],
    ]

    usuarios = await obtener_dos_paginas(cliente_api)

    assert usuarios == ["Ana", "Luis"]
    assert cliente_api.obtener_pagina.await_count == 2

25.11 await_args_list

Para inspeccionar varios awaits, usamos await_args_list:

def test_fragmento():
    assert cliente_api.obtener_pagina.await_args_list[0].args == (1,)
    assert cliente_api.obtener_pagina.await_args_list[1].args == (2,)

En una prueba completa:

@pytest.mark.asyncio
async def test_obtener_dos_paginas_verifica_argumentos():
    cliente_api = AsyncMock()
    cliente_api.obtener_pagina.side_effect = [[], []]

    await obtener_dos_paginas(cliente_api)

    assert cliente_api.obtener_pagina.await_args_list[0].args == (1,)
    assert cliente_api.obtener_pagina.await_args_list[1].args == (2,)

25.12 assert_has_awaits

También podemos usar call con assert_has_awaits:

from unittest.mock import call


@pytest.mark.asyncio
async def test_obtener_dos_paginas_con_assert_has_awaits():
    cliente_api = AsyncMock()
    cliente_api.obtener_pagina.side_effect = [[], []]

    await obtener_dos_paginas(cliente_api)

    cliente_api.obtener_pagina.assert_has_awaits([
        call(1),
        call(2),
    ])

Es útil cuando queremos verificar una secuencia de awaits.

25.13 Patch de función async

Si parcheamos una función async, podemos usar new_callable=AsyncMock:

from unittest.mock import AsyncMock, patch


@pytest.mark.asyncio
async def test_patch_funcion_async():
    with patch("usuarios.servicio.obtener_usuario_remoto", new_callable=AsyncMock) as obtener_mock:
        obtener_mock.return_value = {"id": 1, "nombre": "Ana"}

        resultado = await funcion_que_usa_obtener_usuario_remoto(1)

    assert resultado == "Ana"
    obtener_mock.assert_awaited_once_with(1)

La ruta debe seguir la misma regla: parchear donde se usa la dependencia.

25.14 Async context manager

Algunas librerías async usan async with. Para simularlas, configuramos __aenter__ y __aexit__:

async def obtener_estado(cliente):
    async with cliente.sesion() as sesion:
        respuesta = await sesion.get("/estado")
        return await respuesta.json()

Prueba:

@pytest.mark.asyncio
async def test_obtener_estado():
    cliente = AsyncMock()
    sesion = AsyncMock()
    respuesta = AsyncMock()
    respuesta.json.return_value = {"estado": "ok"}
    sesion.get.return_value = respuesta

    cliente.sesion.return_value.__aenter__.return_value = sesion

    estado = await obtener_estado(cliente)

    assert estado == {"estado": "ok"}
    sesion.get.assert_awaited_once_with("/estado")
    respuesta.json.assert_awaited_once_with()

25.15 Cuidado con mocks async demasiado anidados

Los context managers asíncronos pueden generar configuraciones difíciles de leer. Si la prueba se vuelve muy anidada, considera crear un fake async con métodos explícitos.

class RespuestaAsyncFake:
    def __init__(self, datos):
        self.datos = datos

    async def json(self):
        return self.datos

Los fakes async pueden ser más claros cuando se reutilizan.

25.16 Fake async simple

Ejemplo de cliente fake:

class ClienteApiAsyncFake:
    def __init__(self, usuarios):
        self.usuarios = usuarios

    async def obtener_usuario(self, usuario_id):
        return self.usuarios.get(usuario_id)

Prueba:

@pytest.mark.asyncio
async def test_obtener_nombre_usuario_con_fake_async():
    cliente_api = ClienteApiAsyncFake({
        1: {"id": 1, "nombre": "Ana"}
    })

    nombre = await obtener_nombre_usuario(1, cliente_api)

    assert nombre == "Ana"

25.17 Errores frecuentes

• Usar Mock para una función que debe esperarse con await.
• Olvidar @pytest.mark.asyncio en una prueba async.
• Verificar con assert_called_once_with cuando corresponde assert_awaited_once_with.
• No esperar la función bajo prueba con await.
• Configurar mocks async demasiado anidados cuando un fake sería más claro.

25.18 Ejercicio práctico

Prueba esta función:

async def enviar_notificacion_async(usuario_id, repositorio, notificador):
    usuario = await repositorio.buscar_por_id(usuario_id)

    if usuario is None:
        return False

    await notificador.enviar(usuario["email"], "Hola")
    return True

Escribe una prueba para usuario existente y otra para usuario inexistente usando AsyncMock.

25.19 Solución posible del ejercicio

Una solución:

import pytest
from unittest.mock import AsyncMock

from notificaciones import enviar_notificacion_async


@pytest.mark.asyncio
async def test_enviar_notificacion_async_usuario_existente():
    repositorio = AsyncMock()
    repositorio.buscar_por_id.return_value = {
        "id": 1,
        "email": "ana@example.com",
    }
    notificador = AsyncMock()

    resultado = await enviar_notificacion_async(1, repositorio, notificador)

    assert resultado is True
    repositorio.buscar_por_id.assert_awaited_once_with(1)
    notificador.enviar.assert_awaited_once_with("ana@example.com", "Hola")


@pytest.mark.asyncio
async def test_enviar_notificacion_async_usuario_inexistente():
    repositorio = AsyncMock()
    repositorio.buscar_por_id.return_value = None
    notificador = AsyncMock()

    resultado = await enviar_notificacion_async(99, repositorio, notificador)

    assert resultado is False
    repositorio.buscar_por_id.assert_awaited_once_with(99)
    notificador.enviar.assert_not_awaited()

25.20 Conclusión

AsyncMock permite simular dependencias asíncronas y verificar que fueron esperadas correctamente. Es clave distinguir entre llamar una coroutine y esperarla con await.

En el próximo tema veremos cómo diseñar pruebas legibles usando stubs y builders de datos.