How to write unit test for a function with async context managers
async def post_message(url, data, msg_id):
async with aiohttp.ClientSession() as session:
try:
async with session.post(url, json=data, ssl=False) as resp:
result = await resp.text()
resp.raise_for_status()
return msg_id, result, resp.status
except Exception as e:
return msg_id, 'Error pushing data', ''
For pytest there is a async testting framework called pytest-asyncio
Given in pypi page on link there is an example code
@pytest.mark.asyncio
async def test_some_asyncio_code():
res = await library.do_something()
assert b"expected result" == res
Allowing you to create test for async strutures.
A very very in detail source for test writing with pytest-asyncio
Edit: Here is a colab link you can test the link below I have updated the test code
import pytest
import asyncio
from unittest.mock import AsyncMock, patch
from <your_original_file> import post_message # import your fucntion to test
# AsyncMock for mock sesion.post and patch to use our mock insetad of aiohttp.ClientSession()
@pytest.mark.asyncio
async def test_post_message():
# mock response from service
mock_resp = AsyncMock()
mock_resp.text.return_value = "Success"
mock_resp.status = 200
mock_resp.__aenter__.return_value = mock_resp
# post to mock and test if it works as expected
with patch("aiohttp.ClientSession.post", return_value=mock_resp):
msg_id, result, status = await post_message("http://test.com", {}, "123")
assert msg_id == "123"
assert result == "Success"
assert status == 200
# Run the test manually
await test_post_message()
print("Test passed")