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.