test_states.py (11374B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\lib\cherrypy\test\test_states.py 2 from cherrypy._cpcompat import BadStatusLine, ntob 3 import os 4 import sys 5 import threading 6 import time 7 import cherrypy 8 engine = cherrypy.engine 9 thisdir = os.path.join(os.getcwd(), os.path.dirname(__file__)) 10 11 class Dependency: 12 13 def __init__(self, bus): 14 self.bus = bus 15 self.running = False 16 self.startcount = 0 17 self.gracecount = 0 18 self.threads = {} 19 20 def subscribe(self): 21 self.bus.subscribe('start', self.start) 22 self.bus.subscribe('stop', self.stop) 23 self.bus.subscribe('graceful', self.graceful) 24 self.bus.subscribe('start_thread', self.startthread) 25 self.bus.subscribe('stop_thread', self.stopthread) 26 27 def start(self): 28 self.running = True 29 self.startcount += 1 30 31 def stop(self): 32 self.running = False 33 34 def graceful(self): 35 self.gracecount += 1 36 37 def startthread(self, thread_id): 38 self.threads[thread_id] = None 39 40 def stopthread(self, thread_id): 41 del self.threads[thread_id] 42 43 44 db_connection = Dependency(engine) 45 46 def setup_server(): 47 48 class Root: 49 50 def index(self): 51 return 'Hello World' 52 53 index.exposed = True 54 55 def ctrlc(self): 56 raise KeyboardInterrupt() 57 58 ctrlc.exposed = True 59 60 def graceful(self): 61 engine.graceful() 62 return 'app was (gracefully) restarted succesfully' 63 64 graceful.exposed = True 65 66 def block_explicit(self): 67 while True: 68 if cherrypy.response.timed_out: 69 cherrypy.response.timed_out = False 70 return 'broken!' 71 time.sleep(0.01) 72 73 block_explicit.exposed = True 74 75 def block_implicit(self): 76 time.sleep(0.5) 77 return 'response.timeout = %s' % cherrypy.response.timeout 78 79 block_implicit.exposed = True 80 81 cherrypy.tree.mount(Root()) 82 cherrypy.config.update({'environment': 'test_suite', 83 'engine.deadlock_poll_freq': 0.1}) 84 db_connection.subscribe() 85 86 87 from cherrypy.test import helper 88 89 class ServerStateTests(helper.CPWebCase): 90 setup_server = staticmethod(setup_server) 91 92 def setUp(self): 93 cherrypy.server.socket_timeout = 0.1 94 95 def test_0_NormalStateFlow(self): 96 engine.stop() 97 self.assertEqual(db_connection.running, False) 98 self.assertEqual(db_connection.startcount, 1) 99 self.assertEqual(len(db_connection.threads), 0) 100 engine.start() 101 self.assertEqual(engine.state, engine.states.STARTED) 102 host = cherrypy.server.socket_host 103 port = cherrypy.server.socket_port 104 self.assertRaises(IOError, cherrypy._cpserver.check_port, host, port) 105 self.assertEqual(db_connection.running, True) 106 self.assertEqual(db_connection.startcount, 2) 107 self.assertEqual(len(db_connection.threads), 0) 108 self.getPage('/') 109 self.assertBody('Hello World') 110 self.assertEqual(len(db_connection.threads), 1) 111 engine.stop() 112 self.assertEqual(engine.state, engine.states.STOPPED) 113 self.assertEqual(db_connection.running, False) 114 self.assertEqual(len(db_connection.threads), 0) 115 116 def exittest(): 117 self.getPage('/') 118 self.assertBody('Hello World') 119 engine.exit() 120 121 cherrypy.server.start() 122 engine.start_with_callback(exittest) 123 engine.block() 124 self.assertEqual(engine.state, engine.states.EXITING) 125 126 def test_1_Restart(self): 127 cherrypy.server.start() 128 engine.start() 129 self.assertEqual(db_connection.running, True) 130 grace = db_connection.gracecount 131 self.getPage('/') 132 self.assertBody('Hello World') 133 self.assertEqual(len(db_connection.threads), 1) 134 engine.graceful() 135 self.assertEqual(engine.state, engine.states.STARTED) 136 self.getPage('/') 137 self.assertBody('Hello World') 138 self.assertEqual(db_connection.running, True) 139 self.assertEqual(db_connection.gracecount, grace + 1) 140 self.assertEqual(len(db_connection.threads), 1) 141 self.getPage('/graceful') 142 self.assertEqual(engine.state, engine.states.STARTED) 143 self.assertBody('app was (gracefully) restarted succesfully') 144 self.assertEqual(db_connection.running, True) 145 self.assertEqual(db_connection.gracecount, grace + 2) 146 self.assertEqual(len(db_connection.threads), 0) 147 engine.stop() 148 self.assertEqual(engine.state, engine.states.STOPPED) 149 self.assertEqual(db_connection.running, False) 150 self.assertEqual(len(db_connection.threads), 0) 151 152 def test_2_KeyboardInterrupt(self): 153 engine.start() 154 cherrypy.server.start() 155 self.persistent = True 156 try: 157 self.getPage('/') 158 self.assertStatus('200 OK') 159 self.assertBody('Hello World') 160 self.assertNoHeader('Connection') 161 cherrypy.server.httpserver.interrupt = KeyboardInterrupt 162 engine.block() 163 self.assertEqual(db_connection.running, False) 164 self.assertEqual(len(db_connection.threads), 0) 165 self.assertEqual(engine.state, engine.states.EXITING) 166 finally: 167 self.persistent = False 168 169 engine.start() 170 cherrypy.server.start() 171 try: 172 self.getPage('/ctrlc') 173 except BadStatusLine: 174 pass 175 else: 176 print self.body 177 self.fail('AssertionError: BadStatusLine not raised') 178 179 engine.block() 180 self.assertEqual(db_connection.running, False) 181 self.assertEqual(len(db_connection.threads), 0) 182 183 def test_3_Deadlocks(self): 184 cherrypy.config.update({'response.timeout': 0.2}) 185 engine.start() 186 cherrypy.server.start() 187 try: 188 self.assertNotEqual(engine.timeout_monitor.thread, None) 189 self.assertEqual(engine.timeout_monitor.servings, []) 190 self.getPage('/') 191 self.assertBody('Hello World') 192 while engine.timeout_monitor.servings: 193 sys.stdout.write('.') 194 time.sleep(0.01) 195 196 self.getPage('/block_explicit') 197 self.assertBody('broken!') 198 self.getPage('/block_implicit') 199 self.assertStatus(500) 200 self.assertInBody('raise cherrypy.TimeoutError()') 201 finally: 202 engine.exit() 203 204 def test_4_Autoreload(self): 205 p = helper.CPProcess(ssl=self.scheme.lower() == 'https') 206 p.write_conf(extra='test_case_name: "test_4_Autoreload"') 207 p.start(imports='cherrypy.test._test_states_demo') 208 try: 209 self.getPage('/start') 210 start = float(self.body) 211 time.sleep(2) 212 os.utime(os.path.join(thisdir, '_test_states_demo.py'), None) 213 time.sleep(2) 214 host = cherrypy.server.socket_host 215 port = cherrypy.server.socket_port 216 cherrypy._cpserver.wait_for_occupied_port(host, port) 217 self.getPage('/start') 218 self.assert_(float(self.body) > start) 219 finally: 220 self.getPage('/exit') 221 222 p.join() 223 224 def test_5_Start_Error(self): 225 p = helper.CPProcess(ssl=self.scheme.lower() == 'https', wait=True) 226 p.write_conf(extra='starterror: True\ntest_case_name: "test_5_Start_Error"\n') 227 p.start(imports='cherrypy.test._test_states_demo') 228 if p.exit_code == 0: 229 self.fail('Process failed to return nonzero exit code.') 230 231 232 class PluginTests(helper.CPWebCase): 233 234 def test_daemonize(self): 235 if os.name not in ('posix',): 236 return self.skip('skipped (not on posix) ') 237 self.HOST = '127.0.0.1' 238 self.PORT = 8081 239 p = helper.CPProcess(ssl=self.scheme.lower() == 'https', wait=True, daemonize=True, socket_host='127.0.0.1', socket_port=8081) 240 p.write_conf(extra='test_case_name: "test_daemonize"') 241 p.start(imports='cherrypy.test._test_states_demo') 242 try: 243 self.getPage('/pid') 244 self.assertStatus(200) 245 page_pid = int(self.body) 246 self.assertEqual(page_pid, p.get_pid()) 247 finally: 248 self.getPage('/exit') 249 250 p.join() 251 if p.exit_code != 0: 252 self.fail('Daemonized parent process failed to exit cleanly.') 253 254 255 class SignalHandlingTests(helper.CPWebCase): 256 257 def test_SIGHUP_tty(self): 258 try: 259 from signal import SIGHUP 260 except ImportError: 261 return self.skip('skipped (no SIGHUP) ') 262 263 p = helper.CPProcess(ssl=self.scheme.lower() == 'https') 264 p.write_conf(extra='test_case_name: "test_SIGHUP_tty"') 265 p.start(imports='cherrypy.test._test_states_demo') 266 os.kill(p.get_pid(), SIGHUP) 267 p.join() 268 269 def test_SIGHUP_daemonized(self): 270 try: 271 from signal import SIGHUP 272 except ImportError: 273 return self.skip('skipped (no SIGHUP) ') 274 275 if os.name not in ('posix',): 276 return self.skip('skipped (not on posix) ') 277 p = helper.CPProcess(ssl=self.scheme.lower() == 'https', wait=True, daemonize=True) 278 p.write_conf(extra='test_case_name: "test_SIGHUP_daemonized"') 279 p.start(imports='cherrypy.test._test_states_demo') 280 pid = p.get_pid() 281 try: 282 os.kill(pid, SIGHUP) 283 time.sleep(2) 284 self.getPage('/pid') 285 self.assertStatus(200) 286 new_pid = int(self.body) 287 self.assertNotEqual(new_pid, pid) 288 finally: 289 self.getPage('/exit') 290 291 p.join() 292 293 def test_SIGTERM(self): 294 try: 295 from signal import SIGTERM 296 except ImportError: 297 return self.skip('skipped (no SIGTERM) ') 298 299 try: 300 from os import kill 301 except ImportError: 302 return self.skip('skipped (no os.kill) ') 303 304 p = helper.CPProcess(ssl=self.scheme.lower() == 'https') 305 p.write_conf(extra='test_case_name: "test_SIGTERM"') 306 p.start(imports='cherrypy.test._test_states_demo') 307 os.kill(p.get_pid(), SIGTERM) 308 p.join() 309 if os.name in ('posix',): 310 p = helper.CPProcess(ssl=self.scheme.lower() == 'https', wait=True, daemonize=True) 311 p.write_conf(extra='test_case_name: "test_SIGTERM_2"') 312 p.start(imports='cherrypy.test._test_states_demo') 313 os.kill(p.get_pid(), SIGTERM) 314 p.join() 315 316 def test_signal_handler_unsubscribe(self): 317 try: 318 from signal import SIGTERM 319 except ImportError: 320 return self.skip('skipped (no SIGTERM) ') 321 322 try: 323 from os import kill 324 except ImportError: 325 return self.skip('skipped (no os.kill) ') 326 327 p = helper.CPProcess(ssl=self.scheme.lower() == 'https') 328 p.write_conf(extra='unsubsig: True\ntest_case_name: "test_signal_handler_unsubscribe"\n') 329 p.start(imports='cherrypy.test._test_states_demo') 330 os.kill(p.get_pid(), SIGTERM) 331 p.join() 332 target_line = open(p.error_log, 'rb').readlines()[-10] 333 if ntob('I am an old SIGTERM handler.') not in target_line: 334 self.fail('Old SIGTERM handler did not run.\n%r' % target_line)