cherrypy.test.test

1 import httplib 2 from httplib import BadStatusLine 3 4 import os 5 import sys 6 import threading 7 import time 8 9 from cherrypy.test import test 10 test.prefer_parent_path() 11 12 import cherrypy 13 14

15 -class Root:

16 - def index(self):

17 return "Hello World"

18 index.exposed = True 19

20 - def ctrlc(self):

21 raise KeyboardInterrupt()

22 ctrlc.exposed = True 23

24 - def restart(self):

25 cherrypy.engine.restart() 26 return "app was restarted succesfully"

27 restart.exposed = True 28

29 - def block_explicit(self):

30 while True: 31 if cherrypy.response.timed_out: 32 cherrypy.response.timed_out = False 33 return "broken!" 34 time.sleep(0.01)

35 block_explicit.exposed = True 36

37 - def block_implicit(self):

38 time.sleep(0.5) 39 return "response.timeout = %s" % cherrypy.response.timeout

40 block_implicit.exposed = True

41 42 cherrypy.tree.mount(Root()) 43 cherrypy.config.update({ 44 'environment': 'test_suite', 45 'engine.deadlock_poll_freq': 0.1, 46 'response.timeout': 0.2, 47 }) 48

49 -class Dependency:

50

51 - def __init__(self):

52 self.running = False 53 self.startcount = 0 54 self.threads = {}

55

56 - def start(self):

57 self.running = True 58 self.startcount += 1

59

60 - def stop(self):

61 self.running = False

62

63 - def startthread(self, thread_id):

64 self.threads[thread_id] = None

65

66 - def stopthread(self, thread_id):

67 del self.threads[thread_id]

68 69 70 from cherrypy.test import helper 71

72 -class ServerStateTests(helper.CPWebCase):

73

74 - def test_0_NormalStateFlow(self):

75 if not self.server_class: 76 # Without having called "cherrypy.engine.start()", we should 77 # get a 503 Service Unavailable response. 78 self.getPage("/") 79 self.assertStatus(503) 80 81 # And our db_connection should not be running 82 self.assertEqual(db_connection.running, False) 83 self.assertEqual(db_connection.startcount, 0) 84 self.assertEqual(len(db_connection.threads), 0) 85 86 # Test server start 87 cherrypy.server.quickstart(self.server_class) 88 cherrypy.engine.start(blocking=False) 89 self.assertEqual(cherrypy.engine.state, 1) 90 91 if self.server_class: 92 host = cherrypy.server.socket_host 93 port = cherrypy.server.socket_port 94 self.assertRaises(IOError, cherrypy._cpserver.check_port, host, port) 95 96 # The db_connection should be running now 97 self.assertEqual(db_connection.running, True) 98 self.assertEqual(db_connection.startcount, 1) 99 self.assertEqual(len(db_connection.threads), 0) 100 101 self.getPage("/") 102 self.assertBody("Hello World") 103 self.assertEqual(len(db_connection.threads), 1) 104 105 # Test engine stop 106 cherrypy.engine.stop() 107 self.assertEqual(cherrypy.engine.state, 0) 108 109 # Verify that the on_stop_engine function was called 110 self.assertEqual(db_connection.running, False) 111 self.assertEqual(len(db_connection.threads), 0) 112 113 if not self.server_class: 114 # Once the engine has stopped, we should get a 503 115 # error again. (If we were running an HTTP server, 116 # then the connection should not even be processed). 117 self.getPage("/") 118 self.assertStatus(503) 119 120 # Block the main thread now and verify that stop() works. 121 def stoptest(): 122 self.getPage("/") 123 self.assertBody("Hello World") 124 cherrypy.engine.stop()

125 cherrypy.engine.start_with_callback(stoptest) 126 self.assertEqual(cherrypy.engine.state, 0) 127 cherrypy.server.stop()

128

129 - def test_1_Restart(self):

130 cherrypy.server.start() 131 cherrypy.engine.start(blocking=False) 132 133 # The db_connection should be running now 134 self.assertEqual(db_connection.running, True) 135 sc = db_connection.startcount 136 137 self.getPage("/") 138 self.assertBody("Hello World") 139 self.assertEqual(len(db_connection.threads), 1) 140 141 # Test server restart from this thread 142 cherrypy.engine.restart() 143 self.assertEqual(cherrypy.engine.state, 1) 144 self.getPage("/") 145 self.assertBody("Hello World") 146 self.assertEqual(db_connection.running, True) 147 self.assertEqual(db_connection.startcount, sc + 1) 148 self.assertEqual(len(db_connection.threads), 1) 149 150 # Test server restart from inside a page handler 151 self.getPage("/restart") 152 self.assertEqual(cherrypy.engine.state, 1) 153 self.assertBody("app was restarted succesfully") 154 self.assertEqual(db_connection.running, True) 155 self.assertEqual(db_connection.startcount, sc + 2) 156 # Since we are requesting synchronously, is only one thread used? 157 # Note that the "/restart" request has been flushed. 158 self.assertEqual(len(db_connection.threads), 0) 159 160 cherrypy.engine.stop() 161 self.assertEqual(cherrypy.engine.state, 0) 162 self.assertEqual(db_connection.running, False) 163 self.assertEqual(len(db_connection.threads), 0) 164 cherrypy.server.stop()

