test_tools.py (11284B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\lib\cherrypy\test\test_tools.py 2 import gzip 3 import sys 4 from cherrypy._cpcompat import BytesIO, copyitems, itervalues, IncompleteRead, ntob, ntou, xrange 5 import time 6 timeout = 0.2 7 import types 8 import cherrypy 9 from cherrypy import tools 10 europoundUnicode = ntou('\x80\xa3') 11 from cherrypy.test import helper 12 13 class ToolTests(helper.CPWebCase): 14 15 def setup_server(): 16 myauthtools = cherrypy._cptools.Toolbox('myauth') 17 18 def check_access(default = False): 19 if not getattr(cherrypy.request, 'userid', default): 20 raise cherrypy.HTTPError(401) 21 22 myauthtools.check_access = cherrypy.Tool('before_request_body', check_access) 23 24 def numerify(): 25 26 def number_it(body): 27 for chunk in body: 28 for k, v in cherrypy.request.numerify_map: 29 chunk = chunk.replace(k, v) 30 31 yield chunk 32 33 cherrypy.response.body = number_it(cherrypy.response.body) 34 35 class NumTool(cherrypy.Tool): 36 37 def _setup(self): 38 39 def makemap(): 40 m = self._merged_args().get('map', {}) 41 cherrypy.request.numerify_map = copyitems(m) 42 43 cherrypy.request.hooks.attach('on_start_resource', makemap) 44 45 def critical(): 46 cherrypy.request.error_response = cherrypy.HTTPError(502).set_response 47 48 critical.failsafe = True 49 cherrypy.request.hooks.attach('on_start_resource', critical) 50 cherrypy.request.hooks.attach(self._point, self.callable) 51 52 tools.numerify = NumTool('before_finalize', numerify) 53 54 class NadsatTool: 55 56 def __init__(self): 57 self.ended = {} 58 self._name = 'nadsat' 59 60 def nadsat(self): 61 62 def nadsat_it_up(body): 63 for chunk in body: 64 chunk = chunk.replace(ntob('good'), ntob('horrorshow')) 65 chunk = chunk.replace(ntob('piece'), ntob('lomtick')) 66 yield chunk 67 68 cherrypy.response.body = nadsat_it_up(cherrypy.response.body) 69 70 nadsat.priority = 0 71 72 def cleanup(self): 73 cherrypy.response.body = [ntob('razdrez')] 74 id = cherrypy.request.params.get('id') 75 if id: 76 self.ended[id] = True 77 78 cleanup.failsafe = True 79 80 def _setup(self): 81 cherrypy.request.hooks.attach('before_finalize', self.nadsat) 82 cherrypy.request.hooks.attach('on_end_request', self.cleanup) 83 84 tools.nadsat = NadsatTool() 85 86 def pipe_body(): 87 cherrypy.request.process_request_body = False 88 clen = int(cherrypy.request.headers['Content-Length']) 89 cherrypy.request.body = cherrypy.request.rfile.read(clen) 90 91 class Rotator(object): 92 93 def __call__(self, scale): 94 r = cherrypy.response 95 r.collapse_body() 96 r.body = [ chr((ord(x) + scale) % 256) for x in r.body[0] ] 97 98 cherrypy.tools.rotator = cherrypy.Tool('before_finalize', Rotator()) 99 100 def stream_handler(next_handler, *args, **kwargs): 101 cherrypy.response.output = o = BytesIO() 102 try: 103 response = next_handler(*args, **kwargs) 104 return o.getvalue() 105 finally: 106 o.close() 107 108 cherrypy.tools.streamer = cherrypy._cptools.HandlerWrapperTool(stream_handler) 109 110 class Root: 111 112 def index(self): 113 return 'Howdy earth!' 114 115 index.exposed = True 116 117 def tarfile(self): 118 cherrypy.response.output.write(ntob('I am ')) 119 cherrypy.response.output.write(ntob('a tarfile')) 120 121 tarfile.exposed = True 122 tarfile._cp_config = {'tools.streamer.on': True} 123 124 def euro(self): 125 hooks = list(cherrypy.request.hooks['before_finalize']) 126 hooks.sort() 127 cbnames = [ x.callback.__name__ for x in hooks ] 128 priorities = [ x.priority for x in hooks ] 129 yield ntou('Hello,') 130 yield ntou('world') 131 yield europoundUnicode 132 133 euro.exposed = True 134 135 def pipe(self): 136 return cherrypy.request.body 137 138 pipe.exposed = True 139 pipe._cp_config = {'hooks.before_request_body': pipe_body} 140 141 def decorated_euro(self, *vpath): 142 yield ntou('Hello,') 143 yield ntou('world') 144 yield europoundUnicode 145 146 decorated_euro.exposed = True 147 decorated_euro = tools.gzip(compress_level=6)(decorated_euro) 148 decorated_euro = tools.rotator(scale=3)(decorated_euro) 149 150 root = Root() 151 152 class TestType(type): 153 154 def __init__(cls, name, bases, dct): 155 type.__init__(cls, name, bases, dct) 156 for value in itervalues(dct): 157 if isinstance(value, types.FunctionType): 158 value.exposed = True 159 160 setattr(root, name.lower(), cls()) 161 162 class Test(object): 163 __metaclass__ = TestType 164 165 class Demo(Test): 166 _cp_config = {'tools.nadsat.on': True} 167 168 def index(self, id = None): 169 return 'A good piece of cherry pie' 170 171 def ended(self, id): 172 return repr(tools.nadsat.ended[id]) 173 174 def err(self, id = None): 175 raise ValueError() 176 177 def errinstream(self, id = None): 178 yield 'nonconfidential' 179 raise ValueError() 180 yield 'confidential' 181 182 def restricted(self): 183 return 'Welcome!' 184 185 restricted = myauthtools.check_access()(restricted) 186 userid = restricted 187 188 def err_in_onstart(self): 189 return 'success!' 190 191 def stream(self, id = None): 192 for x in xrange(100000000): 193 yield str(x) 194 195 stream._cp_config = {'response.stream': True} 196 197 conf = {'/demo': {'tools.numerify.on': True, 198 'tools.numerify.map': {ntob('pie'): ntob('3.14159')}}, 199 '/demo/restricted': {'request.show_tracebacks': False}, 200 '/demo/userid': {'request.show_tracebacks': False, 201 'myauth.check_access.default': True}, 202 '/demo/errinstream': {'response.stream': True}, 203 '/demo/err_in_onstart': {'tools.numerify.map': 'pie->3.14159'}, 204 '/euro': {'tools.gzip.on': True, 205 'tools.encode.on': True}, 206 '/decorated_euro/subpath': {'tools.gzip.priority': 10}, 207 '/tarfile': {'tools.streamer.on': True}} 208 app = cherrypy.tree.mount(root, config=conf) 209 app.request_class.namespaces['myauth'] = myauthtools 210 if sys.version_info >= (2, 5): 211 from cherrypy.test import _test_decorators 212 root.tooldecs = _test_decorators.ToolExamples() 213 214 setup_server = staticmethod(setup_server) 215 216 def testHookErrors(self): 217 self.getPage('/demo/?id=1') 218 self.assertBody('A horrorshow lomtick of cherry 3.14159') 219 time.sleep(0.1) 220 self.getPage('/demo/ended/1') 221 self.assertBody('True') 222 valerr = '\n raise ValueError()\nValueError' 223 self.getPage('/demo/err?id=3') 224 self.assertErrorPage(502, pattern=valerr) 225 time.sleep(0.1) 226 self.getPage('/demo/ended/3') 227 self.assertBody('True') 228 if cherrypy.server.protocol_version == 'HTTP/1.0' or getattr(cherrypy.server, 'using_apache', False): 229 self.getPage('/demo/errinstream?id=5') 230 self.assertStatus('200 OK') 231 self.assertBody('nonconfidential') 232 else: 233 self.assertRaises((ValueError, IncompleteRead), self.getPage, '/demo/errinstream?id=5') 234 time.sleep(0.1) 235 self.getPage('/demo/ended/5') 236 self.assertBody('True') 237 self.getPage('/demo/restricted') 238 self.assertErrorPage(401) 239 self.getPage('/demo/userid') 240 self.assertBody('Welcome!') 241 242 def testEndRequestOnDrop(self): 243 old_timeout = None 244 try: 245 httpserver = cherrypy.server.httpserver 246 old_timeout = httpserver.timeout 247 except (AttributeError, IndexError): 248 return self.skip() 249 250 try: 251 httpserver.timeout = timeout 252 self.persistent = True 253 try: 254 conn = self.HTTP_CONN 255 conn.putrequest('GET', '/demo/stream?id=9', skip_host=True) 256 conn.putheader('Host', self.HOST) 257 conn.endheaders() 258 finally: 259 self.persistent = False 260 261 time.sleep(timeout * 2) 262 self.getPage('/demo/ended/9') 263 self.assertBody('True') 264 finally: 265 if old_timeout is not None: 266 httpserver.timeout = old_timeout 267 268 def testGuaranteedHooks(self): 269 self.getPage('/demo/err_in_onstart') 270 self.assertErrorPage(502) 271 self.assertInBody("AttributeError: 'str' object has no attribute 'items'") 272 273 def testCombinedTools(self): 274 expectedResult = (ntou('Hello,world') + europoundUnicode).encode('utf-8') 275 zbuf = BytesIO() 276 zfile = gzip.GzipFile(mode='wb', fileobj=zbuf, compresslevel=9) 277 zfile.write(expectedResult) 278 zfile.close() 279 self.getPage('/euro', headers=[('Accept-Encoding', 'gzip'), ('Accept-Charset', 'ISO-8859-1,utf-8;q=0.7,*;q=0.7')]) 280 self.assertInBody(zbuf.getvalue()[:3]) 281 zbuf = BytesIO() 282 zfile = gzip.GzipFile(mode='wb', fileobj=zbuf, compresslevel=6) 283 zfile.write(expectedResult) 284 zfile.close() 285 self.getPage('/decorated_euro', headers=[('Accept-Encoding', 'gzip')]) 286 self.assertInBody(zbuf.getvalue()[:3]) 287 self.getPage('/decorated_euro/subpath', headers=[('Accept-Encoding', 'gzip')]) 288 self.assertInBody(''.join([ chr((ord(x) + 3) % 256) for x in zbuf.getvalue() ])) 289 290 def testBareHooks(self): 291 content = 'bit of a pain in me gulliver' 292 self.getPage('/pipe', headers=[('Content-Length', str(len(content))), ('Content-Type', 'text/plain')], method='POST', body=content) 293 self.assertBody(content) 294 295 def testHandlerWrapperTool(self): 296 self.getPage('/tarfile') 297 self.assertBody('I am a tarfile') 298 299 def testToolWithConfig(self): 300 if not sys.version_info >= (2, 5): 301 return self.skip('skipped (Python 2.5+ only)') 302 self.getPage('/tooldecs/blah') 303 self.assertHeader('Content-Type', 'application/data') 304 305 def testWarnToolOn(self): 306 try: 307 numon = cherrypy.tools.numerify.on 308 except AttributeError: 309 pass 310 else: 311 raise AssertionError('Tool.on did not error as it should have.') 312 313 try: 314 cherrypy.tools.numerify.on = True 315 except AttributeError: 316 pass 317 else: 318 raise AssertionError('Tool.on did not error as it should have.')