helper.py (12727B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\lib\cherrypy\test\helper.py 2 import datetime 3 import logging 4 log = logging.getLogger(__name__) 5 import os 6 thisdir = os.path.abspath(os.path.dirname(__file__)) 7 serverpem = os.path.join(os.getcwd(), thisdir, 'test.pem') 8 import re 9 import sys 10 import time 11 import warnings 12 import cherrypy 13 from cherrypy._cpcompat import basestring, copyitems, HTTPSConnection, ntob 14 from cherrypy.lib import httputil 15 from cherrypy.lib.reprconf import unrepr 16 from cherrypy.test import webtest 17 import nose 18 _testconfig = None 19 20 def get_tst_config(overconf = {}): 21 global _testconfig 22 if _testconfig is None: 23 conf = {'scheme': 'http', 24 'protocol': 'HTTP/1.1', 25 'port': 8080, 26 'host': '127.0.0.1', 27 'validate': False, 28 'conquer': False, 29 'server': 'wsgi'} 30 try: 31 import testconfig 32 _conf = testconfig.config.get('supervisor', None) 33 if _conf is not None: 34 for k, v in _conf.items(): 35 if isinstance(v, basestring): 36 _conf[k] = unrepr(v) 37 38 conf.update(_conf) 39 except ImportError: 40 pass 41 42 _testconfig = conf 43 conf = _testconfig.copy() 44 conf.update(overconf) 45 return conf 46 47 48 class Supervisor(object): 49 50 def __init__(self, **kwargs): 51 for k, v in kwargs.items(): 52 if k == 'port': 53 setattr(self, k, int(v)) 54 setattr(self, k, v) 55 56 57 log_to_stderr = lambda msg, level: sys.stderr.write(msg + os.linesep) 58 59 class LocalSupervisor(Supervisor): 60 using_apache = False 61 using_wsgi = False 62 63 def __init__(self, **kwargs): 64 for k, v in kwargs.items(): 65 setattr(self, k, v) 66 67 cherrypy.server.httpserver = self.httpserver_class 68 engine = cherrypy.engine 69 if hasattr(engine, 'signal_handler'): 70 engine.signal_handler.subscribe() 71 if hasattr(engine, 'console_control_handler'): 72 engine.console_control_handler.subscribe() 73 74 def start(self, modulename = None): 75 if modulename: 76 cherrypy.server.httpserver = None 77 cherrypy.engine.start() 78 self.sync_apps() 79 80 def sync_apps(self): 81 pass 82 83 def stop(self): 84 td = getattr(self, 'teardown', None) 85 if td: 86 td() 87 cherrypy.engine.exit() 88 for name, server in copyitems(getattr(cherrypy, 'servers', {})): 89 server.unsubscribe() 90 del cherrypy.servers[name] 91 92 93 class NativeServerSupervisor(LocalSupervisor): 94 httpserver_class = 'cherrypy._cpnative_server.CPHTTPServer' 95 using_apache = False 96 using_wsgi = False 97 98 def __str__(self): 99 return 'Builtin HTTP Server on %s:%s' % (self.host, self.port) 100 101 102 class LocalWSGISupervisor(LocalSupervisor): 103 httpserver_class = 'cherrypy._cpwsgi_server.CPWSGIServer' 104 using_apache = False 105 using_wsgi = True 106 107 def __str__(self): 108 return 'Builtin WSGI Server on %s:%s' % (self.host, self.port) 109 110 def sync_apps(self): 111 cherrypy.server.httpserver.wsgi_app = self.get_app() 112 113 def get_app(self, app = None): 114 if app is None: 115 app = cherrypy.tree 116 if self.conquer: 117 try: 118 import wsgiconq 119 except ImportError: 120 warnings.warn('Error importing wsgiconq. pyconquer will not run.') 121 else: 122 app = wsgiconq.WSGILogger(app, c_calls=True) 123 124 if self.validate: 125 try: 126 from wsgiref import validate 127 except ImportError: 128 warnings.warn('Error importing wsgiref. The validator will not run.') 129 else: 130 app = validate.validator(app) 131 132 return app 133 134 135 def get_cpmodpy_supervisor(**options): 136 from cherrypy.test import modpy 137 sup = modpy.ModPythonSupervisor(**options) 138 sup.template = modpy.conf_cpmodpy 139 return sup 140 141 142 def get_modpygw_supervisor(**options): 143 from cherrypy.test import modpy 144 sup = modpy.ModPythonSupervisor(**options) 145 sup.template = modpy.conf_modpython_gateway 146 sup.using_wsgi = True 147 return sup 148 149 150 def get_modwsgi_supervisor(**options): 151 from cherrypy.test import modwsgi 152 return modwsgi.ModWSGISupervisor(**options) 153 154 155 def get_modfcgid_supervisor(**options): 156 from cherrypy.test import modfcgid 157 return modfcgid.ModFCGISupervisor(**options) 158 159 160 def get_modfastcgi_supervisor(**options): 161 from cherrypy.test import modfastcgi 162 return modfastcgi.ModFCGISupervisor(**options) 163 164 165 def get_wsgi_u_supervisor(**options): 166 cherrypy.server.wsgi_version = ('u', 0) 167 return LocalWSGISupervisor(**options) 168 169 170 class CPWebCase(webtest.WebCase): 171 script_name = '' 172 scheme = 'http' 173 available_servers = {'wsgi': LocalWSGISupervisor, 174 'wsgi_u': get_wsgi_u_supervisor, 175 'native': NativeServerSupervisor, 176 'cpmodpy': get_cpmodpy_supervisor, 177 'modpygw': get_modpygw_supervisor, 178 'modwsgi': get_modwsgi_supervisor, 179 'modfcgid': get_modfcgid_supervisor, 180 'modfastcgi': get_modfastcgi_supervisor} 181 default_server = 'wsgi' 182 183 def _setup_server(cls, supervisor, conf): 184 v = sys.version.split()[0] 185 log.info('Python version used to run this test script: %s' % v) 186 log.info('CherryPy version: %s' % cherrypy.__version__) 187 if supervisor.scheme == 'https': 188 ssl = ' (ssl)' 189 else: 190 ssl = '' 191 log.info('HTTP server version: %s%s' % (supervisor.protocol, ssl)) 192 log.info('PID: %s' % os.getpid()) 193 cherrypy.server.using_apache = supervisor.using_apache 194 cherrypy.server.using_wsgi = supervisor.using_wsgi 195 if sys.platform[:4] == 'java': 196 cherrypy.config.update({'server.nodelay': False}) 197 if isinstance(conf, basestring): 198 parser = cherrypy.lib.reprconf.Parser() 199 conf = parser.dict_from_file(conf).get('global', {}) 200 else: 201 conf = conf or {} 202 baseconf = conf.copy() 203 baseconf.update({'server.socket_host': supervisor.host, 204 'server.socket_port': supervisor.port, 205 'server.protocol_version': supervisor.protocol, 206 'environment': 'test_suite'}) 207 if supervisor.scheme == 'https': 208 baseconf['server.ssl_certificate'] = serverpem 209 baseconf['server.ssl_private_key'] = serverpem 210 if supervisor.scheme == 'https': 211 webtest.WebCase.HTTP_CONN = HTTPSConnection 212 return baseconf 213 214 _setup_server = classmethod(_setup_server) 215 216 def setup_class(cls): 217 conf = get_tst_config() 218 supervisor_factory = cls.available_servers.get(conf.get('server', 'wsgi')) 219 if supervisor_factory is None: 220 raise RuntimeError('Unknown server in config: %s' % conf['server']) 221 supervisor = supervisor_factory(**conf) 222 cherrypy.config.reset() 223 baseconf = cls._setup_server(supervisor, conf) 224 cherrypy.config.update(baseconf) 225 setup_client() 226 if hasattr(cls, 'setup_server'): 227 cherrypy.tree = cherrypy._cptree.Tree() 228 cherrypy.server.httpserver = None 229 cls.setup_server() 230 supervisor.start(cls.__module__) 231 cls.supervisor = supervisor 232 233 setup_class = classmethod(setup_class) 234 235 def teardown_class(cls): 236 if hasattr(cls, 'setup_server'): 237 cls.supervisor.stop() 238 239 teardown_class = classmethod(teardown_class) 240 241 def prefix(self): 242 return self.script_name.rstrip('/') 243 244 def base(self): 245 if self.scheme == 'http' and self.PORT == 80 or self.scheme == 'https' and self.PORT == 443: 246 port = '' 247 else: 248 port = ':%s' % self.PORT 249 return '%s://%s%s%s' % (self.scheme, 250 self.HOST, 251 port, 252 self.script_name.rstrip('/')) 253 254 def exit(self): 255 sys.exit() 256 257 def getPage(self, url, headers = None, method = 'GET', body = None, protocol = None): 258 if self.script_name: 259 url = httputil.urljoin(self.script_name, url) 260 return webtest.WebCase.getPage(self, url, headers, method, body, protocol) 261 262 def skip(self, msg = 'skipped '): 263 raise nose.SkipTest(msg) 264 265 def assertErrorPage(self, status, message = None, pattern = ''): 266 page = cherrypy._cperror.get_error_page(status, message=message) 267 esc = re.escape 268 epage = esc(page) 269 epage = epage.replace(esc('<pre id="traceback"></pre>'), esc('<pre id="traceback">') + '(.*)' + esc('</pre>')) 270 m = re.match(ntob(epage, self.encoding), self.body, re.DOTALL) 271 if not m: 272 self._handlewebError('Error page does not match; expected:\n' + page) 273 return 274 if pattern is None: 275 if m and m.group(1): 276 self._handlewebError('Error page contains traceback') 277 elif m is None or not re.search(ntob(re.escape(pattern), self.encoding), m.group(1)): 278 msg = 'Error page does not contain %s in traceback' 279 self._handlewebError(msg % repr(pattern)) 280 281 date_tolerance = 2 282 283 def assertEqualDates(self, dt1, dt2, seconds = None): 284 if seconds is None: 285 seconds = self.date_tolerance 286 if dt1 > dt2: 287 diff = dt1 - dt2 288 else: 289 diff = dt2 - dt1 290 if not diff < datetime.timedelta(seconds=seconds): 291 raise AssertionError('%r and %r are not within %r seconds.' % (dt1, dt2, seconds)) 292 293 294 def setup_client(): 295 webtest.WebCase.PORT = cherrypy.server.socket_port 296 webtest.WebCase.HOST = cherrypy.server.socket_host 297 if cherrypy.server.ssl_certificate: 298 CPWebCase.scheme = 'https' 299 300 301 class CPProcess(object): 302 pid_file = os.path.join(thisdir, 'test.pid') 303 config_file = os.path.join(thisdir, 'test.conf') 304 config_template = "[global]\nserver.socket_host: '%(host)s'\nserver.socket_port: %(port)s\nchecker.on: False\nlog.screen: False\nlog.error_file: r'%(error_log)s'\nlog.access_file: r'%(access_log)s'\n%(ssl)s\n%(extra)s\n" 305 error_log = os.path.join(thisdir, 'test.error.log') 306 access_log = os.path.join(thisdir, 'test.access.log') 307 308 def __init__(self, wait = False, daemonize = False, ssl = False, socket_host = None, socket_port = None): 309 self.wait = wait 310 self.daemonize = daemonize 311 self.ssl = ssl 312 self.host = socket_host or cherrypy.server.socket_host 313 self.port = socket_port or cherrypy.server.socket_port 314 315 def write_conf(self, extra = ''): 316 if self.ssl: 317 serverpem = os.path.join(thisdir, 'test.pem') 318 ssl = "\nserver.ssl_certificate: r'%s'\nserver.ssl_private_key: r'%s'\n" % (serverpem, serverpem) 319 else: 320 ssl = '' 321 conf = self.config_template % {'host': self.host, 322 'port': self.port, 323 'error_log': self.error_log, 324 'access_log': self.access_log, 325 'ssl': ssl, 326 'extra': extra} 327 f = open(self.config_file, 'wb') 328 f.write(ntob(conf, 'utf-8')) 329 f.close() 330 331 def start(self, imports = None): 332 cherrypy._cpserver.wait_for_free_port(self.host, self.port) 333 args = [sys.executable, 334 os.path.join(thisdir, '..', 'cherryd'), 335 '-c', 336 self.config_file, 337 '-p', 338 self.pid_file] 339 if not isinstance(imports, (list, tuple)): 340 imports = [imports] 341 for i in imports: 342 if i: 343 args.append('-i') 344 args.append(i) 345 346 if self.daemonize: 347 args.append('-d') 348 env = os.environ.copy() 349 grandparentdir = os.path.abspath(os.path.join(thisdir, '..', '..')) 350 if env.get('PYTHONPATH', ''): 351 env['PYTHONPATH'] = os.pathsep.join((grandparentdir, env['PYTHONPATH'])) 352 else: 353 env['PYTHONPATH'] = grandparentdir 354 if self.wait: 355 self.exit_code = os.spawnve(os.P_WAIT, sys.executable, args, env) 356 else: 357 os.spawnve(os.P_NOWAIT, sys.executable, args, env) 358 cherrypy._cpserver.wait_for_occupied_port(self.host, self.port) 359 if self.daemonize: 360 time.sleep(2) 361 else: 362 time.sleep(1) 363 364 def get_pid(self): 365 return int(open(self.pid_file, 'rb').read()) 366 367 def join(self): 368 try: 369 try: 370 os.wait() 371 except AttributeError: 372 try: 373 pid = self.get_pid() 374 except IOError: 375 pass 376 else: 377 os.waitpid(pid, 0) 378 379 except OSError: 380 x = sys.exc_info()[1] 381 if x.args != (10, 'No child processes'): 382 raise