165

166 - def test_2_KeyboardInterrupt(self):

167 if self.server_class: 168 169 # Raise a keyboard interrupt in the HTTP server's main thread. 170 # We must start the server in this, the main thread 171 cherrypy.engine.start(blocking=False) 172 cherrypy.server.start() 173 174 self.persistent = True 175 try: 176 # Make the first request and assert there's no "Connection: close". 177 self.getPage("/") 178 self.assertStatus('200 OK') 179 self.assertBody("Hello World") 180 self.assertNoHeader("Connection") 181 182 cherrypy.server.httpservers.keys()[0].interrupt = KeyboardInterrupt 183 while cherrypy.engine.state != 0: 184 time.sleep(0.1) 185 186 self.assertEqual(db_connection.running, False) 187 self.assertEqual(len(db_connection.threads), 0) 188 self.assertEqual(cherrypy.engine.state, 0) 189 finally: 190 self.persistent = False 191 192 # Raise a keyboard interrupt in a page handler; on multithreaded 193 # servers, this should occur in one of the worker threads. 194 # This should raise a BadStatusLine error, since the worker 195 # thread will just die without writing a response. 196 cherrypy.engine.start(blocking=False) 197 cherrypy.server.start() 198 199 try: 200 self.getPage("/ctrlc") 201 except BadStatusLine: 202 pass 203 else: 204 print self.body 205 self.fail("AssertionError: BadStatusLine not raised") 206 207 while cherrypy.engine.state != 0: 208 time.sleep(0.1) 209 self.assertEqual(db_connection.running, False) 210 self.assertEqual(len(db_connection.threads), 0)

211

212 - def test_3_Deadlocks(self):

213 cherrypy.engine.start(blocking=False) 214 cherrypy.server.start() 215 try: 216 self.assertNotEqual(cherrypy.engine.monitor_thread, None) 217 218 # Request a "normal" page. 219 self.assertEqual(cherrypy.engine.servings, []) 220 self.getPage("/") 221 self.assertBody("Hello World") 222 # request.close is called async. 223 while cherrypy.engine.servings: 224 time.sleep(0.01) 225 226 # Request a page that explicitly checks itself for deadlock. 227 # The deadlock_timeout should be 2 secs. 228 self.getPage("/block_explicit") 229 self.assertBody("broken!") 230 231 # Request a page that implicitly breaks deadlock. 232 # If we deadlock, we want to touch as little code as possible, 233 # so we won't even call handle_error, just bail ASAP. 234 self.getPage("/block_implicit") 235 self.assertStatus(500) 236 self.assertInBody("raise cherrypy.TimeoutError()") 237 finally: 238 cherrypy.engine.stop() 239 cherrypy.server.stop()

240

241 - def test_4_Autoreload(self):

242 if not self.server_class: 243 print "skipped (no server) ", 244 return 245 246 # Start the demo script in a new process 247 demoscript = os.path.join(os.getcwd(), os.path.dirname(__file__), 248 "test_states_demo.py") 249 host = cherrypy.server.socket_host 250 port = cherrypy.server.socket_port 251 cherrypy._cpserver.wait_for_free_port(host, port) 252 253 args = [sys.executable, demoscript, host, str(port)] 254 if self.scheme == "https": 255 args.append('-ssl') 256 pid = os.spawnl(os.P_NOWAIT, sys.executable, *args) 257 pid = str(pid) 258 cherrypy._cpserver.wait_for_occupied_port(host, port) 259 260 try: 261 self.getPage("/pid") 262 assert self.body.isdigit(), self.body 263 pid = self.body 264 265 # Give the autoreloader time to cache the file time. 266 time.sleep(2) 267 268 # Touch the file 269 f = open(demoscript, 'ab') 270 f.write(" ") 271 f.close() 272 273 # Give the autoreloader time to re-exec the process 274 time.sleep(2) 275 cherrypy._cpserver.wait_for_occupied_port(host, port) 276 277 self.getPage("/pid") 278 assert self.body.isdigit(), self.body 279 self.assertNotEqual(self.body, pid) 280 pid = self.body 281 finally: 282 # Shut down the spawned process 283 self.getPage("/stop") 284 285 try: 286 try: 287 # Mac, UNIX 288 print os.wait() 289 except AttributeError: 290 # Windows 291 print os.waitpid(int(

Source Code for Module cherrypy.test.test_states