| from io import BytesIO |
| from os.path import getmtime |
| import tempfile |
| from time import gmtime |
| import os |
| import shutil |
| import unittest |
| |
| from webob import static |
| from webob.compat import bytes_ |
| from webob.request import Request, environ_from_url |
| from webob.response import Response |
| |
| |
| def get_response(app, path='/', **req_kw): |
| """Convenient function to query an application""" |
| req = Request(environ_from_url(path), **req_kw) |
| return req.get_response(app) |
| |
| |
| def create_file(content, *paths): |
| """Convenient function to create a new file with some content""" |
| path = os.path.join(*paths) |
| with open(path, 'wb') as fp: |
| fp.write(bytes_(content)) |
| return path |
| |
| |
| class TestFileApp(unittest.TestCase): |
| def setUp(self): |
| fp = tempfile.NamedTemporaryFile(suffix=".py", delete=False) |
| self.tempfile = fp.name |
| fp.write(b"import this\n") |
| fp.close() |
| |
| def tearDown(self): |
| os.unlink(self.tempfile) |
| |
| def test_fileapp(self): |
| app = static.FileApp(self.tempfile) |
| resp1 = get_response(app) |
| assert resp1.content_type in ('text/x-python', 'text/plain') |
| self.assertEqual(resp1.charset, 'UTF-8') |
| self.assertEqual(resp1.last_modified.timetuple(), gmtime(getmtime(self.tempfile))) |
| self.assertEqual(resp1.body, b"import this\n") |
| |
| resp2 = get_response(app) |
| assert resp2.content_type in ('text/x-python', 'text/plain') |
| self.assertEqual(resp2.last_modified.timetuple(), gmtime(getmtime(self.tempfile))) |
| self.assertEqual(resp2.body, b"import this\n") |
| |
| resp3 = get_response(app, range=(7, 11)) |
| self.assertEqual(resp3.status_code, 206) |
| self.assertEqual(tuple(resp3.content_range)[:2], (7, 11)) |
| self.assertEqual(resp3.last_modified.timetuple(), gmtime(getmtime(self.tempfile))) |
| self.assertEqual(resp3.body, bytes_('this')) |
| |
| def test_unexisting_file(self): |
| app = static.FileApp('/tmp/this/doesnt/exist') |
| self.assertEqual(404, get_response(app).status_code) |
| |
| def test_allowed_methods(self): |
| app = static.FileApp(self.tempfile) |
| |
| # Alias |
| resp = lambda method: get_response(app, method=method) |
| |
| self.assertEqual(200, resp(method='GET').status_code) |
| self.assertEqual(200, resp(method='HEAD').status_code) |
| self.assertEqual(405, resp(method='POST').status_code) |
| # Actually any other method is not allowed |
| self.assertEqual(405, resp(method='xxx').status_code) |
| |
| def test_exception_while_opening_file(self): |
| # Mock the built-in ``open()`` function to allow finner control about |
| # what we are testing. |
| def open_ioerror(*args, **kwargs): |
| raise IOError() |
| |
| def open_oserror(*args, **kwargs): |
| raise OSError() |
| |
| app = static.FileApp(self.tempfile) |
| |
| app._open = open_ioerror |
| self.assertEqual(403, get_response(app).status_code) |
| |
| app._open = open_oserror |
| self.assertEqual(403, get_response(app).status_code) |
| |
| def test_use_wsgi_filewrapper(self): |
| class TestWrapper(object): |
| def __init__(self, file, block_size): |
| self.file = file |
| self.block_size = block_size |
| |
| environ = environ_from_url('/') |
| environ['wsgi.file_wrapper'] = TestWrapper |
| app = static.FileApp(self.tempfile) |
| app_iter = Request(environ).get_response(app).app_iter |
| |
| self.assertTrue(isinstance(app_iter, TestWrapper)) |
| self.assertEqual(bytes_('import this\n'), app_iter.file.read()) |
| self.assertEqual(static.BLOCK_SIZE, app_iter.block_size) |
| |
| |
| class TestFileIter(unittest.TestCase): |
| def test_empty_file(self): |
| fp = BytesIO() |
| fi = static.FileIter(fp) |
| self.assertRaises(StopIteration, next, iter(fi)) |
| |
| def test_seek(self): |
| fp = BytesIO(bytes_("0123456789")) |
| i = static.FileIter(fp).app_iter_range(seek=4) |
| |
| self.assertEqual(bytes_("456789"), next(i)) |
| self.assertRaises(StopIteration, next, i) |
| |
| def test_limit(self): |
| fp = BytesIO(bytes_("0123456789")) |
| i = static.FileIter(fp).app_iter_range(limit=4) |
| |
| self.assertEqual(bytes_("0123"), next(i)) |
| self.assertRaises(StopIteration, next, i) |
| |
| def test_limit_and_seek(self): |
| fp = BytesIO(bytes_("0123456789")) |
| i = static.FileIter(fp).app_iter_range(limit=4, seek=1) |
| |
| self.assertEqual(bytes_("123"), next(i)) |
| self.assertRaises(StopIteration, next, i) |
| |
| def test_multiple_reads(self): |
| fp = BytesIO(bytes_("012")) |
| i = static.FileIter(fp).app_iter_range(block_size=1) |
| |
| self.assertEqual(bytes_("0"), next(i)) |
| self.assertEqual(bytes_("1"), next(i)) |
| self.assertEqual(bytes_("2"), next(i)) |
| self.assertRaises(StopIteration, next, i) |
| |
| def test_seek_bigger_than_limit(self): |
| fp = BytesIO(bytes_("0123456789")) |
| i = static.FileIter(fp).app_iter_range(limit=1, seek=2) |
| |
| # XXX: this should not return anything actually, since we are starting |
| # to read after the place we wanted to stop. |
| self.assertEqual(bytes_("23456789"), next(i)) |
| self.assertRaises(StopIteration, next, i) |
| |
| def test_limit_is_zero(self): |
| fp = BytesIO(bytes_("0123456789")) |
| i = static.FileIter(fp).app_iter_range(limit=0) |
| |
| self.assertRaises(StopIteration, next, i) |
| |
| |
| |
| class TestDirectoryApp(unittest.TestCase): |
| def setUp(self): |
| self.test_dir = tempfile.mkdtemp() |
| |
| def tearDown(self): |
| shutil.rmtree(self.test_dir) |
| |
| def test_empty_directory(self): |
| app = static.DirectoryApp(self.test_dir) |
| self.assertEqual(404, get_response(app).status_code) |
| self.assertEqual(404, get_response(app, '/foo').status_code) |
| |
| def test_serve_file(self): |
| app = static.DirectoryApp(self.test_dir) |
| create_file('abcde', self.test_dir, 'bar') |
| self.assertEqual(404, get_response(app).status_code) |
| self.assertEqual(404, get_response(app, '/foo').status_code) |
| |
| resp = get_response(app, '/bar') |
| self.assertEqual(200, resp.status_code) |
| self.assertEqual(bytes_('abcde'), resp.body) |
| |
| def test_dont_serve_file_in_parent_directory(self): |
| # We'll have: |
| # /TEST_DIR/ |
| # /TEST_DIR/bar |
| # /TEST_DIR/foo/ <- serve this directory |
| create_file('abcde', self.test_dir, 'bar') |
| serve_path = os.path.join(self.test_dir, 'foo') |
| os.mkdir(serve_path) |
| app = static.DirectoryApp(serve_path) |
| |
| # The file exists, but is outside the served dir. |
| self.assertEqual(403, get_response(app, '/../bar').status_code) |
| |
| def test_dont_leak_parent_directory_file_existance(self): |
| # We'll have: |
| # /TEST_DIR/ |
| # /TEST_DIR/foo/ <- serve this directory |
| serve_path = os.path.join(self.test_dir, 'foo') |
| os.mkdir(serve_path) |
| app = static.DirectoryApp(serve_path) |
| |
| # The file exists, but is outside the served dir. |
| self.assertEqual(403, get_response(app, '/../bar2').status_code) |
| |
| def test_file_app_arguments(self): |
| app = static.DirectoryApp(self.test_dir, content_type='xxx/yyy') |
| create_file('abcde', self.test_dir, 'bar') |
| |
| resp = get_response(app, '/bar') |
| self.assertEqual(200, resp.status_code) |
| self.assertEqual('xxx/yyy', resp.content_type) |
| |
| def test_file_app_factory(self): |
| def make_fileapp(*args, **kwargs): |
| make_fileapp.called = True |
| return Response() |
| make_fileapp.called = False |
| |
| app = static.DirectoryApp(self.test_dir) |
| app.make_fileapp = make_fileapp |
| create_file('abcde', self.test_dir, 'bar') |
| |
| get_response(app, '/bar') |
| self.assertTrue(make_fileapp.called) |
| |
| def test_must_serve_directory(self): |
| serve_path = create_file('abcde', self.test_dir, 'bar') |
| self.assertRaises(IOError, static.DirectoryApp, serve_path) |
| |
| def test_index_page(self): |
| os.mkdir(os.path.join(self.test_dir, 'index-test')) |
| create_file(bytes_('index'), self.test_dir, 'index-test', 'index.html') |
| app = static.DirectoryApp(self.test_dir) |
| resp = get_response(app, '/index-test') |
| self.assertEqual(resp.status_code, 301) |
| self.assertTrue(resp.location.endswith('/index-test/')) |
| resp = get_response(app, '/index-test?test') |
| self.assertTrue(resp.location.endswith('/index-test/?test')) |
| resp = get_response(app, '/index-test/') |
| self.assertEqual(resp.status_code, 200) |
| self.assertEqual(resp.body, bytes_('index')) |
| self.assertEqual(resp.content_type, 'text/html') |
| resp = get_response(app, '/index-test/index.html') |
| self.assertEqual(resp.status_code, 200) |
| self.assertEqual(resp.body, bytes_('index')) |
| redir_app = static.DirectoryApp(self.test_dir, hide_index_with_redirect=True) |
| resp = get_response(redir_app, '/index-test/index.html') |
| self.assertEqual(resp.status_code, 301) |
| self.assertTrue(resp.location.endswith('/index-test/')) |
| resp = get_response(redir_app, '/index-test/index.html?test') |
| self.assertTrue(resp.location.endswith('/index-test/?test')) |
| page_app = static.DirectoryApp(self.test_dir, index_page='something-else.html') |
| self.assertEqual(get_response(page_app, '/index-test/').status_code, 404) |
