Unit testing a Python AWS Lambda handler function that uses asyncio with its own event loop
If you’re using asyncio in an AWS Lambda, the Lambda handler function has to manage the asyncio event loop inside the Lambda, as the handler function itself cannot be an async function.
This is fairly easy to do by having the top-level Lambda handler function use
asyncio.run to call an asynchronous function that does the actual work:
import asyncio
async def foobar_handler() -> str:
return "foobar"
def lambda_handler(event, context) -> str:
return asyncio.run(main())
Or you might be using AWS Lambda Powertools to do async batch processing:
from aws_lambda_powertools.utilities.batch import (
AsyncBatchProcessor,
EventType,
async_process_partial_response,
)
from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
processor = AsyncBatchProcessor(event_type=EventType.SQS)
async def async_record_handler(record: SQSRecord) -> None:
...
def lambda_handler(event, context):
return async_process_partial_response(
event=event,
record_handler=async_record_handler,
processor=processor,
context=context,
)
In either case, the Lambda is managing its own asyncio event loop internally.
You might be using pytest-asyncio to run tests against this Lambda handler function, e.g. like this:
@pytest.mark.asyncio
async def test__foobar_async_lambda_handler(
lambda_event: dict,
lambda_context: LambdaContext,
):
...
lambda: lambda_handler(
lambda_event,
lambda_context,
)
...
Unfortunately that will error during the test process due to the existing event loop for the tests:
E RuntimeError: asyncio.run() cannot be called from a running event loop
The workaround solution is to use run_in_executor to run the Lambda handler in
a thread, so that it can manage its own event loop internally as normal:
@pytest.mark.asyncio
async def test__foobar_async_lambda_handler(
lambda_event: dict,
lambda_context: LambdaContext,
):
...
await asyncio.get_event_loop().run_in_executor(
None,
lambda: lambda_handler(
lambda_event,
lambda_context,
)
)
...