Make mocked request helper

aiohttp/test_utils.py

@@ -18,13 +18,19 @@
 import traceback
 import urllib.parse
 import unittest
+from unittest import mock
 
 import asyncio
 import aiohttp
+from multidict import CIMultiDict
+
 from . import server
 from . import helpers
 from . import ClientSession
 from . import hdrs
+from .protocol import HttpVersion
+from .protocol import RawRequestMessage
+from .signals import Signal
 
 
 def run_briefly(loop):
@@ -522,3 +528,111 @@ def teardown_test_loop(loop):
         loop.close()
     gc.collect()
     asyncio.set_event_loop(None)
+
+
+def _create_app_mock():
+    app = mock.Mock()
+    app._debug = False
+    app.on_response_prepare = Signal(app)
+    return app
+
+
+def _create_transport(sslcontext=None):
+    transport = mock.Mock()
+
+    def get_extra_info(key):
+        if key == 'sslcontext':
+            return sslcontext
+        else:
+            return None
+
+    transport.get_extra_info.side_effect = get_extra_info
+    return transport
+
+
+_not_set = object()
+
+
+def make_mocked_request(method, path, headers=CIMultiDict(), *,
+                        version=HttpVersion(1, 1), closing=False,
+                        app=None,
+                        reader=_not_set,
+                        writer=_not_set,
+                        transport=_not_set,
+                        payload=_not_set,
+                        sslcontext=None,
+                        secure_proxy_ssl_header=None):
+    """Creates mocked web.Request testing purposes. Useful in unit tests,
+    when spinning full web server is overkill or specific conditions and
+    errors is hard to trigger.
+
+    :param method: str, that represents HTTP method, like; GET, POST.
+    :type method: str
+
+    :param path: str, The URL including *PATH INFO* without the host or scheme
+    :type path: multidict.CIMultiDict
+
+    :param headers: str, The URL including *PATH INFO* without the host or scheme
+    :type headers: str
+
+    :param version: namedtuple with encoded HTTP version
+    :type version: aiohttp.protocol.HttpVersion
+
+    :param closing: flag idicates that connection should be closed after
+        response.
+    :type closing: bool
+
+    :param app: the aiohttp.web application attached for fake request
+    :type app: aiohttp.web.Application
+
+    :param reader: object for storing and managing incoming data
+    :type reader: aiohttp.parsers.StreamParser
+
+    :param writer: object for managing outcoming data
+    :type wirter: aiohttp.parsers.StreamWriter
+
+    :param transport: asyncio transport instance
+    :type transport: asyncio.transports.Transport
+
+    :param payload: raw payload reader object
+    :type  payload: aiohttp.streams.FlowControlStreamReader
+
+    :param sslcontext: ssl.SSLContext object, for HTTPS connection
+    :type sslcontext: ssl.SSLContext
+
+    :param secure_proxy_ssl_header: A tuple representing a HTTP header/value
+        combination that signifies a request is secure.
+    :type secure_proxy_ssl_header: tuple
+    """
+
+    if version < HttpVersion(1, 1):
+        closing = True
+    message = RawRequestMessage(method, path, version, headers,
+                                [(k.encode('utf-8'), v.encode('utf-8'))
+                                 for k, v in headers.items()],
+                                closing, False)
+    if app is None:
+        app = _create_app_mock()
+
+    if reader is _not_set:
+        reader = mock.Mock()
+
+    if writer is _not_set:
+        writer = mock.Mock()
+
+    if transport is _not_set:
+        transport = _create_transport(sslcontext)
+
+    if payload is _not_set:
+        payload = mock.Mock()
+
+    from .web import Request
+    req = Request(app, message, payload,
+                  transport, reader, writer,
+                  secure_proxy_ssl_header=secure_proxy_ssl_header)
+
+    assert req.app is app
+    assert req.content is payload
+    assert req.transport is transport
+
+    return req

docs/testing.rst

@@ -143,6 +143,26 @@ functionality, the AioHTTPTestCase is provided::
 
             self.loop.run_until_complete(test_get_route())
 
