Ticket #2879: django_live_server_r7936.diff

django/test/testcases.py

@@ -1 @@
-import re
-import unittest
+import re, socket, threading, unittest
 from urlparse import urlsplit, urlunsplit
 
 from django.http import QueryDict
 from django.db import transaction
 from django.conf import settings
 from django.core import mail
+from django.core.handlers.wsgi import WSGIHandler
 from django.core.management import call_command
+from django.core.servers import basehttp
 from django.test import _doctest as doctest
 from django.test.client import Client
+from django.test.utils import create_test_db
 from django.core.urlresolvers import clear_url_caches
 
 normalize_long_ints = lambda s: re.sub(r'(?<![\w])(\d+)L(?![\w])', '\\1', s)
@@ -49 +51 @@
         # side effects on other tests.
         transaction.rollback_unless_managed()
 
+class TestServerThread(threading.Thread):
+    """Thread for running a http server while tests are running."""
+
+    def __init__(self, address, port):
+        self.address = address
+        self.port = port
+        self._stopevent = threading.Event()
+        self.started = threading.Event()
+        self.error = None
+        super(TestServerThread, self).__init__()
+
+    def run(self):
+        """Sets up test server and database and loops over handling http requests."""
+        try:
+            handler = basehttp.AdminMediaHandler(WSGIHandler())
+            server_address = (self.address, self.port)
+            httpd = basehttp.StoppableWSGIServer(server_address, basehttp.WSGIRequestHandler)
+            httpd.set_app(handler)
+            self.started.set()
+        except basehttp.WSGIServerException, e:
+            self.error = e
+            self.started.set()
+            return
+
+        # Must do database stuff in this new thread if database in memory.
+        from django.conf import settings
+        if settings.DATABASE_ENGINE == 'sqlite3' \
+            and (not settings.TEST_DATABASE_NAME or settings.TEST_DATABASE_NAME == ':memory:'):
+            db_name = create_test_db(0)
+            # Import the fixture data into the test database.
+            if hasattr(self, 'fixtures'):
+                # We have to use this slightly awkward syntax due to the fact
+                # that we're using *args and **kwargs together.
+                call_command('loaddata', *self.fixtures, **{'verbosity': 0})
+
+        # Loop until we get a stop event.
+        while not self._stopevent.isSet():
+            httpd.handle_request()
+
+    def join(self, timeout=None):
+        """Stop the thread and wait for it to finish."""
+        self._stopevent.set()
+        threading.Thread.join(self, timeout)
+
 class TestCase(unittest.TestCase):
     def _pre_setup(self):
         """Performs any pre-test setup. This includes:
@@ -105 +151 @@
             settings.ROOT_URLCONF = self._old_root_urlconf
             clear_url_caches()
 
+    def start_test_server(self, address='localhost', port=8000):
+        """Creates a live test server object (instance of WSGIServer)."""
+        self.server_thread = TestServerThread(address, port)
+        self.server_thread.start()
+        self.server_thread.started.wait()
+        if self.server_thread.error:
+            raise self.server_thread.error
+
+    def stop_test_server(self):
+        if self.server_thread:
+            self.server_thread.join()
+
     def assertRedirects(self, response, expected_url, status_code=302,
                         target_status_code=200, host=None):
         """Asserts that a response redirected to a specific URL, and that the

django/core/servers/basehttp.py

@@ -11 +11 @@
 import mimetypes
 import os
 import re
+import socket
 import sys
 import urllib
 
@@ -658 +659 @@
         start_response(status, headers.items())
         return output
 
+class StoppableWSGIServer(WSGIServer):
+    """WSGIServer with short timeout, so that server thread can stop this server."""
+
+    def server_bind(self):
+        """Sets timeout to 1 second."""
+        WSGIServer.server_bind(self)
+        self.socket.settimeout(1)
+
+    def get_request(self):
+        """Checks for timeout when getting request."""
+        try:
+            sock, address = self.socket.accept()
+            sock.settimeout(None)
+            return (sock, address)
+        except socket.timeout:
+            raise
+
 def run(addr, port, wsgi_handler):
     server_address = (addr, port)
     httpd = WSGIServer(server_address, WSGIRequestHandler)

tests/regressiontests/test_client_regress/models.py

@@ -7 +7 @@
 from django.core.exceptions import SuspiciousOperation
 import os
 import sha
+import urllib
 
 class AssertContainsTests(TestCase):
     def test_contains(self):
@@ -327 +328 @@
         "URLconf is reverted to original value after modification in a TestCase"
         url = reverse('arg_view', args=['somename'])
         self.assertEquals(url, '/test_client_regress/arg_view/somename/')
+
+class TestServerTests(TestCase):
+
+    def setUp(self):
+        self.start_test_server(address='localhost', port=8001)
+
+    def tearDown(self):
+        self.stop_test_server()
+
+    def test_server_up(self):
+        url = urllib.urlopen('http://localhost:8001')
+        self.assertEqual(url.read(), 'Django Internal Tests: 404 Error')
+        url.close()
+
+    def test_serve_page(self):
+        url = urllib.urlopen('http://localhost:8001/accounts/login')
+        # Just make sure this isn't a 404, and we've gotten something.
+        self.assertNotEqual(url.read(), 'Django Internal Tests: 404 Error')
+        url.close()

AUTHORS

@@ -273 +273 @@
     Robert Myers <myer0052@gmail.com>
     Nebojša Dorđević
     Doug Napoleone <doug@dougma.com>
+    Devin Naquin <dnaquin@gmail.com>
     Gopal Narayanan <gopastro@gmail.com>
     Fraser Nevett <mail@nevett.org>
     Sam Newman <http://www.magpiebrain.com/>

docs/testing.txt

@@ -830 +830 @@
 This test case will use the contents of ``myapp.test_urls`` as the
 URLconf for the duration of the test case.
 
+Running tests with a live test server
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+**New in Django development version**
+
+When running tests that use in-browser frameworks such as Twille_ and
+Selenium_, it's necessary to have a running test server. Django's custom
+``TestCase`` class supports starting a live server for these purposes.
+
+``start_test_server(address='localhost', port=8000)``
+    Starts a test server at ``address`` on ``port``. This should be done in the
+    ``setUp()`` method of a subclass of ``TestCase``. The server then can be
+    accessed at http://address:port.
+
+``stop_test_server()``
+    Stops the test server that was started with ``start_test_server``. This
+    must be done before ``start_test_server`` can be called again, so this
+    should be done in the ``tearDown()`` method of a subclass of ``TestCase``.
+
+This can be used to start a server that can then be accessed by Twill, Selenium
+or another in-browser test framework. For example::
+
+        from django.test.testcases import TestCase
+        from selenium import selenium
+        
+        class TestLogin(TestCase):
+            fixtures = ['login_info.xml']
+
+            def setUp(self):
+                # Start a test server and tell selenium where to find it.
+                self.start_test_server('localhost', 8000)
+                self.selenium = selenium('localhost', 4444, \
+                        '*pifirefox', 'http://localhost:8000')
+
+            def tearDown(self):
+                # Stop server and Selenium
+                self.selenium.stop()
+                self.stop_test_server()
+
+            def testLogin(self):
+                self.selenium.open('/admin/')
+                self.selenium.type('username', 'admin')
+                self.selenium.type('password', 'password')
+                self.selenium.click("//input[@value='Log in']")
+
+.. _Twill: http://twill.idyll.org/
+.. _Selenium: http://www.openqa.org/selenium/
+
 Emptying the test outbox
 ~~~~~~~~~~~~~~~~~~~~~~~~