/usr/share/pyshared/cherrypy/test/test_states.py is in python-cherrypy 2.3.0-3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 | import test
test.prefer_parent_path()
import threading
import cherrypy
class Root:
def index(self):
return "Hello World"
index.exposed = True
def ctrlc(self):
raise KeyboardInterrupt()
ctrlc.exposed = True
def restart(self):
cherrypy.server.restart()
return "app was restarted succesfully"
restart.exposed = True
cherrypy.root = Root()
cherrypy.config.update({
'global': {
'server.log_to_screen': False,
'server.environment': 'production',
},
})
class Dependency:
def __init__(self):
self.running = False
self.startcount = 0
self.threads = {}
def start(self):
self.running = True
self.startcount += 1
def stop(self):
self.running = False
def startthread(self, thread_id):
self.threads[thread_id] = None
def stopthread(self, thread_id):
del self.threads[thread_id]
import helper
class ServerStateTests(helper.CPWebCase):
def test_0_NormalStateFlow(self):
if not self.server_class:
# Without having called "cherrypy.server.start()", we should
# get a NotReady error
self.assertRaises(cherrypy.NotReady, self.getPage, "/")
# And our db_connection should not be running
self.assertEqual(db_connection.running, False)
self.assertEqual(db_connection.startcount, 0)
self.assertEqual(len(db_connection.threads), 0)
# Test server start
cherrypy.server.start(True, self.server_class)
self.assertEqual(cherrypy.server.state, 1)
if self.server_class:
host = cherrypy.config.get('server.socket_host')
port = cherrypy.config.get('server.socket_port')
self.assertRaises(IOError, cherrypy._cpserver.check_port, host, port)
# The db_connection should be running now
self.assertEqual(db_connection.running, True)
self.assertEqual(db_connection.startcount, 1)
self.assertEqual(len(db_connection.threads), 0)
self.getPage("/")
self.assertBody("Hello World")
self.assertEqual(len(db_connection.threads), 1)
# Test server stop
cherrypy.server.stop()
self.assertEqual(cherrypy.server.state, 0)
# Verify that the on_stop_server function was called
self.assertEqual(db_connection.running, False)
self.assertEqual(len(db_connection.threads), 0)
if not self.server_class:
# Once the server has stopped, we should get a NotReady
# error again. (If we were running an HTTP server,
# then the connection should not even be processed).
self.assertRaises(cherrypy.NotReady, self.getPage, "/")
# Block the main thread now and verify that stop() works.
def stoptest():
self.getPage("/")
self.assertBody("Hello World")
cherrypy.server.stop()
cherrypy.server.start_with_callback(stoptest, server_class=self.server_class)
self.assertEqual(cherrypy.server.state, 0)
def test_1_Restart(self):
cherrypy.server.start(True, self.server_class)
# The db_connection should be running now
self.assertEqual(db_connection.running, True)
sc = db_connection.startcount
self.getPage("/")
self.assertBody("Hello World")
self.assertEqual(len(db_connection.threads), 1)
# Test server restart from this thread
cherrypy.server.restart()
self.assertEqual(cherrypy.server.state, 1)
self.getPage("/")
self.assertBody("Hello World")
self.assertEqual(db_connection.running, True)
self.assertEqual(db_connection.startcount, sc + 1)
self.assertEqual(len(db_connection.threads), 1)
# Test server restart from inside a page handler
self.getPage("/restart")
self.assertEqual(cherrypy.server.state, 1)
self.assertBody("app was restarted succesfully")
self.assertEqual(db_connection.running, True)
self.assertEqual(db_connection.startcount, sc + 2)
# Since we are requesting synchronously, is only one thread used?
# Note that the "/restart" request has been flushed.
self.assertEqual(len(db_connection.threads), 0)
cherrypy.server.stop()
self.assertEqual(cherrypy.server.state, 0)
self.assertEqual(db_connection.running, False)
self.assertEqual(len(db_connection.threads), 0)
def test_2_KeyboardInterrupt(self):
if self.server_class:
# Raise a keyboard interrupt in the HTTP server's main thread.
def interrupt():
cherrypy.server.wait()
cherrypy.server.httpserver.interrupt = KeyboardInterrupt
threading.Thread(target=interrupt).start()
# We must start the server in this, the main thread
cherrypy.server.start(False, self.server_class)
# Time passes...
self.assertEqual(cherrypy.server.state, 0)
self.assertEqual(db_connection.running, False)
self.assertEqual(len(db_connection.threads), 0)
# Raise a keyboard interrupt in a page handler; on multithreaded
# servers, this should occur in one of the worker threads.
# This should raise a BadStatusLine error, since the worker
# thread will just die without writing a response.
def interrupt():
cherrypy.server.wait()
from httplib import BadStatusLine
self.assertRaises(BadStatusLine, self.getPage, "/ctrlc")
threading.Thread(target=interrupt).start()
cherrypy.server.start(False, self.server_class)
# Time passes...
self.assertEqual(cherrypy.server.state, 0)
self.assertEqual(db_connection.running, False)
self.assertEqual(len(db_connection.threads), 0)
db_connection = None
def run(server, conf):
helper.setConfig(conf)
ServerStateTests.server_class = server
suite = helper.CPTestLoader.loadTestsFromTestCase(ServerStateTests)
try:
global db_connection
db_connection = Dependency()
cherrypy.server.on_start_server_list.append(db_connection.start)
cherrypy.server.on_stop_server_list.append(db_connection.stop)
cherrypy.server.on_start_thread_list.append(db_connection.startthread)
cherrypy.server.on_stop_thread_list.append(db_connection.stopthread)
helper.CPTestRunner.run(suite)
finally:
cherrypy.server.stop()
def run_all(host, port):
conf = {'server.socket_host': host,
'server.socket_port': port,
'server.thread_pool': 10,
'server.log_to_screen': False,
'server.log_config_options': False,
'server.environment': "production",
'server.show_tracebacks': True,
}
def _run(server):
print
print "Testing %s on %s:%s..." % (server, host, port)
run(server, conf)
_run("cherrypy._cpwsgi.WSGIServer")
def run_localhosts(port):
for host in ("", "127.0.0.1", "localhost"):
conf = {'server.socket_host': host,
'server.socket_port': port,
'server.thread_pool': 10,
'server.log_to_screen': False,
'server.log_config_options': False,
'server.environment': "production",
'server.show_tracebacks': True,
}
def _run(server):
print
print "Testing %s on %s:%s..." % (server, host, port)
run(server, conf)
_run("cherrypy._cpwsgi.WSGIServer")
if __name__ == "__main__":
import sys
host = '127.0.0.1'
port = 8000
if len(sys.argv) > 1:
cmd = sys.argv[1]
if cmd in [prefix + atom for atom in ("?", "h", "help")
for prefix in ("", "-", "--", "\\")]:
print
print "test_states.py -? -> this help page"
print "test_states.py [host] [port] -> run the tests on the given host/port"
print "test_states.py -localhosts [port] -> try various localhost strings"
sys.exit(0)
if len(sys.argv) > 2:
port = int(sys.argv[2])
if cmd == "-localhosts":
run_localhosts(port)
sys.exit(0)
host = sys.argv[1].strip("\"'")
run_all(host, port)
|