+Faking request object
+---------------------
+
+aiohttp provides test utility for creating fake `web.Request` objects:
+:data:`aiohttp.test_utils.make_mocked_request`, it could be useful in case of
+simple unit tests, like handler tests, or simulate error conditions that
+hard to reproduce on real server. ::
+
+    from aiohttp import web
+
+    def handler(request):
+        assert request.headers.get('token') == 'x'
+        return web.Response(body=b'data')
+
+    def test_handler()
+        req = make_request('get', 'http://python.org/', headers={'token': 'x')
+        resp = header(req)
+        assert resp.body == b'data'
+
+
 aiohttp.test_utils
 ------------------
 

tests/test_urldispatch.py

@@ -6,20 +6,19 @@
 from collections.abc import Sized, Container, Iterable, Mapping, MutableMapping
 from unittest import mock
 from urllib.parse import unquote
-from multidict import CIMultiDict
 import aiohttp.web
 from aiohttp import hdrs, helpers
-from aiohttp.web import (UrlDispatcher, Request, Response,
+from aiohttp.web import (UrlDispatcher, Response,
                          HTTPMethodNotAllowed, HTTPNotFound,
                          HTTPCreated)
-from aiohttp.protocol import HttpVersion, RawRequestMessage
 from aiohttp.web_urldispatcher import (_defaultExpectHandler,
                                        DynamicRoute,
                                        PlainRoute,
                                        SystemRoute,
                                        ResourceRoute,
                                        AbstractResource,
                                        View)
+from aiohttp.test_utils import make_mocked_request
 
 
 class TestUrlDispatcher(unittest.TestCase):
@@ -33,16 +32,7 @@ def tearDown(self):
         self.loop.close()
 
     def make_request(self, method, path):
-        self.app = mock.Mock()
-        message = RawRequestMessage(method, path, HttpVersion(1, 1),
-                                    CIMultiDict(), [], False, False)
-        self.payload = mock.Mock()
-        self.transport = mock.Mock()
-        self.reader = mock.Mock()
-        self.writer = mock.Mock()
-        req = Request(self.app, message, self.payload,
-                      self.transport, self.reader, self.writer)
-        return req
+        return make_mocked_request(method, path)
 
     def make_handler(self):
 

tests/test_web_request.py

@@ -1,49 +1,14 @@
 import pytest
 from unittest import mock
+
 from multidict import MultiDict, CIMultiDict
-from aiohttp.signals import Signal
-from aiohttp.web import Request
 from aiohttp.protocol import HttpVersion
-from aiohttp.protocol import RawRequestMessage
+from aiohttp.test_utils import make_mocked_request
 
 
 @pytest.fixture
 def make_request():
-    def maker(method, path, headers=CIMultiDict(), *,
-              version=HttpVersion(1, 1), closing=False,
-              sslcontext=None,
-              secure_proxy_ssl_header=None):
-        if version < HttpVersion(1, 1):
-            closing = True
-        app = mock.Mock()
-        app._debug = False
-        app.on_response_prepare = Signal(app)
-        message = RawRequestMessage(method, path, version, headers,
-                                    [(k.encode('utf-8'), v.encode('utf-8'))
-                                     for k, v in headers.items()],
-                                    closing, False)
-        payload = mock.Mock()
-        transport = mock.Mock()
-
-        def get_extra_info(key):
-            if key == 'sslcontext':
-                return sslcontext
-            else:
-                return None
-
-        transport.get_extra_info.side_effect = get_extra_info
-        writer = mock.Mock()
-        reader = mock.Mock()
-        req = Request(app, message, payload,
-                      transport, reader, writer,
-                      secure_proxy_ssl_header=secure_proxy_ssl_header)
-
-        assert req.app is app
-        assert req.content is payload
-        assert req.transport is transport
-
-        return req
-    return maker
+    return make_mocked_request
 
 
 def test_ctor(make_request, warning):
@@ -66,6 +31,20 @@ def test_ctor(make_request, warning):
 
     assert req.keep_alive
 
+    # just make sure that all lines of make_mocked_request covered
+    reader = mock.Mock()
+    writer = mock.Mock()
+    payload = mock.Mock()
+    transport = mock.Mock()
+    app = mock.Mock()
+    req = make_request('GET', '/path/to?a=1&b=2', writer=writer, reader=reader,
+                       payload=payload, transport=transport, app=app)
+    assert req.app is app
+    assert req.content is payload
+    assert req.transport is transport
+    assert req._reader is reader
+    assert req._writer is writer
+
 
 def test_doubleslashes(make_request):
     req = make_request('GET', '//foo/')