===============
testing-aiohttp
.. image:: https://travis-ci.org/genericclient/testing-aiohttp.svg?branch=master
:target: https://travis-ci.org/genericclient/testing-aiohttp
Testing utilities for aiohttp
. Python 3.5+ only.
Installation
::
$ pip install testing-aiohttp
Usage
RouteManager
RouteManager
will mock up responses for `aiohttp.Client``.
The API is inspired by the responses
library::
from aiohttp import ClientSession
from asynctest import TestCase
from testing_aiohttp import RouteManager
# Create your tests here.
class MyTestCase(TestCase):
async def test_response_data(self):
with RouteManager() as rsps:
rsps.add('GET', 'http://example.org/users', json=[
{
'id': 1,
'username': 'user1',
'group': 'watchers',
},
{
'id': 2,
'username': 'user2',
'group': 'watchers',
},
])
async with ClientSession() as session:
response = await session.get('http://example.org/users')
self.assertEqual(response, 200)
users = await response.json()
self.assertEqual(len(users), 2)
::
from aiohttp import ClientSession
from asynctest import TestCase
from testing_aiohttp import RouteManager
async def request_callback(request):
return (200, {}, 'ok')
class MyTestCase(TestCase):
async def test_endpoint_detail_route(self):
with RouteManager() as rsps:
rsps.add_callback(
'POST', 'http://example.org/users/2/notify',
callback=request_callback,
content_type='application/json',
)
async with ClientSession() as session:
response = await session.post('http://example.org/users/2/notify')
self.assertEqual(await response.text(), 'ok')
::
from aiohttp import ClientSession
from asynctest import TestCase
from testing_aiohttp.rsps import RouteManager, RouteNotFoundError
class MyTestCase(TestCase):
async def test_response_match_querystring(self):
with RouteManager() as rsps:
rsps.add('GET', 'http://example.org/users?username=user1', json=[
{
'id': 1,
'username': 'user1',
'group': 'watchers',
},
], match_querystring=True)
with ClientSession() as session:
response = await session.get('http://example.org/users', params={'username': 'user1'})
self.assertEqual(response.status, 200)
users = await response.json()
self.assertEqual(len(users), 1)
with self.assertRaises(RouteNotFoundError):
with RouteManager() as rsps:
rsps.add('GET', 'http://example.org/users?username=user1', json=[
{
'id': 1,
'username': 'user1',
'group': 'watchers',
},
], match_querystring=True)
with ClientSession() as session:
await session.get('http://example.org/users')
License
Licensed under the MIT License.