proposal for #423 adding test utilities to aiohttp.

aiohttp/test_utils.py

@@ -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):
@@ -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)

docs/api.rst

@@ -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
 ------------------------
 
@@ -74,4 +83,5 @@ aiohttp.wsgi module
     :undoc-members:
     :show-inheritance:
 
+
 .. disqus::

docs/testing.rst

@@ -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())

tests/conftest.py

@@ -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:
@@ -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

tests/test_run_app.py

@@ -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)