Ticket #12091: wsgi-apps.diff

django/test/test_wsgi_views.py

@@ +1 @@
+"""
+Tests for the use of WSGI applications within Django.
+
+"""
+from re import compile as compile_regex
+from unittest import TestCase
+
+from django.views.wsgi import call_wsgi_app, make_wsgi_view, _ResponseStarter
+from django.core.handlers.wsgi import WSGIRequest
+
+
+class TestResponseStarter(TestCase):
+    """Tests for the internal _ResponseStarter."""
+    
+    def test_constructor(self):
+        """The constructor should set all the attributes correctly."""
+        start_response = _ResponseStarter()
+        self.assertEqual(None, start_response.status)
+        self.assertEqual([], start_response.response_headers)
+        self.assertEqual(None, start_response.exc_info)
+    
+    def test_call(self):
+        start_response = _ResponseStarter()
+        status = "200 Everything's alright"
+        headers = (
+            ("X-FOO", "ABC"),
+            ("X-BAR", "XYZ"),
+            )
+        exc_info = object()
+        # Starting the mock response:
+        start_response(status, headers, exc_info)
+        # Checking whether they were all set correctly:
+        self.assertEqual(start_response.status, 200)
+        self.assertEqual(start_response.response_headers, headers)
+        self.assertEqual(start_response.exc_info, exc_info)
+
+
+class TestCallWSGIApp(TestCase):
+    """
+    Tests for call_wsgi_app()
+    
+    """
+    
+    def test_original_environ_not_modified(self):
+        """The original environ must have not been modified."""
+        original_environ = complete_environ(SCRIPT_NAME="/blog",
+                                            PATH_INFO="/admin/models")
+        request = make_request(**original_environ)
+        expected_environ = original_environ.copy()
+        # Running the app:
+        app = MockApp("200 OK", [])
+        call_wsgi_app(app, request, "/admin")
+        self.assertEqual(request.environ, expected_environ)
+    
+    def test_routing_args_are_removed(self):
+        """The ``wsgiorg.routing_args`` environment key must be removed."""
+        environ = {
+            'wsgiorg.routing_args': ((), {}),
+            'PATH_INFO': "/admin/models",
+            }
+        environ = complete_environ(**environ)
+        request = make_request(**environ)
+        # Running the app:
+        app = MockApp("200 OK", [])
+        call_wsgi_app(app, request, "/admin")
+        self.assertTrue("wsgiorg.routing_args" not in app.environ)
+    
+    def test_mount_point_as_string(self):
+        environ = complete_environ(SCRIPT_NAME="/dev", PATH_INFO="/trac/wiki")
+        request = make_request(**environ)
+        # Running the app:
+        app = MockApp("200 OK", [])
+        call_wsgi_app(app, request, "/trac")
+        self.assertEqual(app.environ['SCRIPT_NAME'], "/dev/trac")
+        self.assertEqual(app.environ['PATH_INFO'], "/wiki")
+        
+    def test_mount_point_as_regex(self):
+        environ = complete_environ(SCRIPT_NAME="/dev",
+                                   PATH_INFO="/project5/trac/wiki")
+        request = make_request(**environ)
+        mount_point = compile_regex(r"^(?P<project>\w+)/trac")
+        # Running the app:
+        app = MockApp("200 OK", [])
+        call_wsgi_app(app, request, mount_point)
+        self.assertEqual(app.environ['SCRIPT_NAME'], "/dev/project5/trac")
+        self.assertEqual(app.environ['PATH_INFO'], "/wiki")
+        
+    def test_implied_mount_point(self):
+        """The mount point must be implied by the view's path."""
+        environ = complete_environ(SCRIPT_NAME="/dev",
+                                   PATH_INFO="/project5/trac/wiki")
+        request = make_request(**environ)
+        # Forging the matched URL:
+        request.matched_url_regex = compile_regex(r"^(?P<project>\w+)/trac")
+        # Running the app:
+        app = MockApp("200 OK", [])
+        call_wsgi_app(app, request)
+        self.assertEqual(app.environ['SCRIPT_NAME'], "/dev/project5/trac")
+        self.assertEqual(app.environ['PATH_INFO'], "/wiki")
+    
+    def test_incorrect_mount_point(self):
+        environ = complete_environ(SCRIPT_NAME="/dev",
+                                   PATH_INFO="/trac/wiki")
+        request = make_request(**environ)
+        mount_point_string = "/bugzilla"
+        mount_point_regex = compile_regex(r"/bugzilla")
+        # Running the app:
+        app = MockApp("200 OK", [])
+        self.assertRaises(ValueError, call_wsgi_app, app, request,
+                          mount_point_string)
+        self.assertRaises(ValueError, call_wsgi_app, app, request,
+                          mount_point_regex)
+    
+    def test_http_status_code(self):
+        environ = complete_environ(SCRIPT_NAME="/dev", PATH_INFO="/trac/wiki")
+        request = make_request(**environ)
+        # Running the app and make a valid request:
+        app_ok = MockApp("200 OK", [])
+        django_response_ok = call_wsgi_app(app_ok, request, "/trac")
+        self.assertEqual(200, django_response_ok.status_code)
+        # Running the app and make an invalid request:
+        app_bad = MockApp("403 What are you trying to do?", [])
+        django_response_bad = call_wsgi_app(app_bad, request, "/trac")
+        self.assertEqual(403, django_response_bad.status_code)
+    
+    def test_headers_are_copied_over(self):
+        environ = complete_environ(SCRIPT_NAME="/dev", PATH_INFO="/trac/wiki")
+        request = make_request(**environ)
+        headers = [
+            ("X-Foo", "bar"),
+            ("Content-Type", "text/plain"),
+            ]
+        # The same headers, but set in the format used by HttpResponse
+        expected_headers = {
+            'x-foo': ("X-Foo", "bar"),
+            'content-type': ("Content-Type", "text/plain"),
+            }
+        # Running the app:
+        app = MockApp("200 OK", headers)
+        django_response = call_wsgi_app(app, request, "/trac")
+        self.assertEqual(expected_headers, django_response._headers)
+    
+    def test_authenticated_user(self):
+        environ = complete_environ(SCRIPT_NAME="/dev", PATH_INFO="/trac/wiki")
+        request = make_request(authenticated=True, **environ)
+        # Running the app:
+        app = MockApp("200 OK", [])
+        call_wsgi_app(app, request, "/trac")
+        self.assertEqual("foobar", app.environ['REMOTE_USER'])
+    
+    def test_cookies_sent(self):
+        environ = complete_environ(SCRIPT_NAME="/dev", PATH_INFO="/trac/wiki")
+        request = make_request(**environ)
+        headers = [
+            ("Set-Cookie", "arg1=val1"),
+            ("Set-Cookie", "arg2=val2; expires=Fri,%2031-Dec-2010%2023:59:59%20GMT"),
+            ("Set-Cookie", "arg3=val3; path=/"),
+            ("Set-Cookie", "arg4=val4; path=/wiki"),
+            ("Set-Cookie", "arg5=val5; domain=.example.org"),
+            ("Set-Cookie", "arg6=val6; max-age=3600"),
+            ("Set-Cookie", "arg7=val7; expires=Fri,%2031-Dec-2010%2023:59:59%20GMT; max-age=3600; domain=.example.org; path=/wiki"),
+            # Now let's try an Unicode cookie:
+            ("Set-Cookie", u"arg8=val8; max-age=3600"),
+            # TODO: The "secure" cookie *attribute* is broken in SimpleCookie.
+            # See: http://bugs.python.org/issue1028088
+            #("Set-Cookie", "arg9=val9; secure"),
+            ]
+        expected_cookies = {
+            'arg1': {'value': "val1"},
+            'arg2': {'value': "val2", 'expires': "Fri,%2031-Dec-2010%2023:59:59%20GMT"},
+            'arg3': {'value': "val3", 'path': "/"},
+            'arg4': {'value': "val4", 'path': "/wiki"},
+            'arg5': {'value': "val5", 'domain': ".example.org"},
+            'arg6': {'value': "val6", 'max-age': "3600"},
+            'arg7': {
+                'value': "val7",
+                'expires': "Fri,%2031-Dec-2010%2023:59:59%20GMT",
+                'path': "/wiki",
+                'domain': ".example.org",
+                'max-age': "3600",
+                },
+            'arg8': {'value': "val8", 'max-age': "3600"},
+            # Why the next item as disabled? Check the `headers` variable above
+            #'arg9': {'value': "val9", 'secure': True},
+            }
+        # Running the app:
+        app = MockApp("200 OK", headers)
+        django_response = call_wsgi_app(app, request, "/trac")
+        # Checking the cookies:
+        self.assertEqual(len(expected_cookies), len(django_response.cookies))
+        # Finally, let's check each cookie:
+        for (cookie_set_name, cookie_set) in django_response.cookies.items():
+            expected_cookie = expected_cookies[cookie_set_name]
+            expected_cookie_value = expected_cookie.pop("value")
+            self.assertEqual(expected_cookie_value, cookie_set.value,
+                             'Cookie "%s" has a wrong value ("%s")' %
+                             (cookie_set_name, cookie_set.value))
+            for (attr_key, attr_val) in expected_cookie.items():
+                self.assertEqual(cookie_set[attr_key], attr_val,
+                                 'Attribute "%s" in cookie "%s" is wrong (%s)' %
+                                 (attr_key, cookie_set_name, cookie_set[attr_key]))
+    
+    def test_string_as_response(self):
+        app = MockApp("200 It is OK", [("X-HEADER", "Foo")])
+        django_view = make_wsgi_view(app, "/blog")
+        # Running a request:
+        environ = complete_environ(SCRIPT_NAME="/dev", PATH_INFO="/blog/posts")
+        request = make_request(**environ)
+        # Checking the response:
+        django_response = django_view(request)
+        http_response = (
+            "X-HEADER: Foo\n"
+            "Content-Type: text/html; charset=utf-8\n"
+            "\n"
+            "body"
+            )
+        self.assertEqual(http_response, str(django_response))
+    
+    def test_iterable_as_response(self):
+        app = MockGeneratorApp("200 It is OK", [("X-HEADER", "Foo")])
+        django_view = make_wsgi_view(app, "/blog")
+        # Running a request:
+        environ = complete_environ(SCRIPT_NAME="/dev", PATH_INFO="/blog/posts")
+        request = make_request(**environ)
+        # Checking the response:
+        django_response = django_view(request)
+        self.assertFalse(django_response._is_string)
+        self.assertTrue(django_response.has_header("X-HEADER"))
+        http_response = (
+            "X-HEADER: Foo\n"
+            "Content-Type: text/html; charset=utf-8\n"
+            "\n"
+            "body as iterable"
+            )
+        self.assertEqual(http_response, str(django_response))
+    
+    def test_write_response(self):
+        app = MockWriteApp("200 It is OK", [("X-HEADER", "Foo")])
+        django_view = make_wsgi_view(app, "/blog")
+        # Running a request:
+        environ = complete_environ(SCRIPT_NAME="/dev", PATH_INFO="/blog/posts")
+        request = make_request(**environ)
+        # Checking the response:
+        django_response = django_view(request)
+        self.assertFalse(django_response._is_string)
+        self.assertTrue(django_response.has_header("X-HEADER"))
+        http_response = (
+            "X-HEADER: Foo\n"
+            "Content-Type: text/html; charset=utf-8\n"
+            "\n"
+            "body as iterable"
+            )
+        self.assertEqual(http_response, str(django_response))
+
+
+class TestWSGIView(TestCase):
+    """
+    Tests for make_wsgi_view().
+    
+    """
+    
+    def test_it(self):
+        # Loading a WSGI-powered Django view:
+        headers = [("X-SALUTATION", "Hey")]
+        app = MockApp("206 One step at a time", headers)
+        django_view = make_wsgi_view(app, "/blog")
+        # Running a request:
+        environ = complete_environ(SCRIPT_NAME="/dev", PATH_INFO="/blog/posts")
+        request = make_request(**environ)
+        # Checking the response:
+        django_response = django_view(request)
+        self.assertEqual(django_response.status_code, 206)
+        self.assertTrue(("X-SALUTATION", "Hey") ==
+                        django_response._headers['x-salutation'])
+
+
+#{ Test utilities
+
+
+class MockApp(object):
+    """
+    Mock WSGI application.
+    
+    """
+    
+    def __init__(self, status, headers):
+        self.status = status
+        self.headers = headers
+
+    def __call__(self, environ, start_response):
+        self.environ = environ
+        start_response(self.status, self.headers)
+        return ["body"]
+
+
+class MockGeneratorApp(MockApp):
+    """
+    Mock WSGI application that returns an iterator.
+    
+    """
+
+    def __call__(self, environ, start_response):
+        self.environ = environ
+        start_response(self.status, self.headers)
+        def gen():
+            yield "body"
+            yield " as"
+            yield " iterable"
+        return gen()
+
+
+class MockWriteApp(MockApp):
+    """
+    Mock WSGI app which uses the write() function.
+    
+    """
+
+    def __call__(self, environ, start_response):
+        self.environ = environ
+        write = start_response(self.status, self.headers)
+        write( "body")
+        write(" as")
+        write(" iterable")
+        return []
+
+
+def make_request(authenticated=False, **environ):
+    """
+    Make a Django request from the items in the WSGI ``environ``.
+    
+    """
+    class MockDjangoUser(object):
+        def __init__(self, authenticated):
+            self.username = "foobar"
+            self.authenticated = authenticated
+        def is_authenticated(self):
+            return self.authenticated
+    request = WSGIRequest(environ)
+    request.user = MockDjangoUser(authenticated)
+    return request
+
+
+def complete_environ(**environ):
+    """
+    Add the missing items in ``environ``.
+    
+    """
+    full_environ = {
+        'REQUEST_METHOD': "GET",
+        'SERVER_NAME': "example.org",
+        }
+    full_environ.update(environ)
+    return full_environ
+
+
+#}

