root/trunk/windmill/authoring/djangotest.py

Revision 1355, 4.8 KB (checked in by admc, 6 months ago)

Moving localhost to 127.0.0.1 to fix vista problems

Line 
1#   Copyright (c) 2008-2009 Mikeal Rogers <mikeal.rogers@gmail.com>
2#
3#   Licensed under the Apache License, Version 2.0 (the "License");
4#   you may not use this file except in compliance with the License.
5#   You may obtain a copy of the License at
6#
7#       http://www.apache.org/licenses/LICENSE-2.0
8#
9#   Unless required by applicable law or agreed to in writing, software
10#   distributed under the License is distributed on an "AS IS" BASIS,
11#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12#   See the License for the specific language governing permissions and
13#   limitations under the License.
14
15# Code from django_live_server_r8458.diff @  http://code.djangoproject.com/ticket/2879#comment:41
16# Editing to monkey patch django rather than be in trunk
17
18import socket
19import threading
20from django.core.handlers.wsgi import WSGIHandler
21from django.core.servers import basehttp
22from django.test.testcases import call_command
23 
24# support both django 1.0 and 1.1
25try:
26    from django.test.testcases import TransactionTestCase as TestCase
27except ImportError:
28    from django.test.testcases import TestCase
29
30class StoppableWSGIServer(basehttp.WSGIServer):
31    """WSGIServer with short timeout, so that server thread can stop this server."""
32
33    def server_bind(self):
34        """Sets timeout to 1 second."""
35        basehttp.WSGIServer.server_bind(self)
36        self.socket.settimeout(1)
37
38    def get_request(self):
39        """Checks for timeout when getting request."""
40        try:
41            sock, address = self.socket.accept()
42            sock.settimeout(None)
43            return (sock, address)
44        except socket.timeout:
45            raise
46
47class TestServerThread(threading.Thread):
48    """Thread for running a http server while tests are running."""
49
50    def __init__(self, address, port):
51        self.address = address
52        self.port = port
53        self._stopevent = threading.Event()
54        self.started = threading.Event()
55        self.error = None
56        super(TestServerThread, self).__init__()
57
58    def run(self):
59        """Sets up test server and database and loops over handling http requests."""
60        try:
61            handler = basehttp.AdminMediaHandler(WSGIHandler())
62            httpd = None
63            while httpd is None:
64                try:
65                    server_address = (self.address, self.port)
66                    httpd = StoppableWSGIServer(server_address, basehttp.WSGIRequestHandler)
67                except basehttp.WSGIServerException, e:
68                    if "Address already in use" in str(e):
69                        self.port +=1
70                    else:
71                        raise e
72            httpd.set_app(handler)
73            self.started.set()
74        except basehttp.WSGIServerException, e:
75            self.error = e
76            self.started.set()
77            return
78
79        # Must do database stuff in this new thread if database in memory.
80        from django.conf import settings
81        if settings.DATABASE_ENGINE == 'sqlite3' \
82            and (not settings.TEST_DATABASE_NAME or settings.TEST_DATABASE_NAME == ':memory:'):
83            from django.db import connection
84            db_name = connection.creation.create_test_db(0)
85            # Import the fixture data into the test database.
86            if hasattr(self, 'fixtures'):
87                # We have to use this slightly awkward syntax due to the fact
88                # that we're using *args and **kwargs together.
89                call_command('loaddata', *self.fixtures, **{'verbosity': 0})
90
91        # Loop until we get a stop event.
92        while not self._stopevent.isSet():
93            httpd.handle_request()
94        httpd.server_close()
95
96    def join(self, timeout=None):
97        """Stop the thread and wait for it to finish."""
98        self._stopevent.set()
99        threading.Thread.join(self, timeout)
100
101
102def start_test_server(self, address='127.0.0.1', port=8000):
103    """Creates a live test server object (instance of WSGIServer)."""
104    self.server_thread = TestServerThread(address, port)
105    self.server_thread.start()
106    self.server_thread.started.wait()
107    if self.server_thread.error:
108        raise self.server_thread.error
109
110def stop_test_server(self):
111    if self.server_thread:
112        self.server_thread.join()
113
114## New Code
115       
116TestCase.start_test_server = classmethod(start_test_server)
117TestCase.stop_test_server = classmethod(stop_test_server)
118
119from windmill.authoring import unit
120
121class WindmillDjangoUnitTest(TestCase, unit.WindmillUnitTestCase):
122    test_port = 8000
123    def setUp(self):
124        self.start_test_server('127.0.0.1', self.test_port)
125        self.test_url = 'http://127.0.0.1:%d' % self.server_thread.port
126        unit.WindmillUnitTestCase.setUp(self)
127
128    def tearDown(self):
129        unit.WindmillUnitTestCase.tearDown(self)
130        self.stop_test_server()
131
132WindmillDjangoTransactionUnitTest = WindmillDjangoUnitTest
Note: See TracBrowser for help on using the browser.