root/trunk/windmill/authoring/djangotest.py

Revision 1010, 3.6 kB (checked in by mikeal, 8 weeks ago)

Allowing django test server port to be configuration. Fix for 142. Thanks sk1p, sorry it took so long to get this one in, I could have sworn I did it weeks ago :(

Line 
1# Code from django_live_server_r8458.diff @  http://code.djangoproject.com/ticket/2879#comment:41
2# Editing to monkey patch django rather than be in trunk
3
4import socket
5import threading
6from django.core.handlers.wsgi import WSGIHandler
7from django.core.servers import basehttp
8from django.test.testcases import call_command, TestCase
9
10class StoppableWSGIServer(basehttp.WSGIServer):
11    """WSGIServer with short timeout, so that server thread can stop this server."""
12
13    def server_bind(self):
14        """Sets timeout to 1 second."""
15        basehttp.WSGIServer.server_bind(self)
16        self.socket.settimeout(1)
17
18    def get_request(self):
19        """Checks for timeout when getting request."""
20        try:
21            sock, address = self.socket.accept()
22            sock.settimeout(None)
23            return (sock, address)
24        except socket.timeout:
25            raise
26
27class TestServerThread(threading.Thread):
28    """Thread for running a http server while tests are running."""
29
30    def __init__(self, address, port):
31        self.address = address
32        self.port = port
33        self._stopevent = threading.Event()
34        self.started = threading.Event()
35        self.error = None
36        super(TestServerThread, self).__init__()
37
38    def run(self):
39        """Sets up test server and database and loops over handling http requests."""
40        try:
41            handler = basehttp.AdminMediaHandler(WSGIHandler())
42            server_address = (self.address, self.port)
43            httpd = StoppableWSGIServer(server_address, basehttp.WSGIRequestHandler)
44            httpd.set_app(handler)
45            self.started.set()
46        except basehttp.WSGIServerException, e:
47            self.error = e
48            self.started.set()
49            return
50
51        # Must do database stuff in this new thread if database in memory.
52        from django.conf import settings
53        if settings.DATABASE_ENGINE == 'sqlite3' \
54            and (not settings.TEST_DATABASE_NAME or settings.TEST_DATABASE_NAME == ':memory:'):
55            from django.db import connection
56            db_name = connection.creation.create_test_db(0)
57            # Import the fixture data into the test database.
58            if hasattr(self, 'fixtures'):
59                # We have to use this slightly awkward syntax due to the fact
60                # that we're using *args and **kwargs together.
61                call_command('loaddata', *self.fixtures, **{'verbosity': 0})
62
63        # Loop until we get a stop event.
64        while not self._stopevent.isSet():
65            httpd.handle_request()
66
67    def join(self, timeout=None):
68        """Stop the thread and wait for it to finish."""
69        self._stopevent.set()
70        threading.Thread.join(self, timeout)
71
72
73def start_test_server(self, address='localhost', port=8000):
74    """Creates a live test server object (instance of WSGIServer)."""
75    self.server_thread = TestServerThread(address, port)
76    self.server_thread.start()
77    self.server_thread.started.wait()
78    if self.server_thread.error:
79        raise self.server_thread.error
80
81def stop_test_server(self):
82    if self.server_thread:
83        self.server_thread.join()
84
85## New Code
86       
87TestCase.start_test_server = classmethod(start_test_server)
88TestCase.stop_test_server = classmethod(stop_test_server)
89
90from windmill.authoring import unit
91
92class WindmillDjangoUnitTest(TestCase, unit.WindmillUnitTestCase):
93    test_port = 8000
94    def setUp(self):
95        self.test_url = 'http://localhost:%d' % self.test_port
96        self.start_test_server('localhost', self.test_port)
97        unit.WindmillUnitTestCase.setUp(self)
98
99    def tearDown(self):
100        unit.WindmillUnitTestCase.tearDown(self)
101        self.stop_test_server()
Note: See TracBrowser for help on using the browser.