django/core/handlers/base.py

@@ -78 +78 @@
         urlconf = getattr(request, "urlconf", settings.ROOT_URLCONF)
 
         resolver = urlresolvers.RegexURLResolver(r'^/', urlconf)
+        # Let's include the matched URL pattern in the request, which would be
+        # useful when dealing with WSGI applications within Django:
+        request.matched_url_regex = resolver.regex
         try:
             callback, callback_args, callback_kwargs = resolver.resolve(
                     request.path_info)

django/views/wsgi.py

@@ +1 @@
+"""
+Utilities to use WSGI applications within Django.
+
+"""
+
+from Cookie import SimpleCookie
+from itertools import chain
+
+from django.http import HttpResponse
+
+
+__all__ = ("call_wsgi_app", "make_wsgi_view")
+
+
+def call_wsgi_app(wsgi_app, request, mount_point=None):
+    """
+    Call the ``wsgi_app`` with ``request`` and return its response.
+    
+    :param wsgi_app: The WSGI application to be run.
+    :type wsgi_app: callable
+    :param request: The Django request.
+    :type request: :class:`django.http.HttpRequest`
+    :param mount_point: The path where the WSGI application should be mounted.
+    :type mount_point: regex pattern or :class:`basestring`
+    :return: The response from the WSGI application, turned into a Django
+        response.
+    :rtype: :class:`django.http.HttpResponse`
+    
+    If ``mount_point`` is not present, the URL matched for the current request
+    in Django will be used -- This is the desired behavior is most situations.
+    
+    """
+    environ = request.environ.copy()
+    
+    # Moving the portion of the path consumed by the current view, from the
+    # PATH_INTO to the SCRIPT_NAME:
+    final_mount_point = mount_point or request.matched_url_regex
+    if isinstance(final_mount_point, basestring):
+        # It's already an string, so we just have to make sure it's valid:
+        if not environ['PATH_INFO'].startswith(final_mount_point):
+            raise ValueError("Path %s has not been consumed in PATH_INFO" %
+                             final_mount_point)
+    else:
+        # It's a regular expression:
+        match = final_mount_point.search(environ['PATH_INFO'][1:])
+        if not match:
+            regex = final_mount_point.pattern
+            raise ValueError("Path pattern %s has not been consumed in "
+                             "PATH_INFO" % regex)
+        final_mount_point = "/%s" % match.group()
+    environ['PATH_INFO'] = environ['PATH_INFO'][len(final_mount_point):]
+    environ['SCRIPT_NAME'] = environ['SCRIPT_NAME'] + final_mount_point
+    
+    # If the user has been authenticated in Django, log him in the WSGI app:
+    if request.user.is_authenticated():
+        environ['REMOTE_USER'] = request.user.username
+    
+    # Cleaning the routing_args, if any. The application should have its own
+    # arguments, without relying on any arguments from a parent application:
+    if "wsgiorg.routing_args" in environ:
+        del environ['wsgiorg.routing_args']
+    
+    # Calling the WSGI application and getting its response:
+    response_wrapper = _ResponseStarter()
+    wsgi_response = wsgi_app(environ, response_wrapper)
+    body = chain(response_wrapper.body, wsgi_response)
+    
+    # Turning its response into a Django response:
+    cookies = SimpleCookie()
+    django_response = HttpResponse(body, status=response_wrapper.status)
+    for (header, value) in response_wrapper.response_headers:
+        if header.upper() == "SET-COOKIE":
+            if isinstance(value, unicode):
+                # It can't be Unicode:
+                value = value.encode("us-ascii")
+            cookies.load(value)
+        else:
+            django_response[header] = value
+    
+    # Setting the cookies from Django:
+    for (cookie_name, cookie) in cookies.items():
+        cookie_attributes = {
+            'key': cookie_name,
+            'value': cookie.value,
+            'max_age': cookie.get("max-age"),
+            'expires': cookie.get("expires"),
+            'path': cookie.get("path", "/"),
+            'domain': cookie.get("domain"),
+            }
+        django_response.set_cookie(**cookie_attributes)
+    return django_response
+
+
+def make_wsgi_view(wsgi_app, mount_point=None):
+    """
+    Return a callable which can be used as a Django view powered by the
+    ``wsgi_app``.
+    
+    :param wsgi_app: The WSGI which will run the view.
+    :return: The callable.
+    
+    """
+    def view(request):
+        return call_wsgi_app(wsgi_app, request, mount_point)
+    return view
+
+
+#{ Internal WSGI stuff
+
+
+class _ResponseStarter(object):
+    """
+    Callable to be used as ``start_response`` in order to extract the HTTP
+    status code and headers.
+    
+    """
+    
+    def __init__(self):
+        self.status = None
+        self.response_headers = []
+        self.exc_info = None
+        self.body = []
+    
+    def __call__(self, status, response_headers, exc_info=None):
+        self.status = int(status[:3])
+        self.response_headers = response_headers
+        # exc_info is not used at all. It does not seem to be possible to use
+        # it in Django.
+        self.exc_info = exc_info
+        
+        def write(data):
+            self.body.append(data)
+        
+        return write
+
+
+#}