Package cherrypy :: Package test :: Module helper
[hide private]
[frames] | no frames]

Source Code for Module cherrypy.test.helper

  1  """A library of helper functions for the CherryPy test suite. 
  2   
  3  The actual script that runs the entire CP test suite is called 
  4  "test.py" (in this folder); test.py calls this module as a library. 
  5   
  6  Usage 
  7  ===== 
  8  Each individual test_*.py module imports this module (helper), 
  9  usually to make an instance of CPWebCase, and then call testmain(). 
 10   
 11  The CP test suite script (test.py) imports this module and calls 
 12  run_test_suite, possibly more than once. CP applications may also 
 13  import test.py (to use TestHarness), which then calls helper.py. 
 14  """ 
 15   
 16  # GREAT CARE has been taken to separate this module from test.py, 
 17  # because different consumers of each have mutually-exclusive import 
 18  # requirements. So don't go moving functions from here into test.py, 
 19  # or vice-versa, unless you *really* know what you're doing. 
 20   
 21  import os 
 22  thisdir = os.path.abspath(os.path.dirname(__file__)) 
 23  import re 
 24  import sys 
 25  import thread 
 26  import time 
 27  import warnings 
 28   
 29  import cherrypy 
 30  from cherrypy.lib import http, profiler 
 31  from cherrypy.test import webtest 
 32   
 33   
