# -*- coding: utf-8 -*- ''' Tests for the BaseRequest and BaseResponse objects and their subclasses. ''' import unittest import sys import bottle from bottle import request, tob, touni, tonat, json_dumps, _e, HTTPError, parse_date import tools import wsgiref.util import base64 from bottle import BaseRequest, BaseResponse, LocalRequest try: from itertools import product except ImportError: def product(*args): pools = map(tuple, args) result = [[]] for pool in pools: result = [x + [y] for x in result for y in pool] for prod in result: yield tuple(prod) class TestRequest(unittest.TestCase): def test_app_property(self): e = {} r = BaseRequest(e) self.assertRaises(RuntimeError, lambda: r.app) e.update({'bottle.app': 5}) self.assertEqual(r.app, 5) def test_route_property(self): e = {'bottle.route': 5} r = BaseRequest(e) self.assertEqual(r.route, 5) def test_url_for_property(self): e = {} r = BaseRequest(e) self.assertRaises(RuntimeError, lambda: r.url_args) e.update({'route.url_args': {'a': 5}}) self.assertEqual(r.url_args, {'a': 5}) def test_path(self): """ PATH_INFO normalization. """ # Legal paths tests = [('', '/'), ('x','/x'), ('x/', '/x/'), ('/x', '/x'), ('/x/', '/x/')] for raw, norm in tests: self.assertEqual(norm, BaseRequest({'PATH_INFO': raw}).path) # Strange paths tests = [('///', '/'), ('//x','/x')] for raw, norm in tests: self.assertEqual(norm, BaseRequest({'PATH_INFO': raw}).path) # No path at all self.assertEqual('/', BaseRequest({}).path) def test_method(self): self.assertEqual(BaseRequest({}).method, 'GET') self.assertEqual(BaseRequest({'REQUEST_METHOD':'GET'}).method, 'GET') self.assertEqual(BaseRequest({'REQUEST_METHOD':'GeT'}).method, 'GET') self.assertEqual(BaseRequest({'REQUEST_METHOD':'get'}).method, 'GET') self.assertEqual(BaseRequest({'REQUEST_METHOD':'POst'}).method, 'POST') self.assertEqual(BaseRequest({'REQUEST_METHOD':'FanTASY'}).method, 'FANTASY') def test_script_name(self): """ SCRIPT_NAME normalization. """ # Legal paths tests = [('', '/'), ('x','/x/'), ('x/', '/x/'), ('/x', '/x/'), ('/x/', '/x/')] for raw, norm in tests: self.assertEqual(norm, BaseRequest({'SCRIPT_NAME': raw}).script_name) # Strange paths tests = [('///', '/'), ('///x///','/x/')] for raw, norm in tests: self.assertEqual(norm, BaseRequest({'SCRIPT_NAME': raw}).script_name) # No path at all self.assertEqual('/', BaseRequest({}).script_name) def test_pathshift(self): """ Request.path_shift() """ def test_shift(s, p, c): request = BaseRequest({'SCRIPT_NAME': s, 'PATH_INFO': p}) request.path_shift(c) return [request['SCRIPT_NAME'], request.path] self.assertEqual(['/a/b', '/c/d'], test_shift('/a/b', '/c/d', 0)) self.assertEqual(['/a/b', '/c/d/'], test_shift('/a/b', '/c/d/', 0)) self.assertEqual(['/a/b/c', '/d'], test_shift('/a/b', '/c/d', 1)) self.assertEqual(['/a', '/b/c/d'], test_shift('/a/b', '/c/d', -1)) self.assertEqual(['/a/b/c', '/d/'], test_shift('/a/b', '/c/d/', 1)) self.assertEqual(['/a', '/b/c/d/'], test_shift('/a/b', '/c/d/', -1)) self.assertEqual(['/a/b/c', '/d/'], test_shift('/a/b/', '/c/d/', 1)) self.assertEqual(['/a', '/b/c/d/'], test_shift('/a/b/', '/c/d/', -1)) self.assertEqual(['/a/b/c/d', '/'], test_shift('/', '/a/b/c/d', 4)) self.assertEqual(['/', '/a/b/c/d/'], test_shift('/a/b/c/d', '/', -4)) self.assertRaises(AssertionError, test_shift, '/a/b', '/c/d', 3) self.assertRaises(AssertionError, test_shift, '/a/b', '/c/d', -3) def test_url(self): """ Environ: URL building """ request = BaseRequest({'HTTP_HOST':'example.com'}) self.assertEqual('http://example.com/', request.url) request = BaseRequest({'SERVER_NAME':'example.com'}) self.assertEqual('http://example.com/', request.url) request = BaseRequest({'SERVER_NAME':'example.com', 'SERVER_PORT':'81'}) self.assertEqual('http://example.com:81/', request.url) request = BaseRequest({'wsgi.url_scheme':'https', 'SERVER_NAME':'example.com'}) self.assertEqual('https://example.com/', request.url) request = BaseRequest({'HTTP_HOST':'example.com', 'PATH_INFO':'/path', 'QUERY_STRING':'1=b&c=d', 'SCRIPT_NAME':'/sp'}) self.assertEqual('http://example.com/sp/path?1=b&c=d', request.url) request = BaseRequest({'HTTP_HOST':'example.com', 'PATH_INFO':'/pa th', 'SCRIPT_NAME':'/s p'}) self.assertEqual('http://example.com/s%20p/pa%20th', request.url) def test_dict_access(self): """ Environ: request objects are environment dicts """ e = {} wsgiref.util.setup_testing_defaults(e) request = BaseRequest(e) self.assertEqual(list(request), list(e.keys())) self.assertEqual(len(request), len(e)) for k, v in e.items(): self.assertTrue(k in request) self.assertEqual(request[k], v) request[k] = 'test' self.assertEqual(request[k], 'test') del request['PATH_INFO'] self.assertTrue('PATH_INFO' not in request) def test_readonly_environ(self): request = BaseRequest({'bottle.request.readonly':True}) def test(): request['x']='y' self.assertRaises(KeyError, test) def test_header_access(self): """ Environ: Request objects decode headers """ e = {} wsgiref.util.setup_testing_defaults(e) e['HTTP_SOME_HEADER'] = 'some value' request = BaseRequest(e) request['HTTP_SOME_OTHER_HEADER'] = 'some other value' self.assertTrue('Some-Header' in request.headers) self.assertTrue(request.headers['Some-Header'] == 'some value') self.assertTrue(request.headers['Some-Other-Header'] == 'some other value') def test_header_access_special(self): e = {} wsgiref.util.setup_testing_defaults(e) request = BaseRequest(e) request['CONTENT_TYPE'] = 'test' request['CONTENT_LENGTH'] = '123' self.assertEqual(request.headers['Content-Type'], 'test') self.assertEqual(request.headers['Content-Length'], '123') def test_cookie_dict(self): """ Environ: Cookie dict """ t = dict() t['a=a'] = {'a': 'a'} t['a=a; b=b'] = {'a': 'a', 'b':'b'} t['a=a; a=b'] = {'a': 'b'} for k, v in t.items(): request = BaseRequest({'HTTP_COOKIE': k}) for n in v: self.assertEqual(v[n], request.cookies[n]) self.assertEqual(v[n], request.get_cookie(n)) def test_get(self): """ Environ: GET data """ qs = tonat(tob('a=a&a=1&b=b&c=c&cn=%e7%93%b6'), 'latin1') request = BaseRequest({'QUERY_STRING':qs}) self.assertTrue('a' in request.query) self.assertTrue('b' in request.query) self.assertEqual(['a','1'], request.query.getall('a')) self.assertEqual(['b'], request.query.getall('b')) self.assertEqual('1', request.query['a']) self.assertEqual('b', request.query['b']) self.assertEqual(tonat(tob('瓶'), 'latin1'), request.query['cn']) self.assertEqual(touni('瓶'), request.query.cn) def test_post(self): """ Environ: POST data """ sq = tob('a=a&a=1&b=b&c=&d&cn=%e7%93%b6') e = {} wsgiref.util.setup_testing_defaults(e) e['wsgi.input'].write(sq) e['wsgi.input'].seek(0) e['CONTENT_LENGTH'] = str(len(sq)) e['REQUEST_METHOD'] = "POST" request = BaseRequest(e) self.assertTrue('a' in request.POST) self.assertTrue('b' in request.POST) self.assertEqual(['a','1'], request.POST.getall('a')) self.assertEqual(['b'], request.POST.getall('b')) self.assertEqual('1', request.POST['a']) self.assertEqual('b', request.POST['b']) self.assertEqual('', request.POST['c']) self.assertEqual('', request.POST['d']) self.assertEqual(tonat(tob('瓶'), 'latin1'), request.POST['cn']) self.assertEqual(touni('瓶'), request.POST.cn) def test_bodypost(self): sq = tob('foobar') e = {} wsgiref.util.setup_testing_defaults(e) e['wsgi.input'].write(sq) e['wsgi.input'].seek(0) e['CONTENT_LENGTH'] = str(len(sq)) e['REQUEST_METHOD'] = "POST" request = BaseRequest(e) self.assertEqual('', request.POST['foobar']) def test_body_noclose(self): """ Test that the body file handler is not closed after request.POST """ sq = tob('a=a&a=1&b=b&c=&d') e = {} wsgiref.util.setup_testing_defaults(e) e['wsgi.input'].write(sq) e['wsgi.input'].seek(0) e['CONTENT_LENGTH'] = str(len(sq)) e['REQUEST_METHOD'] = "POST" request = BaseRequest(e) self.assertEqual(sq, request.body.read()) request.POST # This caused a body.close() with Python 3.x self.assertEqual(sq, request.body.read()) def test_params(self): """ Environ: GET and POST are combined in request.param """ e = {} wsgiref.util.setup_testing_defaults(e) e['wsgi.input'].write(tob('b=b&c=p')) e['wsgi.input'].seek(0) e['CONTENT_LENGTH'] = '7' e['QUERY_STRING'] = 'a=a&c=g' e['REQUEST_METHOD'] = "POST" request = BaseRequest(e) self.assertEqual(['a','b','c'], sorted(request.params.keys())) self.assertEqual('p', request.params['c']) def test_getpostleak(self): """ Environ: GET and POST should not leak into each other """ e = {} wsgiref.util.setup_testing_defaults(e) e['wsgi.input'].write(tob('b=b')) e['wsgi.input'].seek(0) e['CONTENT_LENGTH'] = '3' e['QUERY_STRING'] = 'a=a' e['REQUEST_METHOD'] = "POST" request = BaseRequest(e) self.assertEqual(['a'], list(request.GET.keys())) self.assertEqual(['b'], list(request.POST.keys())) def test_body(self): """ Environ: Request.body should behave like a file object factory """ e = {} wsgiref.util.setup_testing_defaults(e) e['wsgi.input'].write(tob('abc')) e['wsgi.input'].seek(0) e['CONTENT_LENGTH'] = str(3) request = BaseRequest(e) self.assertEqual(tob('abc'), request.body.read()) self.assertEqual(tob('abc'), request.body.read(3)) self.assertEqual(tob('abc'), request.body.readline()) self.assertEqual(tob('abc'), request.body.readline(3)) def test_bigbody(self): """ Environ: Request.body should handle big uploads using files """ e = {} wsgiref.util.setup_testing_defaults(e) e['wsgi.input'].write(tob('x')*1024*1000) e['wsgi.input'].seek(0) e['CONTENT_LENGTH'] = str(1024*1000) request = BaseRequest(e) self.assertTrue(hasattr(request.body, 'fileno')) self.assertEqual(1024*1000, len(request.body.read())) self.assertEqual(1024, len(request.body.read(1024))) self.assertEqual(1024*1000, len(request.body.readline())) self.assertEqual(1024, len(request.body.readline(1024))) def test_tobigbody(self): """ Environ: Request.body should truncate to Content-Length bytes """ e = {} wsgiref.util.setup_testing_defaults(e) e['wsgi.input'].write(tob('x')*1024) e['wsgi.input'].seek(0) e['CONTENT_LENGTH'] = '42' request = BaseRequest(e) self.assertEqual(42, len(request.body.read())) self.assertEqual(42, len(request.body.read(1024))) self.assertEqual(42, len(request.body.readline())) self.assertEqual(42, len(request.body.readline(1024))) def _test_chunked(self, body, expect): e = {} wsgiref.util.setup_testing_defaults(e) e['wsgi.input'].write(tob(body)) e['wsgi.input'].seek(0) e['HTTP_TRANSFER_ENCODING'] = 'chunked' if isinstance(expect, str): self.assertEquals(tob(expect), BaseRequest(e).body.read()) else: self.assertRaises(expect, lambda: BaseRequest(e).body) def test_chunked(self): self._test_chunked('1\r\nx\r\nff\r\n' + 'y'*255 + '\r\n0\r\n', 'x' + 'y'*255) self._test_chunked('8\r\nxxxxxxxx\r\n0\r\n','xxxxxxxx') self._test_chunked('0\r\n', '') def test_chunked_meta_fields(self): self._test_chunked('8 ; foo\r\nxxxxxxxx\r\n0\r\n','xxxxxxxx') self._test_chunked('8;foo\r\nxxxxxxxx\r\n0\r\n','xxxxxxxx') self._test_chunked('8;foo=bar\r\nxxxxxxxx\r\n0\r\n','xxxxxxxx') def test_chunked_not_terminated(self): self._test_chunked('1\r\nx\r\n', HTTPError) def test_chunked_wrong_size(self): self._test_chunked('2\r\nx\r\n', HTTPError) def test_chunked_illegal_size(self): self._test_chunked('x\r\nx\r\n', HTTPError) def test_chunked_not_chunked_at_all(self): self._test_chunked('abcdef', HTTPError) def test_multipart(self): """ Environ: POST (multipart files and multible values per key) """ fields = [('field1','value1'), ('field2','value2'), ('field2','value3')] files = [('file1','filename1.txt','content1'), ('万难','万难foo.py', 'ä\nö\rü')] e = tools.multipart_environ(fields=fields, files=files) request = BaseRequest(e) # File content self.assertTrue('file1' in request.POST) self.assertTrue('file1' in request.files) self.assertTrue('file1' not in request.forms) cmp = tob('content1') if sys.version_info >= (3,2,0) else 'content1' self.assertEqual(cmp, request.POST['file1'].file.read()) # File name and meta data self.assertTrue('万难' in request.POST) self.assertTrue('万难' in request.files) self.assertTrue('万难' not in request.forms) self.assertEqual('foo.py', request.POST['万难'].filename) self.assertTrue(request.files['万难']) self.assertFalse(request.files.file77) # UTF-8 files x = request.POST['万难'].file.read() if (3,2,0) > sys.version_info >= (3,0,0): x = x.encode('utf8') self.assertEqual(tob('ä\nö\rü'), x) # No file self.assertTrue('file3' not in request.POST) self.assertTrue('file3' not in request.files) self.assertTrue('file3' not in request.forms) # Field (single) self.assertEqual('value1', request.POST['field1']) self.assertTrue('field1' not in request.files) self.assertEqual('value1', request.forms['field1']) # Field (multi) self.assertEqual(2, len(request.POST.getall('field2'))) self.assertEqual(['value2', 'value3'], request.POST.getall('field2')) self.assertEqual(['value2', 'value3'], request.forms.getall('field2')) self.assertTrue('field2' not in request.files) def test_json_empty(self): """ Environ: Request.json property with empty body. """ self.assertEqual(BaseRequest({}).json, None) def test_json_noheader(self): """ Environ: Request.json property with missing content-type header. """ test = dict(a=5, b='test', c=[1,2,3]) e = {} wsgiref.util.setup_testing_defaults(e) e['wsgi.input'].write(tob(json_dumps(test))) e['wsgi.input'].seek(0) e['CONTENT_LENGTH'] = str(len(json_dumps(test))) self.assertEqual(BaseRequest(e).json, None) def test_json_tobig(self): """ Environ: Request.json property with huge body. """ test = dict(a=5, tobig='x' * bottle.BaseRequest.MEMFILE_MAX) e = {'CONTENT_TYPE': 'application/json'} wsgiref.util.setup_testing_defaults(e) e['wsgi.input'].write(tob(json_dumps(test))) e['wsgi.input'].seek(0) e['CONTENT_LENGTH'] = str(len(json_dumps(test))) self.assertRaises(HTTPError, lambda: BaseRequest(e).json) def test_json_valid(self): """ Environ: Request.json property. """ test = dict(a=5, b='test', c=[1,2,3]) e = {'CONTENT_TYPE': 'application/json; charset=UTF-8'} wsgiref.util.setup_testing_defaults(e) e['wsgi.input'].write(tob(json_dumps(test))) e['wsgi.input'].seek(0) e['CONTENT_LENGTH'] = str(len(json_dumps(test))) self.assertEqual(BaseRequest(e).json, test) def test_json_forged_header_issue616(self): test = dict(a=5, b='test', c=[1,2,3]) e = {'CONTENT_TYPE': 'text/plain;application/json'} wsgiref.util.setup_testing_defaults(e) e['wsgi.input'].write(tob(json_dumps(test))) e['wsgi.input'].seek(0) e['CONTENT_LENGTH'] = str(len(json_dumps(test))) self.assertEqual(BaseRequest(e).json, None) def test_json_header_empty_body(self): """Request Content-Type is application/json but body is empty""" e = {'CONTENT_TYPE': 'application/json'} wsgiref.util.setup_testing_defaults(e) wsgiref.util.setup_testing_defaults(e) e['CONTENT_LENGTH'] = "0" self.assertEqual(BaseRequest(e).json, None) def test_isajax(self): e = {} wsgiref.util.setup_testing_defaults(e) self.assertFalse(BaseRequest(e.copy()).is_ajax) e['HTTP_X_REQUESTED_WITH'] = 'XMLHttpRequest' self.assertTrue(BaseRequest(e.copy()).is_ajax) def test_auth(self): user, pwd = 'marc', 'secret' basic = touni(base64.b64encode(tob('%s:%s' % (user, pwd)))) r = BaseRequest({}) self.assertEqual(r.auth, None) r.environ['HTTP_AUTHORIZATION'] = 'basic %s' % basic self.assertEqual(r.auth, (user, pwd)) r.environ['REMOTE_USER'] = user self.assertEqual(r.auth, (user, pwd)) del r.environ['HTTP_AUTHORIZATION'] self.assertEqual(r.auth, (user, None)) def test_remote_route(self): ips = ['1.2.3.4', '2.3.4.5', '3.4.5.6'] r = BaseRequest({}) self.assertEqual(r.remote_route, []) r.environ['HTTP_X_FORWARDED_FOR'] = ', '.join(ips) self.assertEqual(r.remote_route, ips) r.environ['REMOTE_ADDR'] = ips[1] self.assertEqual(r.remote_route, ips) del r.environ['HTTP_X_FORWARDED_FOR'] self.assertEqual(r.remote_route, [ips[1]]) def test_remote_addr(self): ips = ['1.2.3.4', '2.3.4.5', '3.4.5.6'] r = BaseRequest({}) self.assertEqual(r.remote_addr, None) r.environ['HTTP_X_FORWARDED_FOR'] = ', '.join(ips) self.assertEqual(r.remote_addr, ips[0]) r.environ['REMOTE_ADDR'] = ips[1] self.assertEqual(r.remote_addr, ips[0]) del r.environ['HTTP_X_FORWARDED_FOR'] self.assertEqual(r.remote_addr, ips[1]) def test_user_defined_attributes(self): for cls in (BaseRequest, LocalRequest): r = cls() # New attributes go to the environ dict. r.foo = 'somevalue' self.assertEqual(r.foo, 'somevalue') self.assertTrue('somevalue' in r.environ.values()) # Unknown attributes raise AttributeError. self.assertRaises(AttributeError, getattr, r, 'somevalue') class TestResponse(unittest.TestCase): def test_constructor_body(self): self.assertEqual('', BaseResponse('').body) self.assertEqual('YAY', BaseResponse('YAY').body) def test_constructor_status(self): self.assertEqual(200, BaseResponse('YAY', 200).status_code) self.assertEqual('200 OK', BaseResponse('YAY', 200).status_line) self.assertEqual('200 YAY', BaseResponse('YAY', '200 YAY').status_line) self.assertEqual('200 YAY', BaseResponse('YAY', '200 YAY').status_line) def test_constructor_headerlist(self): from functools import partial make_res = partial(BaseResponse, '', 200) self.assertTrue('yay', make_res([('x-test','yay')])['x-test']) def test_constructor_headerlist(self): from functools import partial make_res = partial(BaseResponse, '', 200) self.assertTrue('yay', make_res(x_test='yay')['x-test']) def test_set_status(self): rs = BaseResponse() rs.status = 200 self.assertEqual(rs.status, rs.status_line) self.assertEqual(rs.status_code, 200) self.assertEqual(rs.status_line, '200 OK') rs.status = 999 self.assertEqual(rs.status, rs.status_line) self.assertEqual(rs.status_code, 999) self.assertEqual(rs.status_line, '999 Unknown') rs.status = 404 self.assertEqual(rs.status, rs.status_line) self.assertEqual(rs.status_code, 404) self.assertEqual(rs.status_line, '404 Not Found') def test(): rs.status = -200 self.assertRaises(ValueError, test) self.assertEqual(rs.status, rs.status_line) # last value self.assertEqual(rs.status_code, 404) # last value self.assertEqual(rs.status_line, '404 Not Found') # last value def test(): rs.status = 5 self.assertRaises(ValueError, test) self.assertEqual(rs.status, rs.status_line) # last value self.assertEqual(rs.status_code, 404) # last value self.assertEqual(rs.status_line, '404 Not Found') # last value rs.status = '999 Who knows?' # Illegal, but acceptable three digit code self.assertEqual(rs.status, rs.status_line) self.assertEqual(rs.status_code, 999) self.assertEqual(rs.status_line, '999 Who knows?') rs.status = 555 # Strange code self.assertEqual(rs.status, rs.status_line) self.assertEqual(rs.status_code, 555) self.assertEqual(rs.status_line, '555 Unknown') rs.status = '404 Brain not Found' # Custom reason self.assertEqual(rs.status, rs.status_line) self.assertEqual(rs.status_code, 404) self.assertEqual(rs.status_line, '404 Brain not Found') def test(): rs.status = '5 Illegal Code' self.assertRaises(ValueError, test) self.assertEqual(rs.status, rs.status_line) # last value self.assertEqual(rs.status_code, 404) # last value self.assertEqual(rs.status_line, '404 Brain not Found') # last value def test(): rs.status = '-99 Illegal Code' self.assertRaises(ValueError, test) self.assertEqual(rs.status, rs.status_line) # last value self.assertEqual(rs.status_code, 404) # last value self.assertEqual(rs.status_line, '404 Brain not Found') # last value def test(): rs.status = '1000 Illegal Code' self.assertRaises(ValueError, test) self.assertEqual(rs.status, rs.status_line) # last value self.assertEqual(rs.status_code, 404) # last value self.assertEqual(rs.status_line, '404 Brain not Found') # last value def test(): rs.status = '555' # No reason self.assertRaises(ValueError, test) self.assertEqual(rs.status, rs.status_line) # last value self.assertEqual(rs.status_code, 404) # last value self.assertEqual(rs.status_line, '404 Brain not Found') # last value def test_content_type(self): rs = BaseResponse() rs.content_type = 'test/some' self.assertEquals('test/some', rs.headers.get('Content-Type')) def test_charset(self): rs = BaseResponse() self.assertEqual(rs.charset, 'UTF-8') rs.content_type = 'text/html; charset=latin9' self.assertEqual(rs.charset, 'latin9') rs.content_type = 'text/html' self.assertEqual(rs.charset, 'UTF-8') def test_set_cookie(self): r = BaseResponse() r.set_cookie('name1', 'value', max_age=5) r.set_cookie('name2', 'value 2', path='/foo') cookies = [value for name, value in r.headerlist if name.title() == 'Set-Cookie'] cookies.sort() self.assertEqual(cookies[0], 'name1=value; Max-Age=5') self.assertEqual(cookies[1], 'name2="value 2"; Path=/foo') def test_set_cookie_maxage(self): import datetime r = BaseResponse() r.set_cookie('name1', 'value', max_age=5) r.set_cookie('name2', 'value', max_age=datetime.timedelta(days=1)) cookies = sorted([value for name, value in r.headerlist if name.title() == 'Set-Cookie']) self.assertEqual(cookies[0], 'name1=value; Max-Age=5') self.assertEqual(cookies[1], 'name2=value; Max-Age=86400') def test_set_cookie_expires(self): import datetime r = BaseResponse() r.set_cookie('name1', 'value', expires=42) r.set_cookie('name2', 'value', expires=datetime.datetime(1970,1,1,0,0,43)) cookies = sorted([value for name, value in r.headerlist if name.title() == 'Set-Cookie']) self.assertEqual(cookies[0], 'name1=value; expires=Thu, 01 Jan 1970 00:00:42 GMT') self.assertEqual(cookies[1], 'name2=value; expires=Thu, 01 Jan 1970 00:00:43 GMT') @unittest.skip("broken") def test_delete_cookie(self): response = BaseResponse() response.set_cookie('name', 'value') response.delete_cookie('name') cookies = [value for name, value in response.headerlist if name.title() == 'Set-Cookie'] self.assertTrue('name=;' in cookies[0]) def test_set_header(self): response = BaseResponse() response['x-test'] = 'foo' headers = [value for name, value in response.headerlist if name.title() == 'X-Test'] self.assertEqual(['foo'], headers) self.assertEqual('foo', response['x-test']) response['X-Test'] = 'bar' headers = [value for name, value in response.headerlist if name.title() == 'X-Test'] self.assertEqual(['bar'], headers) self.assertEqual('bar', response['x-test']) def test_append_header(self): response = BaseResponse() response.set_header('x-test', 'foo') headers = [value for name, value in response.headerlist if name.title() == 'X-Test'] self.assertEqual(['foo'], headers) self.assertEqual('foo', response['x-test']) response.add_header('X-Test', 'bar') headers = [value for name, value in response.headerlist if name.title() == 'X-Test'] self.assertEqual(['foo', 'bar'], headers) self.assertEqual('bar', response['x-test']) def test_delete_header(self): response = BaseResponse() response['x-test'] = 'foo' self.assertEqual('foo', response['x-test']) del response['X-tESt'] self.assertRaises(KeyError, lambda: response['x-test']) def test_non_string_header(self): response = BaseResponse() response['x-test'] = 5 self.assertEqual('5', response['x-test']) response['x-test'] = None self.assertEqual('None', response['x-test']) response['x-test'] = touni('瓶') self.assertEqual(tonat(touni('瓶')), response['x-test']) def test_prevent_control_characters_in_headers(self): masks = '{}test', 'test{}', 'te{}st' tests = '\n', '\r', '\n\r', '\0' # Test HeaderDict apis = 'append', 'replace', '__setitem__', 'setdefault' for api, mask, test in product(apis, masks, tests): hd = bottle.HeaderDict() func = getattr(hd, api) value = mask.replace("{}", test) self.assertRaises(ValueError, func, value, "test-value") self.assertRaises(ValueError, func, "test-name", value) # Test functions on BaseResponse apis = 'add_header', 'set_header', '__setitem__' for api, mask, test in product(apis, masks, tests): rs = bottle.BaseResponse() func = getattr(rs, api) value = mask.replace("{}", test) self.assertRaises(ValueError, func, value, "test-value") self.assertRaises(ValueError, func, "test-name", value) def test_expires_header(self): import datetime response = BaseResponse() now = datetime.datetime.now() response.expires = now def seconds(a, b): td = max(a,b) - min(a,b) return td.days*360*24 + td.seconds self.assertEqual(0, seconds(response.expires, now)) now2 = datetime.datetime.utcfromtimestamp( parse_date(response.headers['Expires'])) self.assertEqual(0, seconds(now, now2)) class TestRedirect(unittest.TestCase): def assertRedirect(self, target, result, query=None, status=303, **args): env = {'SERVER_PROTOCOL':'HTTP/1.1'} for key in list(args): if key.startswith('wsgi'): args[key.replace('_', '.', 1)] = args[key] del args[key] env.update(args) request.bind(env) bottle.response.bind() try: bottle.redirect(target, **(query or {})) except bottle.HTTPResponse: r = _e() self.assertEqual(status, r.status_code) self.assertTrue(r.headers) self.assertEqual(result, r.headers['Location']) def test_absolute_path(self): self.assertRedirect('/', 'http://127.0.0.1/') self.assertRedirect('/test.html', 'http://127.0.0.1/test.html') self.assertRedirect('/test.html', 'http://127.0.0.1/test.html', PATH_INFO='/some/sub/path/') self.assertRedirect('/test.html', 'http://127.0.0.1/test.html', PATH_INFO='/some/sub/file.html') self.assertRedirect('/test.html', 'http://127.0.0.1/test.html', SCRIPT_NAME='/some/sub/path/') self.assertRedirect('/foo/test.html', 'http://127.0.0.1/foo/test.html') self.assertRedirect('/foo/test.html', 'http://127.0.0.1/foo/test.html', PATH_INFO='/some/sub/file.html') def test_relative_path(self): self.assertRedirect('./', 'http://127.0.0.1/') self.assertRedirect('./test.html', 'http://127.0.0.1/test.html') self.assertRedirect('./test.html', 'http://127.0.0.1/foo/test.html', PATH_INFO='/foo/') self.assertRedirect('./test.html', 'http://127.0.0.1/foo/test.html', PATH_INFO='/foo/bar.html') self.assertRedirect('./test.html', 'http://127.0.0.1/foo/test.html', SCRIPT_NAME='/foo/') self.assertRedirect('./test.html', 'http://127.0.0.1/foo/bar/test.html', SCRIPT_NAME='/foo/', PATH_INFO='/bar/baz.html') self.assertRedirect('./foo/test.html', 'http://127.0.0.1/foo/test.html') self.assertRedirect('./foo/test.html', 'http://127.0.0.1/bar/foo/test.html', PATH_INFO='/bar/file.html') self.assertRedirect('../test.html', 'http://127.0.0.1/test.html', PATH_INFO='/foo/') self.assertRedirect('../test.html', 'http://127.0.0.1/foo/test.html', PATH_INFO='/foo/bar/') self.assertRedirect('../test.html', 'http://127.0.0.1/test.html', PATH_INFO='/foo/bar.html') self.assertRedirect('../test.html', 'http://127.0.0.1/test.html', SCRIPT_NAME='/foo/') self.assertRedirect('../test.html', 'http://127.0.0.1/foo/test.html', SCRIPT_NAME='/foo/', PATH_INFO='/bar/baz.html') self.assertRedirect('../baz/../test.html', 'http://127.0.0.1/foo/test.html', PATH_INFO='/foo/bar/') def test_sheme(self): self.assertRedirect('./test.html', 'https://127.0.0.1/test.html', wsgi_url_scheme='https') self.assertRedirect('./test.html', 'https://127.0.0.1:80/test.html', wsgi_url_scheme='https', SERVER_PORT='80') def test_host_http_1_0(self): # No HTTP_HOST, just SERVER_NAME and SERVER_PORT. self.assertRedirect('./test.html', 'http://example.com/test.html', SERVER_NAME='example.com', SERVER_PROTOCOL='HTTP/1.0', status=302) self.assertRedirect('./test.html', 'http://127.0.0.1:81/test.html', SERVER_PORT='81', SERVER_PROTOCOL='HTTP/1.0', status=302) def test_host_http_1_1(self): self.assertRedirect('./test.html', 'http://example.com/test.html', HTTP_HOST='example.com') self.assertRedirect('./test.html', 'http://example.com:81/test.html', HTTP_HOST='example.com:81') # Trust HTTP_HOST over SERVER_NAME and PORT. self.assertRedirect('./test.html', 'http://example.com:81/test.html', HTTP_HOST='example.com:81', SERVER_NAME='foobar') self.assertRedirect('./test.html', 'http://example.com:81/test.html', HTTP_HOST='example.com:81', SERVER_PORT='80') def test_host_http_proxy(self): # Trust proxy headers over original header. self.assertRedirect('./test.html', 'http://example.com/test.html', HTTP_X_FORWARDED_HOST='example.com', HTTP_HOST='127.0.0.1') def test_specialchars(self): ''' The target URL is not quoted automatically. ''' self.assertRedirect('./te st.html', 'http://example.com/a%20a/b%20b/te st.html', HTTP_HOST='example.com', SCRIPT_NAME='/a a/', PATH_INFO='/b b/') def test_redirect_preserve_cookies(self): env = {'SERVER_PROTOCOL':'HTTP/1.1'} request.bind(env) bottle.response.bind() try: bottle.response.set_cookie('xxx', 'yyy') bottle.redirect('...') except bottle.HTTPResponse: h = [v for (k, v) in _e().headerlist if k == 'Set-Cookie'] self.assertEqual(h, ['xxx=yyy']) class TestWSGIHeaderDict(unittest.TestCase): def setUp(self): self.env = {} self.headers = bottle.WSGIHeaderDict(self.env) def test_empty(self): self.assertEqual(0, len(bottle.WSGIHeaderDict({}))) def test_native(self): self.env['HTTP_TEST_HEADER'] = 'foobar' self.assertEqual(self.headers['Test-header'], 'foobar') def test_bytes(self): self.env['HTTP_TEST_HEADER'] = tob('foobar') self.assertEqual(self.headers['Test-Header'], 'foobar') def test_unicode(self): self.env['HTTP_TEST_HEADER'] = touni('foobar') self.assertEqual(self.headers['Test-Header'], 'foobar') def test_dict(self): for key in 'foo-bar Foo-Bar foo-Bar FOO-BAR'.split(): self.assertTrue(key not in self.headers) self.assertEqual(self.headers.get(key), None) self.assertEqual(self.headers.get(key, 5), 5) self.assertRaises(KeyError, lambda x: self.headers[x], key) self.env['HTTP_FOO_BAR'] = 'test' for key in 'foo-bar Foo-Bar foo-Bar FOO-BAR'.split(): self.assertTrue(key in self.headers) self.assertEqual(self.headers.get(key), 'test') self.assertEqual(self.headers.get(key, 5), 'test') if __name__ == '__main__': #pragma: no cover unittest.main()