test_request_obj.py (23397B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\lib\cherrypy\test\test_request_obj.py 2 import os 3 localDir = os.path.dirname(__file__) 4 import sys 5 import types 6 from cherrypy._cpcompat import IncompleteRead, ntob, unicodestr 7 import cherrypy 8 from cherrypy import _cptools, tools 9 from cherrypy.lib import httputil 10 defined_http_methods = ('OPTIONS', 'GET', 'HEAD', 'POST', 'PUT', 'DELETE', 'TRACE', 'PROPFIND') 11 from cherrypy.test import helper 12 13 class RequestObjectTests(helper.CPWebCase): 14 15 def setup_server(): 16 17 class Root: 18 19 def index(self): 20 return 'hello' 21 22 index.exposed = True 23 24 def scheme(self): 25 return cherrypy.request.scheme 26 27 scheme.exposed = True 28 29 root = Root() 30 31 class TestType(type): 32 33 def __init__(cls, name, bases, dct): 34 type.__init__(cls, name, bases, dct) 35 for value in dct.values(): 36 if isinstance(value, types.FunctionType): 37 value.exposed = True 38 39 setattr(root, name.lower(), cls()) 40 41 class Test(object): 42 __metaclass__ = TestType 43 44 class Params(Test): 45 46 def index(self, thing): 47 return repr(thing) 48 49 def ismap(self, x, y): 50 return 'Coordinates: %s, %s' % (x, y) 51 52 def default(self, *args, **kwargs): 53 return 'args: %s kwargs: %s' % (args, kwargs) 54 55 default._cp_config = {'request.query_string_encoding': 'latin1'} 56 57 class ParamErrorsCallable(object): 58 exposed = True 59 60 def __call__(self): 61 return 'data' 62 63 class ParamErrors(Test): 64 65 def one_positional(self, param1): 66 return 'data' 67 68 one_positional.exposed = True 69 70 def one_positional_args(self, param1, *args): 71 return 'data' 72 73 one_positional_args.exposed = True 74 75 def one_positional_args_kwargs(self, param1, *args, **kwargs): 76 return 'data' 77 78 one_positional_args_kwargs.exposed = True 79 80 def one_positional_kwargs(self, param1, **kwargs): 81 return 'data' 82 83 one_positional_kwargs.exposed = True 84 85 def no_positional(self): 86 return 'data' 87 88 no_positional.exposed = True 89 90 def no_positional_args(self, *args): 91 return 'data' 92 93 no_positional_args.exposed = True 94 95 def no_positional_args_kwargs(self, *args, **kwargs): 96 return 'data' 97 98 no_positional_args_kwargs.exposed = True 99 100 def no_positional_kwargs(self, **kwargs): 101 return 'data' 102 103 no_positional_kwargs.exposed = True 104 callable_object = ParamErrorsCallable() 105 106 def raise_type_error(self, **kwargs): 107 raise TypeError('Client Error') 108 109 raise_type_error.exposed = True 110 111 def raise_type_error_with_default_param(self, x, y = None): 112 return '%d' % 'a' 113 114 raise_type_error_with_default_param.exposed = True 115 116 def callable_error_page(status, **kwargs): 117 return "Error %s - Well, I'm very sorry but you haven't paid!" % status 118 119 class Error(Test): 120 _cp_config = {'tools.log_tracebacks.on': True} 121 122 def reason_phrase(self): 123 raise cherrypy.HTTPError("410 Gone fishin'") 124 125 def custom(self, err = '404'): 126 raise cherrypy.HTTPError(int(err), 'No, <b>really</b>, not found!') 127 128 custom._cp_config = {'error_page.404': os.path.join(localDir, 'static/index.html'), 129 'error_page.401': callable_error_page} 130 131 def custom_default(self): 132 return 1 + 'a' 133 134 custom_default._cp_config = {'error_page.default': callable_error_page} 135 136 def noexist(self): 137 raise cherrypy.HTTPError(404, 'No, <b>really</b>, not found!') 138 139 noexist._cp_config = {'error_page.404': 'nonexistent.html'} 140 141 def page_method(self): 142 raise ValueError() 143 144 def page_yield(self): 145 yield 'howdy' 146 raise ValueError() 147 148 def page_streamed(self): 149 yield 'word up' 150 raise ValueError() 151 yield 'very oops' 152 153 page_streamed._cp_config = {'response.stream': True} 154 155 def cause_err_in_finalize(self): 156 cherrypy.response.status = 'ZOO OK' 157 158 cause_err_in_finalize._cp_config = {'request.show_tracebacks': False} 159 160 def rethrow(self): 161 raise ValueError() 162 163 rethrow._cp_config = {'request.throw_errors': True} 164 165 class Expect(Test): 166 167 def expectation_failed(self): 168 expect = cherrypy.request.headers.elements('Expect') 169 if expect and expect[0].value != '100-continue': 170 raise cherrypy.HTTPError(400) 171 raise cherrypy.HTTPError(417, 'Expectation Failed') 172 173 class Headers(Test): 174 175 def default(self, headername): 176 return cherrypy.request.headers[headername] 177 178 def doubledheaders(self): 179 hMap = cherrypy.response.headers 180 hMap['content-type'] = 'text/html' 181 hMap['content-length'] = 18 182 hMap['server'] = 'CherryPy headertest' 183 hMap['location'] = '%s://%s:%s/headers/' % (cherrypy.request.local.ip, cherrypy.request.local.port, cherrypy.request.scheme) 184 hMap['Expires'] = 'Thu, 01 Dec 2194 16:00:00 GMT' 185 return 'double header test' 186 187 def ifmatch(self): 188 val = cherrypy.request.headers['If-Match'] 189 cherrypy.response.headers['ETag'] = val 190 return val 191 192 class HeaderElements(Test): 193 194 def get_elements(self, headername): 195 e = cherrypy.request.headers.elements(headername) 196 return '\n'.join([ unicodestr(x) for x in e ]) 197 198 class Method(Test): 199 200 def index(self): 201 m = cherrypy.request.method 202 if m in defined_http_methods or m == 'CONNECT': 203 return m 204 if m == 'LINK': 205 raise cherrypy.HTTPError(405) 206 else: 207 raise cherrypy.HTTPError(501) 208 209 def parameterized(self, data): 210 return data 211 212 def request_body(self): 213 return cherrypy.request.body 214 215 def reachable(self): 216 return 'success' 217 218 class Divorce: 219 documents = {} 220 221 def index(self): 222 yield '<h1>Choose your document</h1>\n' 223 yield '<ul>\n' 224 for id, contents in self.documents.items(): 225 yield " <li><a href='/divorce/get?ID=%s'>%s</a>: %s</li>\n" % (id, id, contents) 226 227 yield '</ul>' 228 229 index.exposed = True 230 231 def get(self, ID): 232 return 'Divorce document %s: %s' % (ID, self.documents.get(ID, 'empty')) 233 234 get.exposed = True 235 236 root.divorce = Divorce() 237 238 class ThreadLocal(Test): 239 240 def index(self): 241 existing = repr(getattr(cherrypy.request, 'asdf', None)) 242 cherrypy.request.asdf = 'rassfrassin' 243 return existing 244 245 appconf = {'/method': {'request.methods_with_bodies': ('POST', 'PUT', 'PROPFIND')}} 246 cherrypy.tree.mount(root, config=appconf) 247 248 setup_server = staticmethod(setup_server) 249 250 def test_scheme(self): 251 self.getPage('/scheme') 252 self.assertBody(self.scheme) 253 254 def testParams(self): 255 self.getPage('/params/?thing=a') 256 self.assertBody("u'a'") 257 self.getPage('/params/?thing=a&thing=b&thing=c') 258 self.assertBody("[u'a', u'b', u'c']") 259 cherrypy.config.update({'request.show_mismatched_params': True}) 260 self.getPage('/params/?notathing=meeting') 261 self.assertInBody('Missing parameters: thing') 262 self.getPage('/params/?thing=meeting¬athing=meeting') 263 self.assertInBody('Unexpected query string parameters: notathing') 264 cherrypy.config.update({'request.show_mismatched_params': False}) 265 self.getPage('/params/?notathing=meeting') 266 self.assertInBody('Not Found') 267 self.getPage('/params/?thing=meeting¬athing=meeting') 268 self.assertInBody('Not Found') 269 self.getPage('/params/%d4%20%e3/cheese?Gruy%E8re=Bulgn%e9ville') 270 self.assertBody("args: ('\\xd4 \\xe3', 'cheese') kwargs: {'Gruy\\xe8re': u'Bulgn\\xe9ville'}") 271 self.getPage('/params/code?url=http%3A//cherrypy.org/index%3Fa%3D1%26b%3D2') 272 self.assertBody("args: ('code',) kwargs: {'url': u'http://cherrypy.org/index?a=1&b=2'}") 273 self.getPage('/params/ismap?223,114') 274 self.assertBody('Coordinates: 223, 114') 275 self.getPage('/params/dictlike?a[1]=1&a[2]=2&b=foo&b[bar]=baz') 276 self.assertBody("args: ('dictlike',) kwargs: {'a[1]': u'1', 'b[bar]': u'baz', 'b': u'foo', 'a[2]': u'2'}") 277 278 def testParamErrors(self): 279 for uri in ('/paramerrors/one_positional?param1=foo', '/paramerrors/one_positional_args?param1=foo', '/paramerrors/one_positional_args/foo', '/paramerrors/one_positional_args/foo/bar/baz', '/paramerrors/one_positional_args_kwargs?param1=foo¶m2=bar', '/paramerrors/one_positional_args_kwargs/foo?param2=bar¶m3=baz', '/paramerrors/one_positional_args_kwargs/foo/bar/baz?param2=bar¶m3=baz', '/paramerrors/one_positional_kwargs?param1=foo¶m2=bar¶m3=baz', '/paramerrors/one_positional_kwargs/foo?param4=foo¶m2=bar¶m3=baz', '/paramerrors/no_positional', '/paramerrors/no_positional_args/foo', '/paramerrors/no_positional_args/foo/bar/baz', '/paramerrors/no_positional_args_kwargs?param1=foo¶m2=bar', '/paramerrors/no_positional_args_kwargs/foo?param2=bar', '/paramerrors/no_positional_args_kwargs/foo/bar/baz?param2=bar¶m3=baz', '/paramerrors/no_positional_kwargs?param1=foo¶m2=bar', '/paramerrors/callable_object'): 280 self.getPage(uri) 281 self.assertStatus(200) 282 283 error_msgs = ['Missing parameters', 284 'Nothing matches the given URI', 285 'Multiple values for parameters', 286 'Unexpected query string parameters', 287 'Unexpected body parameters'] 288 for uri, msg in (('/paramerrors/one_positional', error_msgs[0]), 289 ('/paramerrors/one_positional?foo=foo', error_msgs[0]), 290 ('/paramerrors/one_positional/foo/bar/baz', error_msgs[1]), 291 ('/paramerrors/one_positional/foo?param1=foo', error_msgs[2]), 292 ('/paramerrors/one_positional/foo?param1=foo¶m2=foo', error_msgs[2]), 293 ('/paramerrors/one_positional_args/foo?param1=foo¶m2=foo', error_msgs[2]), 294 ('/paramerrors/one_positional_args/foo/bar/baz?param2=foo', error_msgs[3]), 295 ('/paramerrors/one_positional_args_kwargs/foo/bar/baz?param1=bar¶m3=baz', error_msgs[2]), 296 ('/paramerrors/one_positional_kwargs/foo?param1=foo¶m2=bar¶m3=baz', error_msgs[2]), 297 ('/paramerrors/no_positional/boo', error_msgs[1]), 298 ('/paramerrors/no_positional?param1=foo', error_msgs[3]), 299 ('/paramerrors/no_positional_args/boo?param1=foo', error_msgs[3]), 300 ('/paramerrors/no_positional_kwargs/boo?param1=foo', error_msgs[1]), 301 ('/paramerrors/callable_object?param1=foo', error_msgs[3]), 302 ('/paramerrors/callable_object/boo', error_msgs[1])): 303 for show_mismatched_params in (True, False): 304 cherrypy.config.update({'request.show_mismatched_params': show_mismatched_params}) 305 self.getPage(uri) 306 self.assertStatus(404) 307 if show_mismatched_params: 308 self.assertInBody(msg) 309 else: 310 self.assertInBody('Not Found') 311 312 for uri, body, msg in (('/paramerrors/one_positional/foo', 'param1=foo', error_msgs[2]), 313 ('/paramerrors/one_positional/foo', 'param1=foo¶m2=foo', error_msgs[2]), 314 ('/paramerrors/one_positional_args/foo', 'param1=foo¶m2=foo', error_msgs[2]), 315 ('/paramerrors/one_positional_args/foo/bar/baz', 'param2=foo', error_msgs[4]), 316 ('/paramerrors/one_positional_args_kwargs/foo/bar/baz', 'param1=bar¶m3=baz', error_msgs[2]), 317 ('/paramerrors/one_positional_kwargs/foo', 'param1=foo¶m2=bar¶m3=baz', error_msgs[2]), 318 ('/paramerrors/no_positional', 'param1=foo', error_msgs[4]), 319 ('/paramerrors/no_positional_args/boo', 'param1=foo', error_msgs[4]), 320 ('/paramerrors/callable_object', 'param1=foo', error_msgs[4])): 321 for show_mismatched_params in (True, False): 322 cherrypy.config.update({'request.show_mismatched_params': show_mismatched_params}) 323 self.getPage(uri, method='POST', body=body) 324 self.assertStatus(400) 325 if show_mismatched_params: 326 self.assertInBody(msg) 327 else: 328 self.assertInBody('Bad Request') 329 330 for uri, body, msg in (('/paramerrors/one_positional?param2=foo', 'param1=foo', error_msgs[3]), 331 ('/paramerrors/one_positional/foo/bar', 'param2=foo', error_msgs[1]), 332 ('/paramerrors/one_positional_args/foo/bar?param2=foo', 'param3=foo', error_msgs[3]), 333 ('/paramerrors/one_positional_kwargs/foo/bar', 'param2=bar¶m3=baz', error_msgs[1]), 334 ('/paramerrors/no_positional?param1=foo', 'param2=foo', error_msgs[3]), 335 ('/paramerrors/no_positional_args/boo?param2=foo', 'param1=foo', error_msgs[3]), 336 ('/paramerrors/callable_object?param2=bar', 'param1=foo', error_msgs[3])): 337 for show_mismatched_params in (True, False): 338 cherrypy.config.update({'request.show_mismatched_params': show_mismatched_params}) 339 self.getPage(uri, method='POST', body=body) 340 self.assertStatus(404) 341 if show_mismatched_params: 342 self.assertInBody(msg) 343 else: 344 self.assertInBody('Not Found') 345 346 for uri in ('/paramerrors/raise_type_error', '/paramerrors/raise_type_error_with_default_param?x=0', '/paramerrors/raise_type_error_with_default_param?x=0&y=0'): 347 self.getPage(uri, method='GET') 348 self.assertStatus(500) 349 self.assertTrue('Client Error', self.body) 350 351 def testErrorHandling(self): 352 self.getPage('/error/missing') 353 self.assertStatus(404) 354 self.assertErrorPage(404, "The path '/error/missing' was not found.") 355 ignore = helper.webtest.ignored_exceptions 356 ignore.append(ValueError) 357 try: 358 valerr = '\n raise ValueError()\nValueError' 359 self.getPage('/error/page_method') 360 self.assertErrorPage(500, pattern=valerr) 361 self.getPage('/error/page_yield') 362 self.assertErrorPage(500, pattern=valerr) 363 if cherrypy.server.protocol_version == 'HTTP/1.0' or getattr(cherrypy.server, 'using_apache', False): 364 self.getPage('/error/page_streamed') 365 self.assertStatus(200) 366 self.assertBody('word up') 367 else: 368 self.assertRaises((ValueError, IncompleteRead), self.getPage, '/error/page_streamed') 369 self.getPage('/error/cause_err_in_finalize') 370 msg = "Illegal response status from server ('ZOO' is non-numeric)." 371 self.assertErrorPage(500, msg, None) 372 finally: 373 ignore.pop() 374 375 self.getPage('/error/reason_phrase') 376 self.assertStatus("410 Gone fishin'") 377 self.getPage('/error/custom') 378 self.assertStatus(404) 379 self.assertBody('Hello, world\r\n' + ' ' * 499) 380 self.getPage('/error/custom?err=401') 381 self.assertStatus(401) 382 self.assertBody("Error 401 Unauthorized - Well, I'm very sorry but you haven't paid!") 383 self.getPage('/error/custom_default') 384 self.assertStatus(500) 385 self.assertBody("Error 500 Internal Server Error - Well, I'm very sorry but you haven't paid!".ljust(513)) 386 self.getPage('/error/noexist') 387 self.assertStatus(404) 388 msg = "No, <b>really</b>, not found!<br />In addition, the custom error page failed:\n<br />IOError: [Errno 2] No such file or directory: 'nonexistent.html'" 389 self.assertInBody(msg) 390 if getattr(cherrypy.server, 'using_apache', False): 391 pass 392 else: 393 self.getPage('/error/rethrow') 394 self.assertInBody('raise ValueError()') 395 396 def testExpect(self): 397 e = ('Expect', '100-continue') 398 self.getPage('/headerelements/get_elements?headername=Expect', [e]) 399 self.assertBody('100-continue') 400 self.getPage('/expect/expectation_failed', [e]) 401 self.assertStatus(417) 402 403 def testHeaderElements(self): 404 h = [('Accept', 'audio/*; q=0.2, audio/basic')] 405 self.getPage('/headerelements/get_elements?headername=Accept', h) 406 self.assertStatus(200) 407 self.assertBody('audio/basic\naudio/*;q=0.2') 408 h = [('Accept', 'text/plain; q=0.5, text/html, text/x-dvi; q=0.8, text/x-c')] 409 self.getPage('/headerelements/get_elements?headername=Accept', h) 410 self.assertStatus(200) 411 self.assertBody('text/x-c\ntext/html\ntext/x-dvi;q=0.8\ntext/plain;q=0.5') 412 h = [('Accept', 'text/*, text/html, text/html;level=1, */*')] 413 self.getPage('/headerelements/get_elements?headername=Accept', h) 414 self.assertStatus(200) 415 self.assertBody('text/html;level=1\ntext/html\ntext/*\n*/*') 416 h = [('Accept-Charset', 'iso-8859-5, unicode-1-1;q=0.8')] 417 self.getPage('/headerelements/get_elements?headername=Accept-Charset', h) 418 self.assertStatus('200 OK') 419 self.assertBody('iso-8859-5\nunicode-1-1;q=0.8') 420 h = [('Accept-Encoding', 'gzip;q=1.0, identity; q=0.5, *;q=0')] 421 self.getPage('/headerelements/get_elements?headername=Accept-Encoding', h) 422 self.assertStatus('200 OK') 423 self.assertBody('gzip;q=1.0\nidentity;q=0.5\n*;q=0') 424 h = [('Accept-Language', 'da, en-gb;q=0.8, en;q=0.7')] 425 self.getPage('/headerelements/get_elements?headername=Accept-Language', h) 426 self.assertStatus('200 OK') 427 self.assertBody('da\nen-gb;q=0.8\nen;q=0.7') 428 self.getPage('/headerelements/get_elements?headername=Content-Type', headers=[('Content-Type', 'text/html; charset=utf-8;')]) 429 self.assertStatus(200) 430 self.assertBody('text/html;charset=utf-8') 431 432 def test_repeated_headers(self): 433 self.getPage('/headers/Accept-Charset', headers=[('Accept-Charset', 'iso-8859-5'), ('Accept-Charset', 'unicode-1-1;q=0.8')]) 434 self.assertBody('iso-8859-5, unicode-1-1;q=0.8') 435 self.getPage('/headers/doubledheaders') 436 self.assertBody('double header test') 437 hnames = [ name.title() for name, val in self.headers ] 438 for key in ['Content-Length', 439 'Content-Type', 440 'Date', 441 'Expires', 442 'Location', 443 'Server']: 444 self.assertEqual(hnames.count(key), 1, self.headers) 445 446 def test_encoded_headers(self): 447 self.assertEqual(httputil.decode_TEXT(u'=?utf-8?q?f=C3=BCr?='), u'f\xfcr') 448 if cherrypy.server.protocol_version == 'HTTP/1.1': 449 u = u'\u212bngstr\xf6m' 450 c = u'=E2=84=ABngstr=C3=B6m' 451 self.getPage('/headers/ifmatch', [('If-Match', u'=?utf-8?q?%s?=' % c)]) 452 self.assertBody('\xe2\x84\xabngstr\xc3\xb6m') 453 self.assertHeader('ETag', u'=?utf-8?b?4oSrbmdzdHLDtm0=?=') 454 self.getPage('/headers/ifmatch', [('If-Match', u'=?utf-8?q?%s?=' % (c * 10))]) 455 self.assertBody('\xe2\x84\xabngstr\xc3\xb6m' * 10) 456 etag = self.assertHeader('ETag', '=?utf-8?b?4oSrbmdzdHLDtm3ihKtuZ3N0csO2beKEq25nc3Ryw7Zt4oSrbmdzdHLDtm3ihKtuZ3N0csO2beKEq25nc3Ryw7Zt4oSrbmdzdHLDtm3ihKtuZ3N0csO2beKEq25nc3Ryw7Zt4oSrbmdzdHLDtm0=?=') 457 self.assertEqual(httputil.decode_TEXT(etag), u * 10) 458 459 def test_header_presence(self): 460 self.getPage('/headers/Content-Type', headers=[]) 461 self.assertStatus(500) 462 self.getPage('/headers/Content-Type', headers=[('Content-type', 'application/json')]) 463 self.assertBody('application/json') 464 465 def test_basic_HTTPMethods(self): 466 helper.webtest.methods_with_bodies = ('POST', 'PUT', 'PROPFIND') 467 for m in defined_http_methods: 468 self.getPage('/method/', method=m) 469 if m == 'HEAD': 470 self.assertBody('') 471 elif m == 'TRACE': 472 self.assertEqual(self.body[:5], ntob('TRACE')) 473 else: 474 self.assertBody(m) 475 476 self.getPage('/method/parameterized', method='PUT', body='data=on+top+of+other+things') 477 self.assertBody('on top of other things') 478 b = 'one thing on top of another' 479 h = [('Content-Type', 'text/plain'), ('Content-Length', str(len(b)))] 480 self.getPage('/method/request_body', headers=h, method='PUT', body=b) 481 self.assertStatus(200) 482 self.assertBody(b) 483 b = ntob('one thing on top of another') 484 self.persistent = True 485 try: 486 conn = self.HTTP_CONN 487 conn.putrequest('PUT', '/method/request_body', skip_host=True) 488 conn.putheader('Host', self.HOST) 489 conn.putheader('Content-Length', str(len(b))) 490 conn.endheaders() 491 conn.send(b) 492 response = conn.response_class(conn.sock, method='PUT') 493 response.begin() 494 self.assertEqual(response.status, 200) 495 self.body = response.read() 496 self.assertBody(b) 497 finally: 498 self.persistent = False 499 500 h = [('Content-Type', 'text/plain')] 501 self.getPage('/method/reachable', headers=h, method='PUT') 502 self.assertStatus(411) 503 b = '<?xml version="1.0" encoding="utf-8" ?>\n\n<propfind xmlns="DAV:"><prop><getlastmodified/></prop></propfind>' 504 h = [('Content-Type', 'text/xml'), ('Content-Length', str(len(b)))] 505 self.getPage('/method/request_body', headers=h, method='PROPFIND', body=b) 506 self.assertStatus(200) 507 self.assertBody(b) 508 self.getPage('/method/', method='LINK') 509 self.assertStatus(405) 510 self.getPage('/method/', method='SEARCH') 511 self.assertStatus(501) 512 self.getPage('/divorce/get?ID=13') 513 self.assertBody('Divorce document 13: empty') 514 self.assertStatus(200) 515 self.getPage('/divorce/', method='GET') 516 self.assertBody('<h1>Choose your document</h1>\n<ul>\n</ul>') 517 self.assertStatus(200) 518 519 def test_CONNECT_method(self): 520 if getattr(cherrypy.server, 'using_apache', False): 521 return self.skip('skipped due to known Apache differences... ') 522 self.getPage('/method/', method='CONNECT') 523 self.assertBody('CONNECT') 524 525 def testEmptyThreadlocals(self): 526 results = [] 527 for x in range(20): 528 self.getPage('/threadlocal/') 529 results.append(self.body) 530 531 self.assertEqual(results, [ntob('None')] * 20)