34 -class CPWebCase(webtest.WebCase):
35 36 script_name = "" 37 scheme = "http" 38
39 - def prefix(self):
40 return self.script_name.rstrip("/")
41
42 - def base(self):
43 if ((self.scheme == "http" and self.PORT == 80) or 44 (self.scheme == "https" and self.PORT == 443)): 45 port = "" 46 else: 47 port = ":%s" % self.PORT 48 49 return "%s://%s%s%s" % (self.scheme, self.HOST, port, 50 self.script_name.rstrip("/"))
51
52 - def exit(self):
53 sys.exit()
54
55 - def tearDown(self):
56 pass
57
58 - def getPage(self, url, headers=None, method="GET", body=None, protocol=None):
59 """Open the url. Return status, headers, body.""" 60 if self.script_name: 61 url = http.urljoin(self.script_name, url) 62 return webtest.WebCase.getPage(self, url, headers, method, body, protocol)
63
64 - def assertErrorPage(self, status, message=None, pattern=''):
65 """Compare the response body with a built in error page. 66 67 The function will optionally look for the regexp pattern, 68 within the exception embedded in the error page.""" 69 70 # This will never contain a traceback 71 page = cherrypy._cperror.get_error_page(status, message=message) 72 73 # First, test the response body without checking the traceback. 74 # Stick a match-all group (.*) in to grab the traceback. 75 esc = re.escape 76 epage = esc(page) 77 epage = epage.replace(esc('<pre id="traceback"></pre>'), 78 esc('<pre id="traceback">') + '(.*)' + esc('</pre>')) 79 m = re.match(epage, self.body, re.DOTALL) 80 if not m: 81 self._handlewebError('Error page does not match\n' + page) 82 return 83 84 # Now test the pattern against the traceback 85 if pattern is None: 86 # Special-case None to mean that there should be *no* traceback. 87 if m and m.group(1): 88 self._handlewebError('Error page contains traceback') 89 else: 90 if (m is None) or (not re.search(re.escape(pattern), m.group(1))): 91 msg = 'Error page does not contain %s in traceback' 92 self._handlewebError(msg % repr(pattern))
93 94 95 CPTestLoader = webtest.ReloadingTestLoader() 96 CPTestRunner = webtest.TerseTestRunner(verbosity=2) 97
98 -def setConfig(conf):
99 """Set the global config using a copy of conf.""" 100 if isinstance(conf, basestring): 101 # assume it's a filename 102 cherrypy.config.update(conf) 103 else: 104 cherrypy.config.update(conf.copy())
105 106
107 -def run_test_suite(moduleNames, server, conf):
108 """Run the given test modules using the given server and [global] conf. 109 110 The server is started and stopped once, regardless of the number 111 of test modules. The config, however, is reset for each module. 112 """ 113 cherrypy.config.reset() 114 setConfig(conf) 115 engine = cherrypy.engine 116 if hasattr(engine, "signal_handler"): 117 engine.signal_handler.subscribe() 118 if hasattr(engine, "console_control_handler"): 119 engine.console_control_handler.subscribe() 120 # The Pybots automatic testing system needs the suite to exit 121 # with a non-zero value if there were any problems. 122 # Might as well stick it in the engine... :/ 123 engine.test_success = True 124 engine.start_with_callback(_run_test_suite_thread, 125 args=(moduleNames, conf)) 126 engine.block() 127 if engine.test_success: 128 return 0 129 else: 130 return 1
131
132 -def sync_apps(profile=False, validate=False, conquer=False):
133 app = cherrypy.tree 134 if profile: 135 app = profiler.make_app(app, aggregate=False) 136 if conquer: 137 try: 138 import wsgiconq 139 except ImportError: 140 warnings.warn("Error importing wsgiconq. pyconquer will not run.") 141 else: 142 app = wsgiconq.WSGILogger(app) 143 if validate: 144 try: 145 from wsgiref import validate 146 except ImportError: 147 warnings.warn("Error importing wsgiref. The validator will not run.") 148 else: 149 app = validate.validator(app) 150 151 h = cherrypy.server.httpserver 152 if hasattr(h, 'wsgi_app'): 153 # CherryPy's wsgiserver 154 h.wsgi_app = app 155 elif hasattr(h, 'fcgiserver'): 156 # flup's WSGIServer 157 h.fcgiserver.application = app 158 elif hasattr(h, 'scgiserver'): 159 # flup's WSGIServer 160 h.scgiserver.application = app
161
162 -def _run_test_suite_thread(moduleNames, conf):
163 try: 164 for testmod in moduleNames: 165 # Must run each module in a separate suite, 166 # because each module uses/overwrites cherrypy globals. 167 cherrypy.tree = cherrypy._cptree.Tree() 168 cherrypy.config.reset() 169 setConfig(conf) 170 171 m = __import__(testmod, globals(), locals()) 172 setup = getattr(m, "setup_server", None) 173 if setup: 174 setup() 175 176 # The setup functions probably mounted new apps. 177 # Tell our server about them. 178 sync_apps(profile=conf.get("profiling.on", False), 179 validate=conf.get("validator.on", False), 180 conquer=conf.get("conquer.on", False), 181 ) 182 183 suite = CPTestLoader.loadTestsFromName(testmod) 184 result = CPTestRunner.run(suite) 185 cherrypy.engine.test_success &= result.wasSuccessful() 186 187 teardown = getattr(m, "teardown_server", None) 188 if teardown: 189 teardown() 190 finally: 191 cherrypy.engine.exit()
192
193 -def testmain(conf=None):
194 """Run __main__ as a test module, with webtest debugging.""" 195 engine = cherrypy.engine 196 if '--server' in sys.argv: 197 # Run the test module server-side only; wait for Ctrl-C to break. 198 conf = conf or {} 199 conf['server.socket_host'] = '0.0.0.0' 200 setConfig(conf) 201 if hasattr(engine, "signal_handler"): 202 engine.signal_handler.subscribe() 203 if hasattr(engine, "console_control_handler"): 204 engine.console_control_handler.subscribe() 205 engine.start() 206 engine.block() 207 else: 208 for arg in sys.argv: 209 if arg.startswith('--client='): 210 # Run the test module client-side only. 211 sys.argv.remove(arg) 212 conf = conf or {} 213 conf['server.socket_host'] = host = arg.split('=', 1)[1].strip() 214 setConfig(conf) 215 webtest.WebCase.HOST = host 216 webtest.WebCase.PORT = cherrypy.server.socket_port 217 webtest.main() 218 break 219 else: 220 # Run normally (both server and client in same process). 221 conf = conf or {} 222 conf['server.socket_host'] = '127.0.0.1' 223 setConfig(conf) 224 engine.start_with_callback(_test_main_thread) 225 engine.block()
226
227 -def _test_main_thread():
228 try: 229 webtest.WebCase.PORT = cherrypy.server.socket_port 230 webtest.main() 231 finally: 232 cherrypy.engine.exit()
233 234 235 236 # --------------------------- Spawning helpers --------------------------- # 237 238
239 -class CPProcess(object):
240 241 pid_file = os.path.join(thisdir, 'test.pid') 242 config_file = os.path.join(thisdir, 'test.conf') 243 config_template = """[global] 244 server.socket_host: '%(host)s' 245 server.socket_port: %(port)s 246 log.screen: False 247 log.error_file: r'%(error_log)s' 248 log.access_file: r'%(access_log)s' 249 %(ssl)s 250 %(extra)s 251 """ 252 error_log = os.path.join(thisdir, 'test.error.log') 253 access_log = os.path.join(thisdir, 'test.access.log') 254
255 - def __init__(self, wait=False, daemonize=False, ssl=False):
256 self.wait = wait 257 self.daemonize = daemonize 258 self.ssl = ssl 259 self.host = cherrypy.server.socket_host 260 self.port = cherrypy.server.socket_port
261
262 - def write_conf(self, extra=""):
263 if self.ssl: 264 serverpem = os.path.join(thisdir, 'test.pem') 265 ssl = """ 266 server.ssl_certificate: r'%s' 267 server.ssl_private_key: r'%s' 268 """ % (serverpem, serverpem) 269 else: 270 ssl = "" 271 272 f = open(self.config_file, 'wb') 273 f.write(self.config_template % 274 {'host': self.host, 275 'port': self.port, 276 'error_log': self.error_log, 277 'access_log': self.access_log, 278 'ssl': ssl, 279 'extra': extra, 280 }) 281 f.close()
282
283 - def start(self, imports=None):
284 """Start cherryd in a subprocess.""" 285 cherrypy._cpserver.wait_for_free_port(self.host, self.port) 286 287 args = [sys.executable, '/usr/sbin/cherryd', 288 '-c', self.config_file, '-p', self.pid_file] 289 290 if not isinstance(imports, (list, tuple)): 291 imports = [imports] 292 for i in imports: 293 if i: 294 args.append('-i') 295 args.append(i) 296 297 if self.daemonize: 298 args.append('-d') 299 300 if self.wait: 301 self.exit_code = os.spawnl(os.P_WAIT, sys.executable, *args) 302 else: 303 os.spawnl(os.P_NOWAIT, sys.executable, *args) 304 cherrypy._cpserver.wait_for_occupied_port(self.host, self.port) 305 306 # Give the engine a wee bit more time to finish STARTING 307 if self.daemonize: 308 time.sleep(2) 309 else: 310 time.sleep(1)
311
312 - def get_pid(self):
313 return int(open(self.pid_file, 'rb').read())
314
315 - def join(self):
316 """Wait for the process to exit.""" 317 try: 318 try: 319 # Mac, UNIX 320 os.wait() 321 except AttributeError: 322 # Windows 323 try: 324 pid = self.get_pid() 325 except IOError: 326 # Assume the subprocess deleted the pidfile on shutdown. 327 pass 328 else: 329 os.waitpid(pid, 0) 330 except OSError, x: 331 if x.args != (10, 'No child processes'): 332 raise
333