1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
# Code from django_live_server_r8458.diff @ http://code.djangoproject.com/ticket/2879#comment:41
# Editing to monkey patch django rather than be in trunk
import socket
import threading
from django.core.handlers.wsgi import WSGIHandler
from django.core.servers import basehttp
from django.test.testcases import call_command
#from django.core.management import call_command
# support both django 1.0 and 1.1
try:
from django.test.testcases import TransactionTestCase as TestCase
except ImportError:
from django.test.testcases import TestCase
try:
from windmill.authoring import unit
except Exception, e:
print "You don't appear to have windmill installed, please install before trying to run windmill tests again."
unit = None
class StoppableWSGIServer(basehttp.WSGIServer):
"""WSGIServer with short timeout, so that server thread can stop this server."""
def server_bind(self):
"""Sets timeout to 1 second."""
basehttp.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
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."""
# 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:'):
from django.db import connection
print 'Creating test DB'
db_name = connection.creation.create_test_db(0,autoclobber=True)
#call_command('syncdb', 0, 0)
# Import the fixture data into the test database.
if hasattr(self, 'fixtures'):
print 'Loading 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})
try:
print "running thread"
handler = basehttp.AdminMediaHandler(WSGIHandler())
httpd = None
while httpd is None:
try:
server_address = (self.address, self.port)
httpd = StoppableWSGIServer(server_address, basehttp.WSGIRequestHandler)
except basehttp.WSGIServerException, e:
if "Address already in use" in str(e):
print "Address already in use"
self.port +=1
else:
raise e
httpd.set_app(handler)
self.started.set()
except basehttp.WSGIServerException, e:
self.error = e
self.started.set()
return
# Loop until we get a stop event.
while not self._stopevent.isSet():
httpd.handle_request()
httpd.server_close()
def join(self, timeout=None):
"""Stop the thread and wait for it to finish."""
self._stopevent.set()
threading.Thread.join(self, timeout)
def start_test_server(self, address='localhost', port=8000):
"""Creates a live test server object (instance of WSGIServer)."""
self.server_thread = TestServerThread(address, port)
if hasattr(self, 'fixtures'):
print 'loading fixtures %s' % self.fixtures
self.server_thread.__setattr__('fixtures', self.fixtures)
self.server_thread.start()
self.server_thread.started.wait()
if self.server_thread.error:
raise self.server_thread.error
return self.server_thread.started
def stop_test_server(self):
if self.server_thread:
self.server_thread.join()
## New Code
TestCase.start_test_server = classmethod(start_test_server)
TestCase.stop_test_server = classmethod(stop_test_server)
class WindmillDjangoUnitTest(TestCase, unit.WindmillUnitTestCase):
test_port = 8000
def setUp(self):
self.start_test_server('localhost', self.test_port)
self.test_url = 'http://localhost:%d' % self.server_thread.port
unit.WindmillUnitTestCase.setUp(self)
def tearDown(self):
unit.WindmillUnitTestCase.tearDown(self)
self.stop_test_server()
WindmillDjangoTransactionUnitTest = WindmillDjangoUnitTest
|
