Skip to content

Commit

Permalink
Merge pull request #902 from toumorokoshi/master
Browse files Browse the repository at this point in the history
proposal for #423 adding test utilities to aiohttp.
  • Loading branch information
asvetlov committed Jun 3, 2016
2 parents c059e0b + 035fbde commit 31f0ef9
Show file tree
Hide file tree
Showing 6 changed files with 408 additions and 29 deletions.
145 changes: 145 additions & 0 deletions aiohttp/test_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,22 +4,27 @@
import contextlib
import gc
import email.parser
import functools
import http.server
import json
import logging
import io
import os
import re
import socket
import ssl
import sys
import threading
import traceback
import urllib.parse
import unittest

import asyncio
import aiohttp
from aiohttp import server
from aiohttp import helpers
from aiohttp import ClientSession
from aiohttp.client import _RequestContextManager


def run_briefly(loop):
Expand Down Expand Up @@ -304,3 +309,143 @@ def _response(self, response, body=None,
# keep-alive
if response.keep_alive():
self._srv.keep_alive(True)


def unused_port():
""" return a port that is unused on the current host. """
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(('127.0.0.1', 0))
return s.getsockname()[1]


class TestClient:
"""
A test client implementation, for a aiohttp.web.Application.
:param app: the aiohttp.web application passed
to create_test_server
:type app: aiohttp.web.Application
:param protocol: the aiohttp.web application passed
to create_test_server
:type app: aiohttp.web.Application
"""

def __init__(self, app, protocol="http"):
self._app = app
self._loop = loop = app.loop
self.port = unused_port()
self._handler = handler = app.make_handler()
self._server = loop.run_until_complete(loop.create_server(
handler, '127.0.0.1', self.port
))
self._session = ClientSession(loop=self._loop)
self._root = "{}://127.0.0.1:{}".format(
protocol, self.port
)
self._closed = False

def request(self, method, url, *args, **kwargs):
return _RequestContextManager(self._request(
method, url, *args, **kwargs
))

@asyncio.coroutine
def _request(self, method, url, *args, **kwargs):
""" routes a request to the http server.
the interface is identical to asyncio.request,
except the loop kwarg is overriden
by the instance used by the application.
"""
return (yield from self._session.request(
method, self._root + url, *args, **kwargs
))

def close(self):
if not self._closed:
loop = self._loop
loop.run_until_complete(self._session.close())
loop.run_until_complete(self._handler.finish_connections())
loop.run_until_complete(self._app.finish())
self._server.close()
loop.run_until_complete(self._server.wait_closed())
self._closed = True

def __del__(self):
self.close()

def __enter__(self):
return self

def __exit__(self, exc_type, exc_value, traceback):
self.close()


class AioHTTPTestCase(unittest.TestCase):

def get_app(self, loop):
"""
this method should be overriden
to return the aiohttp.web.Application
object to test.
:param loop: the event_loop to use
:type loop: asyncio.BaseEventLoop
"""
pass

def setUp(self):
self.loop = setup_test_loop()
self.app = self.get_app(self.loop)
self.client = TestClient(self.app)

def tearDown(self):
del self.client
teardown_test_loop(self.loop)


def run_loop(func):
"""
to be used with AioHTTPTestCase. Handles
executing an asynchronous function, using
the event loop of AioHTTPTestCase.
"""

@functools.wraps(func)
def new_func(self):
return self.loop.run_until_complete(func(self))

return new_func


@contextlib.contextmanager
def loop_context():
"""
create an event_loop, for test purposes.
handles the creation and cleanup of a test loop.
"""
loop = setup_test_loop()
yield loop
teardown_test_loop(loop)


def setup_test_loop():
loop = asyncio.new_event_loop()
asyncio.set_event_loop(None)
return loop


def teardown_test_loop(loop):
is_closed = getattr(loop, 'is_closed')
if is_closed is not None:
closed = is_closed()
else:
closed = loop._closed
if not closed:
loop.call_soon(loop.stop)
loop.run_forever()
loop.close()
gc.collect()
asyncio.set_event_loop(None)
10 changes: 10 additions & 0 deletions docs/api.rst
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,15 @@ aiohttp.streams module
:undoc-members:
:show-inheritance:


aiohttp.test_utils module
-------------------------

.. automodule:: aiohttp.test_utils
:members:
:undoc-members:
:show-inheritance:

aiohttp.websocket module
------------------------

Expand All @@ -74,4 +83,5 @@ aiohttp.wsgi module
:undoc-members:
:show-inheritance:


.. disqus::
126 changes: 126 additions & 0 deletions docs/testing.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
.. _aiohttp-testing:

Testing
=======

Testing aiohttp servers
-----------------------

aiohttp provides test framework agnostic utilities for web
servers. An example would be::

from aiohttp.test_utils import TestClient, loop_context
from aiohttp import request

# loop_context is provided as a utility. You can use any
# asyncio.BaseEventLoop class in it's place.
with loop_context() as loop:
app = _create_example_app(loop)
with TestClient(app) as client:

async def test_get_route():
nonlocal client
resp = await client.request("GET", "/")
assert resp.status == 200
text = await resp.text()
assert "Hello, world" in text

loop.run_until_complete(test_get_route())


If it's preferred to handle the creation / teardown on a more granular
basis, the TestClient object can be used directly::

from aiohttp.test_utils import TestClient

with loop_context() as loop:
app = _create_example_app(loop)
client = TestClient(app)
root = "http://127.0.0.1:{}".format(port)

async def test_get_route():
resp = await test_client.request("GET", url)
assert resp.status == 200
text = await resp.text()
assert "Hello, world" in text

loop.run_until_complete(test_get_route())
# the server is cleaned up implicitly, through
# the deletion of the TestServer.
del client

pytest example
==============

A pytest example could look like::

# assuming you are using pytest-asyncio
from asyncio.test_utils import TestClient, loop_context

@pytest.yield_fixture
def loop():
with loop_context() as loop:
yield loop

@pytest.fixture
def app(test_loop):
return create_app(event_loop)


@pytest.yield_fixture
def test_client(app):
server = TestClient(app)
yield client
client.close()

def test_get_route(loop, test_client):
@asyncio.coroutine
def test_get_route():
nonlocal test_client
resp = yield from test_client.request("GET", "/")
assert resp.status == 200
text = yield from resp.text()
assert "Hello, world" in text

loop.run_until_complete(test_get_route())


unittest example
================

To test applications with the standard library's unittest or unittest-based
functionality, the AioHTTPTestCase is provided::

from aiohttp.test_utils import AioHTTPTestCase, run_loop
from aiohttp import web

class MyAppTestCase(AioHTTPTestCase):

def get_app(self, loop):
"""
override the get_app method to return
your application.
"""
# it's important to use the loop passed here.
return web.Application(loop=loop)

# the run_loop decorator can be used in tandem with
# the AioHTTPTestCase to simplify running
# tests that are asynchronous
@run_loop
async def test_example(self):
request = await self.client.request("GET", "/")
assert request.status == 200
text = await request.text()
assert "Hello, world" in text

# a vanilla example
def test_example(self):
async def test_get_route():
url = root + "/"
resp = await self.client.request("GET", url, loop=loop)
assert resp.status == 200
text = await resp.text()
assert "Hello, world" in text

self.loop.run_until_complete(test_get_route())
37 changes: 9 additions & 28 deletions tests/conftest.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,15 @@
import asyncio
import aiohttp
import collections
import gc
import logging
import pytest
import re
import socket
import sys
import warnings

from aiohttp import web
from aiohttp.test_utils import (
loop_context, unused_port
)


class _AssertWarnsContext:
Expand Down Expand Up @@ -147,37 +147,18 @@ def log():
yield _AssertLogsContext


@pytest.fixture
def unused_port():
def f():
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(('127.0.0.1', 0))
return s.getsockname()[1]
return f
# add the unused_port and loop fixtures
pytest.fixture(unused_port)


@pytest.yield_fixture
def loop(request):
loop = asyncio.new_event_loop()
asyncio.set_event_loop(None)

yield loop

is_closed = getattr(loop, 'is_closed')
if is_closed is not None:
closed = is_closed()
else:
closed = loop._closed
if not closed:
loop.call_soon(loop.stop)
loop.run_forever()
loop.close()
gc.collect()
asyncio.set_event_loop(None)
def loop():
with loop_context() as loop:
yield loop


@pytest.yield_fixture
def create_server(loop, unused_port):
def create_server(loop):
app = handler = srv = None

@asyncio.coroutine
Expand Down
2 changes: 1 addition & 1 deletion tests/test_run_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ def test_run_app_https(loop):


def test_run_app_nondefault_host_port(loop, unused_port):
port = unused_port()
port = unused_port
host = 'localhost'

loop = mock.Mock(spec=asyncio.AbstractEventLoop, wrap=loop)
Expand Down
Loading

0 comments on commit 31f0ef9

Please sign in to comment.