blob: 8e9b3ae4764768b8d2b115599d8cb2c8765f7072 [file] [edit]
import zlib
import io
import sys
import pytest
from webob.request import BaseRequest
from webob.request import Request
from webob.response import Response
from webob.compat import text_
from webob.compat import bytes_
from webob import cookies
def setup_module(module):
cookies._should_raise = True
def teardown_module(module):
cookies._should_raise = False
def simple_app(environ, start_response):
start_response('200 OK', [
('Content-Type', 'text/html; charset=UTF-8'),
])
return ['OK']
def test_response():
req = BaseRequest.blank('/')
res = req.get_response(simple_app)
assert res.status == '200 OK'
assert res.status_code == 200
assert res.body == "OK"
assert res.charset == "UTF-8"
assert res.content_type == 'text/html'
res.status = 404
assert res.status == '404 Not Found'
assert res.status_code == 404
res.body = b'Not OK'
assert b''.join(res.app_iter) == b'Not OK'
res.charset = 'iso8859-1'
assert 'text/html; charset=iso8859-1' == res.headers['content-type']
res.content_type = 'text/xml'
assert 'text/xml; charset=UTF-8' == res.headers['content-type']
res.content_type = 'text/xml; charset=UTF-8'
assert 'text/xml; charset=UTF-8' == res.headers['content-type']
res.headers = {'content-type': 'text/html'}
assert res.headers['content-type'] == 'text/html'
assert res.headerlist == [('content-type', 'text/html')]
res.set_cookie('x', 'y')
assert res.headers['set-cookie'].strip(';') == 'x=y; Path=/'
res.set_cookie(text_('x'), text_('y'))
assert res.headers['set-cookie'].strip(';') == 'x=y; Path=/'
res = Response('a body', '200 OK', content_type='text/html')
res.encode_content()
assert res.content_encoding == 'gzip'
assert res.body == b'\x1f\x8b\x08\x00\x00\x00\x00\x00\x02\xffKTH\xcaO\xa9\x04\x00\xf6\x86GI\x06\x00\x00\x00'
res.decode_content()
assert res.content_encoding is None
assert res.body == b'a body'
res.set_cookie('x', text_(b'foo')) # test unicode value
with pytest.raises(TypeError):
Response(app_iter=iter(['a']),
body="somebody")
del req.environ
with pytest.raises(TypeError):
Response(charset=None,
content_type='image/jpeg',
body=text_(b"unicode body"))
with pytest.raises(TypeError):
Response(wrong_key='dummy')
with pytest.raises(TypeError):
resp = Response()
resp.body = text_(b"unicode body")
def test_set_response_status_binary():
req = BaseRequest.blank('/')
res = req.get_response(simple_app)
res.status == b'200 OK'
assert res.status_code == 200
assert res.status == '200 OK'
def test_set_response_status_str_no_reason():
req = BaseRequest.blank('/')
res = req.get_response(simple_app)
res.status = '200'
assert res.status_code == 200
assert res.status == '200 OK'
def test_set_response_status_str_generic_reason():
req = BaseRequest.blank('/')
res = req.get_response(simple_app)
res.status = '299'
assert res.status_code == 299
assert res.status == '299 Success'
def test_set_response_status_code():
req = BaseRequest.blank('/')
res = req.get_response(simple_app)
res.status_code = 200
assert res.status_code == 200
assert res.status == '200 OK'
def test_set_response_status_bad():
req = BaseRequest.blank('/')
res = req.get_response(simple_app)
def status_test():
res.status = 'ThisShouldFail'
with pytest.raises(ValueError):
status_test()
def test_set_response_status_code_generic_reason():
req = BaseRequest.blank('/')
res = req.get_response(simple_app)
res.status_code = 299
assert res.status_code == 299
assert res.status == '299 Success'
def test_content_type():
r = Response()
# default ctype and charset
assert r.content_type == 'text/html'
assert r.charset == 'UTF-8'
# setting to none, removes the header
r.content_type = None
assert r.content_type is None
assert r.charset is None
# can set missing ctype
r.content_type = None
assert r.content_type is None
def test_init_content_type_w_charset():
v = 'text/plain;charset=ISO-8859-1'
assert Response(content_type=v).headers['content-type'] == v
def test_init_adds_default_charset_when_not_json():
content_type = 'text/plain'
expected = 'text/plain; charset=UTF-8'
assert Response(content_type=content_type).headers['content-type'] == expected
def test_init_no_charset_when_json():
content_type = 'application/json'
expected = content_type
assert Response(content_type=content_type).headers['content-type'] == expected
def test_init_keeps_specified_charset_when_json():
content_type = 'application/json; charset=ISO-8859-1'
expected = content_type
assert Response(content_type=content_type).headers['content-type'] == expected
def test_init_doesnt_add_default_content_type_with_bodyless_status():
assert Response(status='204 No Content').content_type is None
def test_cookies():
res = Response()
# test unicode value
res.set_cookie('x', "test")
# utf8 encoded
assert res.headers.getall('set-cookie') == ['x=test; Path=/']
r2 = res.merge_cookies(simple_app)
r2 = BaseRequest.blank('/').get_response(r2)
assert r2.headerlist == [
('Content-Type', 'text/html; charset=UTF-8'),
('Set-Cookie', 'x=test; Path=/'),
]
def test_unicode_cookies_error_raised():
res = Response()
with pytest.raises(ValueError):
Response.set_cookie(
res,
'x',
text_(b'\N{BLACK SQUARE}', 'unicode_escape'))
def test_unicode_cookies_warning_issued():
import warnings
cookies._should_raise = False
with warnings.catch_warnings(record=True) as w:
# Cause all warnings to always be triggered.
warnings.simplefilter("always")
# Trigger a warning.
res = Response()
res.set_cookie('x', text_(b'\N{BLACK SQUARE}', 'unicode_escape'))
assert len(w) == 1
assert issubclass(w[-1].category, RuntimeWarning) is True
assert "ValueError" in str(w[-1].message)
cookies._should_raise = True
def test_cookies_raises_typeerror():
res = Response()
with pytest.raises(TypeError):
res.set_cookie()
def test_http_only_cookie():
req = Request.blank('/')
res = req.get_response(Response('blah'))
res.set_cookie("foo", "foo", httponly=True)
assert res.headers['set-cookie'] == 'foo=foo; Path=/; HttpOnly'
def test_headers():
r = Response()
tval = 'application/x-test'
r.headers.update({'content-type': tval})
assert r.headers.getall('content-type') == [tval]
r.headers.clear()
assert not r.headerlist
def test_response_copy():
r = Response(app_iter=iter(['a']))
r2 = r.copy()
assert r.body == 'a'
assert r2.body == 'a'
def test_response_copy_content_md5():
res = Response()
res.md5_etag(set_content_md5=True)
assert res.content_md5
res2 = res.copy()
assert res.content_md5
assert res2.content_md5
assert res.content_md5 == res2.content_md5
def test_HEAD_closes():
req = Request.blank('/')
req.method = 'HEAD'
app_iter = io.BytesIO(b'foo')
res = req.get_response(Response(app_iter=app_iter))
assert res.status_code == 200
assert res.body == b''
assert app_iter.closed
def test_HEAD_conditional_response_returns_empty_response():
req = Request.blank('/', method='HEAD', if_none_match='none')
res = Response(conditional_response=True)
def start_response(status, headerlist):
pass
result = res(req.environ, start_response)
assert not list(result)
def test_HEAD_conditional_response_range_empty_response():
req = Request.blank('/', method='HEAD', range=(4, 5))
res = Response('Are we not men?', conditional_response=True)
assert req.get_response(res).body == b''
def test_conditional_response_if_none_match_false():
req = Request.blank('/', if_none_match='foo')
resp = Response(app_iter=['foo\n'], conditional_response=True, etag='bar')
resp = req.get_response(resp)
assert resp.status_code == 200
def test_conditional_response_if_none_match_true():
req = Request.blank('/', if_none_match='foo')
resp = Response(app_iter=['foo\n'], conditional_response=True, etag='foo')
resp = req.get_response(resp)
assert resp.status_code == 304
def test_conditional_response_if_none_match_weak():
req = Request.blank('/', headers={'if-none-match': '"bar"'})
req_weak = Request.blank('/', headers={'if-none-match': 'W/"bar"'})
resp = Response(app_iter=['foo\n'], conditional_response=True, etag='bar')
resp_weak = Response(app_iter=['foo\n'],
conditional_response=True, headers={'etag': 'W/"bar"'})
for rq in [req, req_weak]:
for rp in [resp, resp_weak]:
rq.get_response(rp).status_code == 304
r2 = Response(app_iter=['foo\n'],
conditional_response=True,
headers={'etag': '"foo"'})
r2_weak = Response(app_iter=['foo\n'],
conditional_response=True,
headers={'etag': 'W/"foo"'})
req_weak.get_response(r2).status_code == 200
req.get_response(r2_weak) == 200
def test_conditional_response_if_modified_since_false():
from datetime import datetime, timedelta
req = Request.blank('/', if_modified_since=datetime(2011, 3, 17, 13, 0, 0))
resp = Response(app_iter=['foo\n'], conditional_response=True,
last_modified=req.if_modified_since - timedelta(seconds=1))
resp = req.get_response(resp)
assert resp.status_code == 304
def test_conditional_response_if_modified_since_true():
from datetime import datetime, timedelta
req = Request.blank('/', if_modified_since=datetime(2011, 3, 17, 13, 0, 0))
resp = Response(app_iter=['foo\n'], conditional_response=True,
last_modified=req.if_modified_since + timedelta(seconds=1))
resp = req.get_response(resp)
assert resp.status_code == 200
def test_conditional_response_range_not_satisfiable_response():
req = Request.blank('/', range='bytes=100-200')
resp = Response(app_iter=['foo\n'], content_length=4, conditional_response=True)
resp = req.get_response(resp)
assert resp.status_code == 416
assert resp.content_range.start is None
assert resp.content_range.stop is None
assert resp.content_range.length == 4
assert resp.body == b'Requested range not satisfiable: bytes=100-200'
def test_HEAD_conditional_response_range_not_satisfiable_response():
req = Request.blank('/', method='HEAD', range='bytes=100-200')
resp = Response(app_iter=['foo\n'], content_length=4, conditional_response=True)
resp = req.get_response(resp)
assert resp.status_code == 416
assert resp.content_range.start is None
assert resp.content_range.stop is None
assert resp.content_range.length == 4
assert resp.body == b''
def test_md5_etag():
res = Response()
res.body = b"""\
In A.D. 2101
War was beginning.
Captain: What happen ?
Mechanic: Somebody set up us the bomb.
Operator: We get signal.
Captain: What !
Operator: Main screen turn on.
Captain: It's You !!
Cats: How are you gentlemen !!
Cats: All your base are belong to us.
Cats: You are on the way to destruction.
Captain: What you say !!
Cats: You have no chance to survive make your time.
Cats: HA HA HA HA ....
Captain: Take off every 'zig' !!
Captain: You know what you doing.
Captain: Move 'zig'.
Captain: For great justice."""
res.md5_etag()
assert res.etag
assert '\n' not in res.etag
assert res.etag == 'pN8sSTUrEaPRzmurGptqmw'
assert res.content_md5 is None
def test_md5_etag_set_content_md5():
res = Response()
body = b'The quick brown fox jumps over the lazy dog'
res.md5_etag(body, set_content_md5=True)
assert res.content_md5 == 'nhB9nTcrtoJr2B01QqQZ1g=='
def test_decode_content_defaults_to_identity():
res = Response()
res.body = b'There be dragons'
res.decode_content()
assert res.body == b'There be dragons'
def test_decode_content_with_deflate():
res = Response()
body = b'Hey Hey Hey'
# Simulate inflate by chopping the headers off
# the gzip encoded data
res.body = zlib.compress(body)[2:-4]
res.content_encoding = 'deflate'
res.decode_content()
assert res.body == body
assert res.content_encoding is None
def test_content_length():
r0 = Response('x' * 10, content_length=10)
req_head = Request.blank('/', method='HEAD')
r1 = req_head.get_response(r0)
assert r1.status_code == 200
assert r1.body == b''
assert r1.content_length == 10
req_get = Request.blank('/')
r2 = req_get.get_response(r0)
assert r2.status_code == 200
assert r2.body == b'x' * 10
assert r2.content_length == 10
r3 = Response(app_iter=[b'x'] * 10)
assert r3.content_length is None
assert r3.body == b'x' * 10
assert r3.content_length == 10
r4 = Response(app_iter=[b'x'] * 10,
content_length=20) # wrong content_length
assert r4.content_length == 20
with pytest.raises(AssertionError):
r4.body
req_range = Request.blank('/', range=(0, 5))
r0.conditional_response = True
r5 = req_range.get_response(r0)
assert r5.status_code == 206
assert r5.body == b'xxxxx'
assert r5.content_length == 5
def test_app_iter_range():
req = Request.blank('/', range=(2, 5))
for app_iter in [
[b'012345'],
[b'0', b'12345'],
[b'0', b'1234', b'5'],
[b'01', b'2345'],
[b'01', b'234', b'5'],
[b'012', b'34', b'5'],
[b'012', b'3', b'4', b'5'],
[b'012', b'3', b'45'],
[b'0', b'12', b'34', b'5'],
[b'0', b'12', b'345'],
]:
r = Response(
app_iter=app_iter,
content_length=6,
conditional_response=True,
)
res = req.get_response(r)
assert list(res.content_range) == [2, 5, 6]
assert res.body, b'234'
def test_app_iter_range_inner_method():
class FakeAppIter:
def app_iter_range(self, start, stop):
return 'you win', start, stop
res = Response(app_iter=FakeAppIter())
assert res.app_iter_range(30, 40), ('you win', 30 == 40)
def test_has_body():
empty = Response()
assert not empty.has_body
with_list = Response(app_iter=['1'])
assert with_list.has_body
with_empty_list = Response(app_iter=[b''])
assert not with_empty_list.has_body
with_body = Response(body='Seomthing')
assert with_body.has_body
with_none_app_iter = Response(app_iter=None)
assert not with_none_app_iter.has_body
with_none_body = Response(body=None)
assert not with_none_body.has_body
# key feature: has_body should not read app_iter
app_iter = iter(['1', '2'])
not_iterating = Response(app_iter=app_iter)
assert not_iterating.has_body
assert next(app_iter) == '1'
# messed with private attribute but method should nonetheless not
# return True
messing_with_privates = Response()
messing_with_privates._app_iter = None
assert not messing_with_privates.has_body
def test_str_crlf():
res = Response('test')
assert '\r\n' in str(res)
def test_from_file():
res = Response('test')
inp = io.BytesIO(bytes_(str(res)))
equal_resp(res, inp)
def test_from_file2():
res = Response(app_iter=iter([b'test ', b'body']),
content_type='text/plain')
inp = io.BytesIO(bytes_(str(res)))
equal_resp(res, inp)
def test_from_text_file():
res = Response('test')
inp = io.StringIO(text_(str(res), 'utf-8'))
equal_resp(res, inp)
res = Response(app_iter=iter([b'test ', b'body']),
content_type='text/plain')
inp = io.StringIO(text_(str(res), 'utf-8'))
equal_resp(res, inp)
def equal_resp(res, inp):
res2 = Response.from_file(inp)
assert res.body == res2.body
assert res.headers == res2.headers
def test_from_file_w_leading_space_in_header():
# Make sure the removal of code dealing with leading spaces is safe
res1 = Response()
file_w_space = io.BytesIO(
b'200 OK\n\tContent-Type: text/html; charset=UTF-8')
res2 = Response.from_file(file_w_space)
assert res1.headers == res2.headers
def test_file_bad_header():
file_w_bh = io.BytesIO(b'200 OK\nBad Header')
with pytest.raises(ValueError):
Response.from_file(file_w_bh)
def test_from_file_not_unicode_headers():
inp = io.BytesIO(
b'200 OK\n\tContent-Type: text/html; charset=UTF-8')
res = Response.from_file(inp)
assert res.headerlist[0][0].__class__ == str
def test_file_with_http_version():
inp = io.BytesIO(b'HTTP/1.1 200 OK\r\n\r\nSome data...')
res = Response.from_file(inp)
assert res.status_code == 200
assert res.status == '200 OK'
def test_file_with_http_version_more_status():
inp = io.BytesIO(b'HTTP/1.1 404 Not Found\r\n\r\nSome data...')
res = Response.from_file(inp)
assert res.status_code == 404
assert res.status == '404 Not Found'
def test_set_status():
res = Response()
res.status = "200"
assert res.status == "200 OK"
with pytest.raises(TypeError):
setattr(res, 'status', (200,))
def test_set_headerlist():
res = Response()
# looks like a list
res.headerlist = (('Content-Type', 'text/html; charset=UTF-8'),)
assert res.headerlist == [('Content-Type', 'text/html; charset=UTF-8')]
# has items
res.headerlist = {'Content-Type': 'text/html; charset=UTF-8'}
assert res.headerlist == [('Content-Type', 'text/html; charset=UTF-8')]
del res.headerlist
assert res.headerlist == []
def test_request_uri_no_script_name():
from webob.response import _request_uri
environ = {
'wsgi.url_scheme': 'http',
'HTTP_HOST': 'test.com',
}
assert _request_uri(environ) == 'http://test.com/'
def test_request_uri_https():
from webob.response import _request_uri
environ = {
'wsgi.url_scheme': 'https',
'SERVER_NAME': 'test.com',
'SERVER_PORT': '443',
'SCRIPT_NAME': '/foobar',
}
assert _request_uri(environ) == 'https://test.com/foobar'
def test_app_iter_range_starts_after_iter_end():
from webob.response import AppIterRange
range = AppIterRange(iter([]), start=1, stop=1)
assert list(range) == []
def test_resp_write_app_iter_non_list():
res = Response(app_iter=(b'a', b'b'))
assert res.content_length is None
res.write(b'c')
assert res.body == b'abc'
assert res.content_length == 3
def test_response_file_body_writelines():
from webob.response import ResponseBodyFile
res = Response(app_iter=[b'foo'])
rbo = ResponseBodyFile(res)
rbo.writelines(['bar', 'baz'])
assert res.app_iter == [b'foo', b'bar', b'baz']
rbo.flush() # noop
assert res.app_iter, [b'foo', b'bar', b'baz']
@pytest.mark.xfail(sys.version_info >= (3,6),
reason="Python 3.6 and up requires that rbo is seekable.")
def test_response_file_body_tell():
import zipfile
from webob.response import ResponseBodyFile
rbo = ResponseBodyFile(Response())
assert rbo.tell() == 0
writer = zipfile.ZipFile(rbo, 'w')
writer.writestr('zinfo_or_arcname', b'foo')
writer.close()
assert rbo.tell() == 133
def test_response_file_body_tell_text():
from webob.response import ResponseBodyFile
rbo = ResponseBodyFile(Response())
assert rbo.tell() == 0
rbo.write('123456789')
assert rbo.tell() == 9
def test_response_write_non_str():
res = Response()
with pytest.raises(TypeError):
res.write(object())
def test_response_file_body_write_empty_app_iter():
res = Response('foo')
res.write('baz')
assert res.app_iter == [b'foo', b'baz']
def test_response_file_body_write_empty_body():
res = Response('')
res.write('baz')
assert res.app_iter == [b'', b'baz']
def test_response_file_body_close_not_implemented():
rbo = Response().body_file
with pytest.raises(NotImplementedError):
rbo.close()
def test_response_file_body_repr():
rbo = Response().body_file
rbo.response = 'yo'
assert repr(rbo) == "<body_file for 'yo'>"
def test_body_get_is_none():
res = Response()
res._app_iter = None
with pytest.raises(TypeError):
Response(app_iter=iter(['a']), body="somebody")
with pytest.raises(AttributeError):
res.__getattribute__('body')
def test_body_get_is_unicode_notverylong():
res = Response(app_iter=(text_(b'foo'),))
with pytest.raises(TypeError):
res.__getattribute__('body')
def test_body_get_is_unicode():
res = Response(app_iter=(['x'] * 51 + [text_(b'x')]))
with pytest.raises(TypeError):
res.__getattribute__('body')
def test_body_set_not_unicode_or_str():
res = Response()
with pytest.raises(TypeError):
res.__setattr__('body', object())
def test_body_set_unicode():
res = Response()
with pytest.raises(TypeError):
res.__setattr__('body', text_(b'abc'))
def test_body_set_under_body_doesnt_exist():
res = Response('abc')
assert res.body == b'abc'
assert res.content_length == 3
def test_body_del():
res = Response('123')
del res.body
assert res.body == b''
assert res.content_length == 0
def test_text_get_no_charset():
res = Response(charset=None)
assert '' == res.text
def test_text_get_no_default_body_encoding():
res = Response(charset=None)
res.default_body_encoding = None
with pytest.raises(AttributeError):
assert '' == res.text
def test_unicode_body():
res = Response()
res.charset = 'utf-8'
bbody = b'La Pe\xc3\xb1a' # binary string
ubody = text_(bbody, 'utf-8') # unicode string
res.body = bbody
assert res.unicode_body == ubody
res.ubody = ubody
assert res.body == bbody
del res.ubody
assert res.body == b''
def test_text_get_decode():
res = Response()
res.charset = 'utf-8'
res.body = b'La Pe\xc3\xb1a'
assert res.text, text_(b'La Pe\xc3\xb1a')
def test_text_set_no_charset():
res = Response()
res.charset = None
res.text = text_('abc')
assert res.text == 'abc'
def test_text_set_no_default_body_encoding():
res = Response()
res.charset = None
res.default_body_encoding = None
with pytest.raises(AttributeError):
res.text = text_('abc')
def test_text_set_not_unicode():
res = Response()
res.charset = 'utf-8'
with pytest.raises(TypeError):
res.__setattr__('text',
b'La Pe\xc3\xb1a')
def test_text_del():
res = Response('123')
del res.text
assert res.body == b''
assert res.content_length == 0
def test_body_file_del():
res = Response()
res.body = b'123'
assert res.content_length == 3
assert res.app_iter == [b'123']
del res.body_file
assert res.body == b''
assert res.content_length == 0
def test_write_unicode():
res = Response()
res.text = text_(b'La Pe\xc3\xb1a', 'utf-8')
res.write(text_(b'a'))
assert res.text, text_(b'La Pe\xc3\xb1aa' == 'utf-8')
def test_write_unicode_no_charset():
res = Response(charset=None)
with pytest.raises(TypeError):
res.write(text_(b'a'))
def test_write_text():
res = Response()
res.body = b'abc'
res.write(text_(b'a'))
assert res.text == 'abca'
def test_app_iter_del():
res = Response(
content_length=3,
app_iter=['123'],
)
del res.app_iter
assert res.body == b''
assert res.content_length is None
def test_charset_set_no_content_type_header():
res = Response()
res.headers.pop('Content-Type', None)
with pytest.raises(AttributeError):
res.charset = 'utf-8'
def test_charset_del_no_content_type_header():
res = Response()
res.headers.pop('Content-Type', None)
assert res._charset__del() is None
def test_content_type_params_get_no_semicolon_in_content_type_header():
res = Response()
res.headers['Content-Type'] = 'foo'
assert res.content_type_params == {}
def test_content_type_params_get_semicolon_in_content_type_header():
res = Response()
res.headers['Content-Type'] = 'foo;encoding=utf-8'
assert res.content_type_params == {'encoding': 'utf-8'}
def test_content_type_params_set_value_dict_empty():
res = Response()
res.headers['Content-Type'] = 'foo;bar'
res.content_type_params = None
assert res.headers['Content-Type'] == 'foo'
def test_content_type_params_set_ok_param_quoting():
res = Response()
res.content_type_params = {'a': ''}
assert res.headers['Content-Type'] == 'text/html; a=""'
def test_charset_delete():
res = Response()
del res.charset
assert res.charset is None
def test_set_cookie_overwrite():
res = Response()
res.set_cookie('a', '1')
res.set_cookie('a', '2', overwrite=True)
assert res.headerlist[-1] == ('Set-Cookie', 'a=2; Path=/')
def test_set_cookie_value_is_None():
res = Response()
res.set_cookie('a', None)
assert res.headerlist[-1][0] == 'Set-Cookie'
val = [x.strip() for x in res.headerlist[-1][1].split(';')]
assert len(val) == 4
val.sort()
assert val[0] == 'Max-Age=0'
assert val[1] == 'Path=/'
assert val[2] == 'a='
assert val[3].startswith('expires')
def test_set_cookie_expires_is_None_and_max_age_is_int():
res = Response()
res.set_cookie('a', '1', max_age=100)
assert res.headerlist[-1][0] == 'Set-Cookie'
val = [x.strip() for x in res.headerlist[-1][1].split(';')]
assert len(val) == 4
val.sort()
assert val[0] == 'Max-Age=100'
assert val[1] == 'Path=/'
assert val[2] == 'a=1'
assert val[3].startswith('expires')
def test_set_cookie_expires_is_None_and_max_age_is_timedelta():
from datetime import timedelta
res = Response()
res.set_cookie('a', '1', max_age=timedelta(seconds=100))
assert res.headerlist[-1][0] == 'Set-Cookie'
val = [x.strip() for x in res.headerlist[-1][1].split(';')]
assert len(val) == 4
val.sort()
assert val[0] == 'Max-Age=100'
assert val[1] == 'Path=/'
assert val[2] == 'a=1'
assert val[3].startswith('expires')
def test_set_cookie_expires_is_datetime_and_max_age_is_None():
import datetime
res = Response()
then = datetime.datetime.utcnow() + datetime.timedelta(days=1)
res.set_cookie('a', '1', expires=then)
assert res.headerlist[-1][0] == 'Set-Cookie'
val = [x.strip() for x in res.headerlist[-1][1].split(';')]
assert len(val) == 4
val.sort()
assert val[0] in ('Max-Age=86399', 'Max-Age=86400')
assert val[1] == 'Path=/'
assert val[2] == 'a=1'
assert val[3].startswith('expires')
def test_set_cookie_expires_is_timedelta_and_max_age_is_None():
import datetime
res = Response()
then = datetime.timedelta(days=1)
res.set_cookie('a', '1', expires=then)
assert res.headerlist[-1][0] == 'Set-Cookie'
val = [x.strip() for x in res.headerlist[-1][1].split(';')]
assert len(val) == 4
val.sort()
assert val[0] in ('Max-Age=86399', 'Max-Age=86400')
assert val[1] == 'Path=/'
assert val[2] == 'a=1'
assert val[3].startswith('expires')
def test_set_cookie_expires_is_datetime_tz_and_max_age_is_None():
import datetime
res = Response()
class FixedOffset(datetime.tzinfo):
def __init__(self, offset, name):
self.__offset = datetime.timedelta(minutes=offset)
self.__name = name
def utcoffset(self, dt):
return self.__offset
def tzname(self, dt):
return self.__name
def dst(self, dt):
return datetime.timedelta(0)
then = datetime.datetime.now(FixedOffset(60, 'UTC+1')) + datetime.timedelta(days=1)
res.set_cookie('a', '1', expires=then)
assert res.headerlist[-1][0] == 'Set-Cookie'
val = [x.strip() for x in res.headerlist[-1][1].split(';')]
assert len(val) == 4
val.sort()
assert val[0] in ('Max-Age=86399', 'Max-Age=86400')
assert val[1] == 'Path=/'
assert val[2] == 'a=1'
assert val[3].startswith('expires')
def test_delete_cookie():
res = Response()
res.headers['Set-Cookie'] = 'a=2; Path=/'
res.delete_cookie('a')
assert res.headerlist[-1][0] == 'Set-Cookie'
val = [x.strip() for x in res.headerlist[-1][1].split(';')]
assert len(val) == 4
val.sort()
assert val[0] == 'Max-Age=0'
assert val[1] == 'Path=/'
assert val[2] == 'a='
assert val[3].startswith('expires')
def test_delete_cookie_with_path():
res = Response()
res.headers['Set-Cookie'] = 'a=2; Path=/'
res.delete_cookie('a', path='/abc')
assert res.headerlist[-1][0] == 'Set-Cookie'
val = [x.strip() for x in res.headerlist[-1][1].split(';')]
assert len(val) == 4
val.sort()
assert val[0] == 'Max-Age=0'
assert val[1] == 'Path=/abc'
assert val[2] == 'a='
assert val[3].startswith('expires')
def test_delete_cookie_with_domain():
res = Response()
res.headers['Set-Cookie'] = 'a=2; Path=/'
res.delete_cookie('a', path='/abc', domain='example.com')
assert res.headerlist[-1][0] == 'Set-Cookie'
val = [x.strip() for x in res.headerlist[-1][1].split(';')]
assert len(val) == 5
val.sort()
assert val[0] == 'Domain=example.com'
assert val[1] == 'Max-Age=0'
assert val[2] == 'Path=/abc'
assert val[3] == 'a='
assert val[4].startswith('expires')
def test_unset_cookie_not_existing_and_not_strict():
res = Response()
res.unset_cookie('a', strict=False) # no exception
def test_unset_cookie_not_existing_and_strict():
res = Response()
with pytest.raises(KeyError):
res.unset_cookie('a')
def test_unset_cookie_key_in_cookies():
res = Response()
res.headers.add('Set-Cookie', 'a=2; Path=/')
res.headers.add('Set-Cookie', 'b=3; Path=/')
res.unset_cookie('a')
assert res.headers.getall('Set-Cookie') == ['b=3; Path=/']
res.unset_cookie(text_('b'))
assert res.headers.getall('Set-Cookie') == []
def test_merge_cookies_no_set_cookie():
res = Response()
result = res.merge_cookies('abc')
assert result == 'abc'
def test_merge_cookies_resp_is_Response():
inner_res = Response()
res = Response()
res.set_cookie('a', '1')
result = res.merge_cookies(inner_res)
assert result.headers.getall('Set-Cookie') == ['a=1; Path=/']
def test_merge_cookies_resp_is_wsgi_callable():
L = []
def dummy_wsgi_callable(environ, start_response):
L.append((environ, start_response))
return 'abc'
res = Response()
res.set_cookie('a', '1')
wsgiapp = res.merge_cookies(dummy_wsgi_callable)
environ = {}
def dummy_start_response(status, headers, exc_info=None):
assert headers, [('Set-Cookie' == 'a=1; Path=/')]
result = wsgiapp(environ, dummy_start_response)
assert result == 'abc'
assert len(L) == 1
L[0][1]('200 OK', []) # invoke dummy_start_response assertion
def test_body_get_body_is_None_len_app_iter_is_zero():
res = Response()
res._app_iter = io.BytesIO()
res._body = None
result = res.body
assert result == b''
def test_cache_control_get():
res = Response()
assert repr(res.cache_control) == "<CacheControl ''>"
assert res.cache_control.max_age is None
def test_location():
res = Response()
res.status = '301'
res.location = '/test.html'
assert res.location == '/test.html'
req = Request.blank('/')
assert req.get_response(res).location == 'http://localhost/test.html'
res.location = '/test2.html'
assert req.get_response(res).location == 'http://localhost/test2.html'
@pytest.mark.xfail(sys.version_info < (3,0),
reason="Python 2.x unicode != str, WSGI requires str. Test "
"added due to https://github.com/Pylons/webob/issues/247. "
"PEP3333 requires environ variables are str, Django messes "
"with the environ and changes it from str to unicode.")
def test_location_unicode():
environ = {
'REQUEST_METHOD': 'GET',
'wsgi.url_scheme': 'http',
'HTTP_HOST': u'test.com',
}
res = Response()
res.status = '301'
res.location = '/test.html'
def start_response(status, headerlist):
for (header, val) in headerlist:
if header.lower() == 'location':
assert val == 'http://test.com/test.html'
assert isinstance(val, str)
res(environ, start_response)
def test_request_uri_http():
# covers webob/response.py:1152
from webob.response import _request_uri
environ = {
'wsgi.url_scheme': 'http',
'SERVER_NAME': 'test.com',
'SERVER_PORT': '80',
'SCRIPT_NAME': '/foobar',
}
assert _request_uri(environ) == 'http://test.com/foobar'
def test_request_uri_no_script_name2():
# covers webob/response.py:1160
# There is a test_request_uri_no_script_name in test_response.py, but it
# sets SCRIPT_NAME.
from webob.response import _request_uri
environ = {
'wsgi.url_scheme': 'http',
'HTTP_HOST': 'test.com',
'PATH_INFO': '/foobar',
}
assert _request_uri(environ) == 'http://test.com/foobar'
def test_cache_control_object_max_age_ten():
res = Response()
res.cache_control.max_age = 10
assert repr(res.cache_control) == "<CacheControl 'max-age=10'>"
assert res.headers['cache-control'] == 'max-age=10'
def test_cache_control_set_object_error():
res = Response()
with pytest.raises(AttributeError):
setattr(res.cache_control, 'max_stale', 10)
def test_cache_expires_set():
res = Response()
res.cache_expires = True
assert repr(res.cache_control) == "<CacheControl 'max-age=0, must-revalidate, no-cache, no-store'>"
def test_status_code_set():
res = Response()
res.status_code = 400
assert res._status == '400 Bad Request'
res.status_int = 404
assert res._status == '404 Not Found'
def test_cache_control_set_dict():
res = Response()
res.cache_control = {'a': 'b'}
assert repr(res.cache_control) == "<CacheControl 'a=b'>"
def test_cache_control_set_None():
res = Response()
res.cache_control = None
assert repr(res.cache_control) == "<CacheControl ''>"
def test_cache_control_set_unicode():
res = Response()
res.cache_control = text_(b'abc')
assert repr(res.cache_control) == "<CacheControl 'abc'>"
def test_cache_control_set_control_obj_is_not_None():
class DummyCacheControl(object):
def __init__(self):
self.header_value = 1
self.properties = {'bleh': 1}
res = Response()
res._cache_control_obj = DummyCacheControl()
res.cache_control = {}
assert res.cache_control.properties == {}
def test_cache_control_del():
res = Response()
del res.cache_control
assert repr(res.cache_control) == "<CacheControl ''>"
def test_body_file_get():
res = Response()
result = res.body_file
from webob.response import ResponseBodyFile
assert result.__class__ == ResponseBodyFile
def test_body_file_write_no_charset():
res = Response
with pytest.raises(TypeError):
res.write(text_('foo'))
def test_body_file_write_unicode_encodes():
s = text_(b'La Pe\xc3\xb1a', 'utf-8')
res = Response()
res.write(s)
assert res.app_iter, [b'' == b'La Pe\xc3\xb1a']
def test_repr():
res = Response()
assert repr(res).endswith('200 OK>')
def test_cache_expires_set_timedelta():
res = Response()
from datetime import timedelta
delta = timedelta(seconds=60)
res.cache_expires(seconds=delta)
assert res.cache_control.max_age == 60
def test_cache_expires_set_int():
res = Response()
res.cache_expires(seconds=60)
assert res.cache_control.max_age == 60
def test_cache_expires_set_None():
res = Response()
res.cache_expires(seconds=None, a=1)
assert res.cache_control.a == 1
def test_cache_expires_set_zero():
res = Response()
res.cache_expires(seconds=0)
assert res.cache_control.no_store is True
assert res.cache_control.no_cache == '*'
assert res.cache_control.must_revalidate is True
assert res.cache_control.max_age == 0
assert res.cache_control.post_check == 0
def test_encode_content_unknown():
res = Response()
with pytest.raises(AssertionError):
res.encode_content('badencoding')
def test_encode_content_identity():
res = Response()
result = res.encode_content('identity')
assert result is None
def test_encode_content_gzip_already_gzipped():
res = Response()
res.content_encoding = 'gzip'
result = res.encode_content('gzip')
assert result is None
def test_encode_content_gzip_notyet_gzipped():
res = Response()
res.app_iter = io.BytesIO(b'foo')
result = res.encode_content('gzip')
assert result is None
assert res.content_length == 23
assert res.app_iter == [
b'\x1f\x8b\x08\x00\x00\x00\x00\x00\x02\xff',
b'K\xcb\xcf\x07\x00',
b'!es\x8c\x03\x00\x00\x00'
]
def test_encode_content_gzip_notyet_gzipped_lazy():
res = Response()
res.app_iter = io.BytesIO(b'foo')
result = res.encode_content('gzip', lazy=True)
assert result is None
assert res.content_length is None
assert list(res.app_iter) == [
b'\x1f\x8b\x08\x00\x00\x00\x00\x00\x02\xff',
b'K\xcb\xcf\x07\x00',
b'!es\x8c\x03\x00\x00\x00'
]
def test_encode_content_gzip_buffer_coverage():
# this test is to provide 100% coverage of
# Response.encode_content was necessary in order to get
# request https://github.com/Pylons/webob/pull/85 into upstream
res = Response()
DATA = b"abcdefghijklmnopqrstuvwxyz0123456789" * 1000000
res.app_iter = io.BytesIO(DATA)
res.encode_content('gzip')
result = list(res.app_iter)
assert len(b"".join(result)) < len(DATA)
def test_decode_content_identity():
res = Response()
res.content_encoding = 'identity'
result = res.decode_content()
assert result is None
def test_decode_content_weird():
res = Response()
res.content_encoding = 'weird'
with pytest.raises(ValueError):
res.decode_content()
def test_decode_content_gzip():
from gzip import GzipFile
io_ = io.BytesIO()
gzip_f = GzipFile(filename='', mode='w', fileobj=io_)
gzip_f.write(b'abc')
gzip_f.close()
body = io_.getvalue()
res = Response()
res.content_encoding = 'gzip'
res.body = body
res.decode_content()
assert res.body == b'abc'
def test__make_location_absolute_has_scheme_only():
result = Response._make_location_absolute(
{
'wsgi.url_scheme': 'http',
'HTTP_HOST': 'example.com:80'
},
'http:'
)
assert result == 'http:'
def test__make_location_absolute_path():
result = Response._make_location_absolute(
{
'wsgi.url_scheme': 'http',
'HTTP_HOST': 'example.com:80'
},
'/abc'
)
assert result == 'http://example.com/abc'
def test__make_location_absolute_already_absolute():
result = Response._make_location_absolute(
{
'wsgi.url_scheme': 'http',
'HTTP_HOST': 'example.com:80'
},
'https://funcptr.net/'
)
assert result == 'https://funcptr.net/'
def test_response_set_body_file1():
data = b'abc'
file = io.BytesIO(data)
r = Response(body_file=file)
assert r.body == data
def test_response_set_body_file2():
data = b'abcdef' * 1024
file = io.BytesIO(data)
r = Response(body_file=file)
assert r.body == data
def test_response_json_body():
r = Response(json_body={'a': 1})
assert r.body == b'{"a":1}', repr(r.body)
assert r.content_type == 'application/json'
r = Response()
r.json_body = {"b": 1}
assert r.content_type == 'text/html'
del r.json_body
assert r.body == b''
def test_cache_expires_set_zero_then_nonzero():
res = Response()
res.cache_expires(seconds=0)
res.cache_expires(seconds=1)
assert res.pragma is None
assert not res.cache_control.no_cache
assert not res.cache_control.no_store
assert not res.cache_control.must_revalidate
assert res.cache_control.max_age == 1
def test_default_content_type():
class NoDefault(Response):
default_content_type = None
res = NoDefault()
assert res.content_type is None
def test_default_charset():
class DefaultCharset(Response):
default_charset = 'UTF-16'
res = DefaultCharset()
assert res.content_type == 'text/html'
assert res.charset == 'UTF-16'
assert res.headers['Content-Type'] == 'text/html; charset=UTF-16'
def test_header_list_no_defaults():
res = Response(headerlist=[])
assert res.headerlist == [('Content-Length', '0')]
assert res.content_type is None
assert res.charset is None
assert res.body == b''
def test_204_has_no_body():
res = Response(status='204 No Content')
assert res.body == b''
assert res.content_length is None
assert res.headerlist == []
def test_204_app_iter_set():
res = Response(status='204', app_iter=[b'test'])
# You are on your own in this case... you set app_iter you bought it
assert res.body == b'test'
assert res.content_length is None
assert res.headerlist == []
def test_explicit_charset():
res = Response(charset='UTF-16')
assert res.content_type == 'text/html'
assert res.charset == 'UTF-16'
def test_set_content_type():
res = Response(content_type='application/json')
res.content_type = 'application/foo'
assert res.content_type == 'application/foo'
def test_raises_no_charset():
with pytest.raises(TypeError):
Response(content_type='image/jpeg', body=text_(b'test'))
def test_raises_none_charset():
with pytest.raises(TypeError):
Response(
content_type='image/jpeg',
body=text_(b'test'),
charset=None)
def test_doesnt_raise_with_charset_content_type_has_no_charset():
res = Response(content_type='image/jpeg', body=text_(b'test'), charset='utf-8')
assert res.body == b'test'
assert res.content_type == 'image/jpeg'
assert res.charset is None
def test_content_type_has_charset():
res = Response(content_type='application/foo; charset=UTF-8', body=text_(b'test'))
assert res.body == b'test'
assert res.content_type == 'application/foo'
assert res.charset == 'UTF-8'
assert res.headers['Content-Type'] == 'application/foo; charset=UTF-8'
def test_app_iter_is_same():
class app_iter(object):
pass
my_app_iter = app_iter()
res = Response(status=204, app_iter=my_app_iter)
assert res.app_iter == my_app_iter
assert isinstance(res.app_iter, app_iter)