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  import datetime 
  4  import logging 
  5  log = logging.getLogger(__name__) 
  6  import os 
  7  thisdir = os.path.abspath(os.path.dirname(__file__)) 
  8  serverpem = os.path.join(os.getcwd(), thisdir, 'test.pem') 
  9   
 10  import re 
 11  import sys 
 12  import time 
 13  import warnings 
 14   
 15  import cherrypy 
 16  from cherrypy._cpcompat import basestring, copyitems, HTTPSConnection, ntob 
 17  from cherrypy.lib import httputil 
 18  from cherrypy.lib import gctools 
 19  from cherrypy.lib.reprconf import unrepr 
 20  from cherrypy.test import webtest 
 21   
 22  import nose 
 23   
 24  _testconfig = None 
 25   
26 -def get_tst_config(overconf = {}):
27 global _testconfig 28 if _testconfig is None: 29 conf = { 30 'scheme': 'http', 31 'protocol': "HTTP/1.1", 32 'port': 54583, 33 'host': '127.0.0.1', 34 'validate': False, 35 'conquer': False, 36 'server': 'wsgi', 37 } 38 try: 39 import testconfig 40 _conf = testconfig.config.get('supervisor', None) 41 if _conf is not None: 42 for k, v in _conf.items(): 43 if isinstance(v, basestring): 44 _conf[k] = unrepr(v) 45 conf.update(_conf) 46 except ImportError: 47 pass 48 _testconfig = conf 49 conf = _testconfig.copy() 50 conf.update(overconf) 51 52 return conf
53
54 -class Supervisor(object):
55 """Base class for modeling and controlling servers during testing.""" 56
57 - def __init__(self, **kwargs):
58 for k, v in kwargs.items(): 59 if k == 'port': 60 setattr(self, k, int(v)) 61 setattr(self, k, v)
62 63 64 log_to_stderr = lambda msg, level: sys.stderr.write(msg + os.linesep) 65
66 -class LocalSupervisor(Supervisor):
67 """Base class for modeling/controlling servers which run in the same process. 68 69 When the server side runs in a different process, start/stop can dump all 70 state between each test module easily. When the server side runs in the 71 same process as the client, however, we have to do a bit more work to ensure 72 config and mounted apps are reset between tests. 73 """ 74 75 using_apache = False 76 using_wsgi = False 77
78 - def __init__(self, **kwargs):
79 for k, v in kwargs.items(): 80 setattr(self, k, v) 81 82 cherrypy.server.httpserver = self.httpserver_class 83 84 # This is perhaps the wrong place for this call but this is the only 85 # place that i've found so far that I KNOW is early enough to set this. 86 cherrypy.config.update({'log.screen': False}) 87 engine = cherrypy.engine 88 if hasattr(engine, "signal_handler"): 89 engine.signal_handler.subscribe() 90 if hasattr(engine, "console_control_handler"): 91 engine.console_control_handler.subscribe()
92 #engine.subscribe('log', log_to_stderr) 93
94 - def start(self, modulename=None):
95 """Load and start the HTTP server.""" 96 if modulename: 97 # Unhook httpserver so cherrypy.server.start() creates a new 98 # one (with config from setup_server, if declared). 99 cherrypy.server.httpserver = None 100 101 cherrypy.engine.start() 102 103 self.sync_apps()
104
105 - def sync_apps(self):
106 """Tell the server about any apps which the setup functions mounted.""" 107 pass
108
109 - def stop(self):
110 td = getattr(self, 'teardown', None) 111 if td: 112 td() 113 114 cherrypy.engine.exit() 115 116 for name, server in copyitems(getattr(cherrypy, 'servers', {})): 117 server.unsubscribe() 118 del cherrypy.servers[name]
119 120
121 -class NativeServerSupervisor(LocalSupervisor):
122 """Server supervisor for the builtin HTTP server.""" 123 124 httpserver_class = "cherrypy._cpnative_server.CPHTTPServer" 125 using_apache = False 126 using_wsgi = False 127
128 - def __str__(self):
129 return "Builtin HTTP Server on %s:%s" % (self.host, self.port)
130 131
132 -class LocalWSGISupervisor(LocalSupervisor):
133 """Server supervisor for the builtin WSGI server.""" 134 135 httpserver_class = "cherrypy._cpwsgi_server.CPWSGIServer" 136 using_apache = False 137 using_wsgi = True 138
139 - def __str__(self):
140 return "Builtin WSGI Server on %s:%s" % (self.host, self.port)
141
142 - def sync_apps(self):
143 """Hook a new WSGI app into the origin server.""" 144 cherrypy.server.httpserver.wsgi_app = self.get_app()
145
146 - def get_app(self, app=None):
147 """Obtain a new (decorated) WSGI app to hook into the origin server.""" 148 if app is None: 149 app = cherrypy.tree 150 151 if self.conquer: 152 try: 153 import wsgiconq 154 except ImportError: 155 warnings.warn("Error importing wsgiconq. pyconquer will not run.") 156 else: 157 app = wsgiconq.WSGILogger(app, c_calls=True) 158 159 if self.validate: 160 try: 161 from wsgiref import validate 162 except ImportError: 163 warnings.warn("Error importing wsgiref. The validator will not run.") 164 else: 165 #wraps the app in the validator 166 app = validate.validator(app) 167 168 return app
169 170
171 -def get_cpmodpy_supervisor(**options):
172 from cherrypy.test import modpy 173 sup = modpy.ModPythonSupervisor(**options) 174 sup.template = modpy.conf_cpmodpy 175 return sup
176
177 -def get_modpygw_supervisor(**options):
178 from cherrypy.test import modpy 179 sup = modpy.ModPythonSupervisor(**options) 180 sup.template = modpy.conf_modpython_gateway 181 sup.using_wsgi = True 182 return sup
183
184 -def get_modwsgi_supervisor(**options):
185 from cherrypy.test import modwsgi 186 return modwsgi.ModWSGISupervisor(**options)
187
188 -def get_modfcgid_supervisor(**options):
189 from cherrypy.test import modfcgid 190 return modfcgid.ModFCGISupervisor(**options)
191
192 -def get_modfastcgi_supervisor(**options):
193 from cherrypy.test import modfastcgi 194 return modfastcgi.ModFCGISupervisor(**options)
195
196 -def get_wsgi_u_supervisor(**options):
197 cherrypy.server.wsgi_version = ('u', 0) 198 return LocalWSGISupervisor(**options)
199 200
201 -class CPWebCase(webtest.WebCase):
202 203 script_name = "" 204 scheme = "http" 205 206 available_servers = {'wsgi': LocalWSGISupervisor, 207 'wsgi_u': get_wsgi_u_supervisor, 208 'native': NativeServerSupervisor, 209 'cpmodpy': get_cpmodpy_supervisor, 210 'modpygw': get_modpygw_supervisor, 211 'modwsgi': get_modwsgi_supervisor, 212 'modfcgid': get_modfcgid_supervisor, 213 'modfastcgi': get_modfastcgi_supervisor, 214 } 215 default_server = "wsgi" 216
217 - def _setup_server(cls, supervisor, conf):
218 v = sys.version.split()[0] 219 log.info("Python version used to run this test script: %s" % v) 220 log.info("CherryPy version: %s" % cherrypy.__version__) 221 if supervisor.scheme == "https": 222 ssl = " (ssl)" 223 else: 224 ssl = "" 225 log.info("HTTP server version: %s%s" % (supervisor.protocol, ssl)) 226 log.info("PID: %s" % os.getpid()) 227 228 cherrypy.server.using_apache = supervisor.using_apache 229 cherrypy.server.using_wsgi = supervisor.using_wsgi 230 231 if sys.platform[:4] == 'java': 232 cherrypy.config.update({'server.nodelay': False}) 233 234 if isinstance(conf, basestring): 235 parser = cherrypy.lib.reprconf.Parser() 236 conf = parser.dict_from_file(conf).get('global', {}) 237 else: 238 conf = conf or {} 239 baseconf = conf.copy() 240 baseconf.update({'server.socket_host': supervisor.host, 241 'server.socket_port': supervisor.port, 242 'server.protocol_version': supervisor.protocol, 243 'environment': "test_suite", 244 }) 245 if supervisor.scheme == "https": 246 #baseconf['server.ssl_module'] = 'builtin' 247 baseconf['server.ssl_certificate'] = serverpem 248 baseconf['server.ssl_private_key'] = serverpem 249 250 # helper must be imported lazily so the coverage tool 251 # can run against module-level statements within cherrypy. 252 # Also, we have to do "from cherrypy.test import helper", 253 # exactly like each test module does, because a relative import 254 # would stick a second instance of webtest in sys.modules, 255 # and we wouldn't be able to globally override the port anymore. 256 if supervisor.scheme == "https": 257 webtest.WebCase.HTTP_CONN = HTTPSConnection 258 return baseconf
259 _setup_server = classmethod(_setup_server) 260
261 - def setup_class(cls):
262 '' 263 #Creates a server 264 conf = get_tst_config() 265 supervisor_factory = cls.available_servers.get(conf.get('server', 'wsgi')) 266 if supervisor_factory is None: 267 raise RuntimeError('Unknown server in config: %s' % conf['server']) 268 supervisor = supervisor_factory(**conf) 269 270 #Copied from "run_test_suite" 271 cherrypy.config.reset() 272 baseconf = cls._setup_server(supervisor, conf) 273 cherrypy.config.update(baseconf) 274 setup_client() 275 276 if hasattr(cls, 'setup_server'): 277 # Clear the cherrypy tree and clear the wsgi server so that 278 # it can be updated with the new root 279 cherrypy.tree = cherrypy._cptree.Tree() 280 cherrypy.server.httpserver = None 281 cls.setup_server() 282 # Add a resource for verifying there are no refleaks 283 # to *every* test class. 284 cherrypy.tree.mount(gctools.GCRoot(), '/gc') 285 cls.do_gc_test = True 286 supervisor.start(cls.__module__) 287 288 cls.supervisor = supervisor
289 setup_class = classmethod(setup_class) 290
291 - def teardown_class(cls):
292 '' 293 if hasattr(cls, 'setup_server'): 294 cls.supervisor.stop()
295 teardown_class = classmethod(teardown_class) 296 297 do_gc_test = False 298
299 - def test_gc(self):
300 if self.do_gc_test: 301 self.getPage("/gc/stats") 302 self.assertBody("Statistics:")
303 # Tell nose to run this last in each class. 304 # Prefer sys.maxint for Python 2.3, which didn't have float('inf') 305 test_gc.compat_co_firstlineno = getattr(sys, 'maxint', None) or float('inf') 306
307 - def prefix(self):
308 return self.script_name.rstrip("/")
309
310 - def base(self):
311 if ((self.scheme == "http" and self.PORT == 80) or 312 (self.scheme == "https" and self.PORT == 443)): 313 port = "" 314 else: 315 port = ":%s" % self.PORT 316 317 return "%s://%s%s%s" % (self.scheme, self.HOST, port, 318 self.script_name.rstrip("/"))
319
320 - def exit(self):
321 sys.exit()
322
323 - def getPage(self, url, headers=None, method="GET", body=None, protocol=None):
324 """Open the url. Return status, headers, body.""" 325 if self.script_name: 326 url = httputil.urljoin(self.script_name, url) 327 return webtest.WebCase.getPage(self, url, headers, method, body, protocol)
328
329 - def skip(self, msg='skipped '):
330 raise nose.SkipTest(msg)
331
332 - def assertErrorPage(self, status, message=None, pattern=''):
333 """Compare the response body with a built in error page. 334 335 The function will optionally look for the regexp pattern, 336 within the exception embedded in the error page.""" 337 338 # This will never contain a traceback 339 page = cherrypy._cperror.get_error_page(status, message=message) 340 341 # First, test the response body without checking the traceback. 342 # Stick a match-all group (.*) in to grab the traceback. 343 esc = re.escape 344 epage = esc(page) 345 epage = epage.replace(esc('<pre id="traceback"></pre>'), 346 esc('<pre id="traceback">') + '(.*)' + esc('</pre>')) 347 m = re.match(ntob(epage, self.encoding), self.body, re.DOTALL) 348 if not m: 349 self._handlewebError('Error page does not match; expected:\n' + page) 350 return 351 352 # Now test the pattern against the traceback 353 if pattern is None: 354 # Special-case None to mean that there should be *no* traceback. 355 if m and m.group(1): 356 self._handlewebError('Error page contains traceback') 357 else: 358 if (m is None) or ( 359 not re.search(ntob(re.escape(pattern), self.encoding), 360 m.group(1))): 361 msg = 'Error page does not contain %s in traceback' 362 self._handlewebError(msg % repr(pattern))
363 364 date_tolerance = 2 365
366 - def assertEqualDates(self, dt1, dt2, seconds=None):
367 """Assert abs(dt1 - dt2) is within Y seconds.""" 368 if seconds is None: 369 seconds = self.date_tolerance 370 371 if dt1 > dt2: 372 diff = dt1 - dt2 373 else: 374 diff = dt2 - dt1 375 if not diff < datetime.timedelta(seconds=seconds): 376 raise AssertionError('%r and %r are not within %r seconds.' % 377 (dt1, dt2, seconds))
378 379
380 -def setup_client():
381 """Set up the WebCase classes to match the server's socket settings.""" 382 webtest.WebCase.PORT = cherrypy.server.socket_port 383 webtest.WebCase.HOST = cherrypy.server.socket_host 384 if cherrypy.server.ssl_certificate: 385 CPWebCase.scheme = 'https'
386 387 # --------------------------- Spawning helpers --------------------------- # 388 389
390 -class CPProcess(object):
391 392 pid_file = os.path.join(thisdir, 'test.pid') 393 config_file = os.path.join(thisdir, 'test.conf') 394 config_template = """[global] 395 server.socket_host: '%(host)s' 396 server.socket_port: %(port)s 397 checker.on: False 398 log.screen: False 399 log.error_file: r'%(error_log)s' 400 log.access_file: r'%(access_log)s' 401 %(ssl)s 402 %(extra)s 403 """ 404 error_log = os.path.join(thisdir, 'test.error.log') 405 access_log = os.path.join(thisdir, 'test.access.log') 406
407 - def __init__(self, wait=False, daemonize=False, ssl=False, socket_host=None, socket_port=None):
408 self.wait = wait 409 self.daemonize = daemonize 410 self.ssl = ssl 411 self.host = socket_host or cherrypy.server.socket_host 412 self.port = socket_port or cherrypy.server.socket_port
413
414 - def write_conf(self, extra=""):
415 if self.ssl: 416 serverpem = os.path.join(thisdir, 'test.pem') 417 ssl = """ 418 server.ssl_certificate: r'%s' 419 server.ssl_private_key: r'%s' 420 """ % (serverpem, serverpem) 421 else: 422 ssl = "" 423 424 conf = self.config_template % { 425 'host': self.host, 426 'port': self.port, 427 'error_log': self.error_log, 428 'access_log': self.access_log, 429 'ssl': ssl, 430 'extra': extra, 431 } 432 f = open(self.config_file, 'wb') 433 f.write(ntob(conf, 'utf-8')) 434 f.close()
435
436 - def start(self, imports=None):
437 """Start cherryd in a subprocess.""" 438 cherrypy._cpserver.wait_for_free_port(self.host, self.port) 439 440 args = [sys.executable, '/usr/sbin/cherryd', 441 '-c', self.config_file, '-p', self.pid_file] 442 443 if not isinstance(imports, (list, tuple)): 444 imports = [imports] 445 for i in imports: 446 if i: 447 args.append('-i') 448 args.append(i) 449 450 if self.daemonize: 451 args.append('-d') 452 453 env = os.environ.copy() 454 # Make sure we import the cherrypy package in which this module is defined. 455 grandparentdir = os.path.abspath(os.path.join(thisdir, '..', '..')) 456 if env.get('PYTHONPATH', ''): 457 env['PYTHONPATH'] = os.pathsep.join((grandparentdir, env['PYTHONPATH'])) 458 else: 459 env['PYTHONPATH'] = grandparentdir 460 if self.wait: 461 self.exit_code = os.spawnve(os.P_WAIT, sys.executable, args, env) 462 else: 463 os.spawnve(os.P_NOWAIT, sys.executable, args, env) 464 cherrypy._cpserver.wait_for_occupied_port(self.host, self.port) 465 466 # Give the engine a wee bit more time to finish STARTING 467 if self.daemonize: 468 time.sleep(2) 469 else: 470 time.sleep(1)
471
472 - def get_pid(self):
473 return int(open(self.pid_file, 'rb').read())
474
475 - def join(self):
476 """Wait for the process to exit.""" 477 try: 478 try: 479 # Mac, UNIX 480 os.wait() 481 except AttributeError: 482 # Windows 483 try: 484 pid = self.get_pid() 485 except IOError: 486 # Assume the subprocess deleted the pidfile on shutdown. 487 pass 488 else: 489 os.waitpid(pid, 0) 490 except OSError: 491 x = sys.exc_info()[1] 492 if x.args != (10, 'No child processes'): 493 raise
494