Source code for cherrypy.test.test_static

# -*- coding: utf-8 -*-
import io
import os
import sys
import re
import platform
import tempfile
import urllib.parse
import unittest.mock
from http.client import HTTPConnection

import pytest
import py.path
import path

import cherrypy
from cherrypy.lib import static
from cherrypy._cpcompat import HTTPSConnection, ntou, tonative
from cherrypy.test import helper


[docs]@pytest.fixture def unicode_filesystem(tmpdir): _check_unicode_filesystem(tmpdir)
def _check_unicode_filesystem(tmpdir): filename = tmpdir / ntou('☃', 'utf-8') tmpl = 'File system encoding ({encoding}) cannot support unicode filenames' msg = tmpl.format(encoding=sys.getfilesystemencoding()) try: io.open(str(filename), 'w').close() except UnicodeEncodeError: pytest.skip(msg)
[docs]def ensure_unicode_filesystem(): """ TODO: replace with simply pytest fixtures once webtest.TestCase no longer implies unittest. """ tmpdir = py.path.local(tempfile.mkdtemp()) try: _check_unicode_filesystem(tmpdir) finally: tmpdir.remove()
curdir = path.Path(__file__).dirname() has_space_filepath = curdir / 'static' / 'has space.html' bigfile_filepath = curdir / 'static' / 'bigfile.log' # The file size needs to be big enough such that half the size of it # won't be socket-buffered (or server-buffered) all in one go. See # test_file_stream. MB = 2 ** 20 BIGFILE_SIZE = 32 * MB
[docs]class StaticTest(helper.CPWebCase): files_to_remove = []
[docs] @staticmethod def setup_server(): if not os.path.exists(has_space_filepath): with open(has_space_filepath, 'wb') as f: f.write(b'Hello, world\r\n') needs_bigfile = ( not os.path.exists(bigfile_filepath) or os.path.getsize(bigfile_filepath) != BIGFILE_SIZE ) if needs_bigfile: with open(bigfile_filepath, 'wb') as f: f.write(b'x' * BIGFILE_SIZE) class Root: @cherrypy.expose @cherrypy.config(**{'response.stream': True}) def bigfile(self): self.f = static.serve_file(bigfile_filepath) return self.f @cherrypy.expose def tell(self): if self.f.input.closed: return '' return repr(self.f.input.tell()).rstrip('L') @cherrypy.expose def fileobj(self): f = open(os.path.join(curdir, 'style.css'), 'rb') return static.serve_fileobj(f, content_type='text/css') @cherrypy.expose def bytesio(self): f = io.BytesIO(b'Fee\nfie\nfo\nfum') return static.serve_fileobj(f, content_type='text/plain') class Static: @cherrypy.expose def index(self): return 'You want the Baron? You can have the Baron!' @cherrypy.expose def dynamic(self): return 'This is a DYNAMIC page' root = Root() root.static = Static() rootconf = { '/static': { 'tools.staticdir.on': True, 'tools.staticdir.dir': 'static', 'tools.staticdir.root': curdir, }, '/static-long': { 'tools.staticdir.on': True, 'tools.staticdir.dir': r'\\?\%s' % curdir, }, '/style.css': { 'tools.staticfile.on': True, 'tools.staticfile.filename': os.path.join(curdir, 'style.css'), }, '/docroot': { 'tools.staticdir.on': True, 'tools.staticdir.root': curdir, 'tools.staticdir.dir': 'static', 'tools.staticdir.index': 'index.html', }, '/error': { 'tools.staticdir.on': True, 'request.show_tracebacks': True, }, '/404test': { 'tools.staticdir.on': True, 'tools.staticdir.root': curdir, 'tools.staticdir.dir': 'static', 'error_page.404': error_page_404, } } rootApp = cherrypy.Application(root) rootApp.merge(rootconf) test_app_conf = { '/test': { 'tools.staticdir.index': 'index.html', 'tools.staticdir.on': True, 'tools.staticdir.root': curdir, 'tools.staticdir.dir': 'static', }, } testApp = cherrypy.Application(Static()) testApp.merge(test_app_conf) vhost = cherrypy._cpwsgi.VirtualHost(rootApp, {'virt.net': testApp}) cherrypy.tree.graft(vhost)
[docs] @classmethod def teardown_class(cls): super(cls, cls).teardown_class() files_to_remove = has_space_filepath, bigfile_filepath files_to_remove += tuple(cls.files_to_remove) for file in files_to_remove: file.remove_p()
[docs] def test_static(self): self.getPage('/static/index.html') self.assertStatus('200 OK') self.assertHeader('Content-Type', 'text/html') self.assertBody('Hello, world\r\n') # Using a staticdir.root value in a subdir... self.getPage('/docroot/index.html') self.assertStatus('200 OK') self.assertHeader('Content-Type', 'text/html') self.assertBody('Hello, world\r\n') # Check a filename with spaces in it self.getPage('/static/has%20space.html') self.assertStatus('200 OK') self.assertHeader('Content-Type', 'text/html') self.assertBody('Hello, world\r\n') self.getPage('/style.css') self.assertStatus('200 OK') self.assertHeader('Content-Type', 'text/css') # Note: The body should be exactly 'Dummy stylesheet\n', but # unfortunately some tools such as WinZip sometimes turn \n # into \r\n on Windows when extracting the CherryPy tarball so # we just check the content self.assertMatchesBody('^Dummy stylesheet')
[docs] @pytest.mark.skipif(platform.system() != 'Windows', reason='Windows only') def test_static_longpath(self): """Test serving of a file in subdir of a Windows long-path staticdir.""" self.getPage('/static-long/static/index.html') self.assertStatus('200 OK') self.assertHeader('Content-Type', 'text/html') self.assertBody('Hello, world\r\n')
[docs] def test_fallthrough(self): # Test that NotFound will then try dynamic handlers (see [878]). self.getPage('/static/dynamic') self.assertBody('This is a DYNAMIC page') # Check a directory via fall-through to dynamic handler. self.getPage('/static/') self.assertStatus('200 OK') self.assertHeader('Content-Type', 'text/html;charset=utf-8') self.assertBody('You want the Baron? You can have the Baron!')
[docs] def test_index(self): # Check a directory via "staticdir.index". self.getPage('/docroot/') self.assertStatus('200 OK') self.assertHeader('Content-Type', 'text/html') self.assertBody('Hello, world\r\n') # The same page should be returned even if redirected. self.getPage('/docroot') self.assertStatus(301) self.assertHeader('Location', '%s/docroot/' % self.base()) self.assertMatchesBody( "This resource .* <a href=(['\"])%s/docroot/\\1>" '%s/docroot/</a>.' % (self.base(), self.base()) )
[docs] def test_config_errors(self): # Check that we get an error if no .file or .dir self.getPage('/error/thing.html') self.assertErrorPage(500) if sys.version_info >= (3, 3): errmsg = ( r'TypeError: staticdir\(\) missing 2 ' 'required positional arguments' ) else: errmsg = ( r'TypeError: staticdir\(\) takes at least 2 ' r'(positional )?arguments \(0 given\)' ) self.assertMatchesBody(errmsg.encode('ascii'))
[docs] def test_security(self): # Test up-level security self.getPage('/static/../../test/style.css') self.assertStatus((400, 403))
[docs] def test_modif(self): # Test modified-since on a reasonably-large file self.getPage('/static/dirback.jpg') self.assertStatus('200 OK') lastmod = '' for k, v in self.headers: if k == 'Last-Modified': lastmod = v ims = ('If-Modified-Since', lastmod) self.getPage('/static/dirback.jpg', headers=[ims]) self.assertStatus(304) self.assertNoHeader('Content-Type') self.assertNoHeader('Content-Length') self.assertNoHeader('Content-Disposition') self.assertBody('')
[docs] def test_755_vhost(self): self.getPage('/test/', [('Host', 'virt.net')]) self.assertStatus(200) self.getPage('/test', [('Host', 'virt.net')]) self.assertStatus(301) self.assertHeader('Location', self.scheme + '://virt.net/test/')
[docs] def test_serve_fileobj(self): self.getPage('/fileobj') self.assertStatus('200 OK') self.assertHeader('Content-Type', 'text/css;charset=utf-8') self.assertMatchesBody('^Dummy stylesheet')
[docs] def test_serve_bytesio(self): self.getPage('/bytesio') self.assertStatus('200 OK') self.assertHeader('Content-Type', 'text/plain;charset=utf-8') self.assertHeader('Content-Length', 14) self.assertMatchesBody('Fee\nfie\nfo\nfum')
[docs] @pytest.mark.xfail(reason='#1475') def test_file_stream(self): if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() self.PROTOCOL = 'HTTP/1.1' # Make an initial request self.persistent = True conn = self.HTTP_CONN conn.putrequest('GET', '/bigfile', skip_host=True) conn.putheader('Host', self.HOST) conn.endheaders() response = conn.response_class(conn.sock, method='GET') response.begin() self.assertEqual(response.status, 200) body = b'' remaining = BIGFILE_SIZE while remaining > 0: data = response.fp.read(65536) if not data: break body += data remaining -= len(data) if self.scheme == 'https': newconn = HTTPSConnection else: newconn = HTTPConnection s, h, b = helper.webtest.openURL( b'/tell', headers=[], host=self.HOST, port=self.PORT, http_conn=newconn) if not b: # The file was closed on the server. tell_position = BIGFILE_SIZE else: tell_position = int(b) read_so_far = len(body) # It is difficult for us to force the server to only read # the bytes that we ask for - there are going to be buffers # inbetween. # # CherryPy will attempt to write as much data as it can to # the socket, and we don't have a way to determine what that # size will be. So we make the following assumption - by # the time we have read in the entire file on the server, # we will have at least received half of it. If this is not # the case, then this is an indicator that either: # - machines that are running this test are using buffer # sizes greater than half of BIGFILE_SIZE; or # - streaming is broken. # # At the time of writing, we seem to have encountered # buffer sizes bigger than 512K, so we've increased # BIGFILE_SIZE to 4MB and in 2016 to 20MB and then 32MB. # This test is going to keep failing according to the # improvements in hardware and OS buffers. if tell_position >= BIGFILE_SIZE: if read_so_far < (BIGFILE_SIZE / 2): self.fail( 'The file should have advanced to position %r, but ' 'has already advanced to the end of the file. It ' 'may not be streamed as intended, or at the wrong ' 'chunk size (64k)' % read_so_far) elif tell_position < read_so_far: self.fail( 'The file should have advanced to position %r, but has ' 'only advanced to position %r. It may not be streamed ' 'as intended, or at the wrong chunk size (64k)' % (read_so_far, tell_position)) if body != b'x' * BIGFILE_SIZE: self.fail("Body != 'x' * %d. Got %r instead (%d bytes)." % (BIGFILE_SIZE, body[:50], len(body))) conn.close()
[docs] def test_file_stream_deadlock(self): if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() self.PROTOCOL = 'HTTP/1.1' # Make an initial request but abort early. self.persistent = True conn = self.HTTP_CONN conn.putrequest('GET', '/bigfile', skip_host=True) conn.putheader('Host', self.HOST) conn.endheaders() response = conn.response_class(conn.sock, method='GET') response.begin() self.assertEqual(response.status, 200) body = response.fp.read(65536) if body != b'x' * len(body): self.fail("Body != 'x' * %d. Got %r instead (%d bytes)." % (65536, body[:50], len(body))) response.close() conn.close() # Make a second request, which should fetch the whole file. self.persistent = False self.getPage('/bigfile') if self.body != b'x' * BIGFILE_SIZE: self.fail("Body != 'x' * %d. Got %r instead (%d bytes)." % (BIGFILE_SIZE, self.body[:50], len(body)))
[docs] def test_error_page_with_serve_file(self): self.getPage('/404test/yunyeen') self.assertStatus(404) self.assertInBody("I couldn't find that thing")
[docs] @unittest.mock.patch( 'http.client._contains_disallowed_url_pchar_re', re.compile(r'[\n]'), create=True, ) def test_null_bytes(self): self.getPage('/static/\x00') self.assertStatus('404 Not Found')
[docs] @classmethod def unicode_file(cls): filename = ntou('Слава Україні.html', 'utf-8') filepath = curdir / 'static' / filename with filepath.open('w', encoding='utf-8')as strm: strm.write(ntou('Героям Слава!', 'utf-8')) cls.files_to_remove.append(filepath)
[docs] def test_unicode(self): ensure_unicode_filesystem() self.unicode_file() url = ntou('/static/Слава Україні.html', 'utf-8') # quote function requires str url = tonative(url, 'utf-8') url = urllib.parse.quote(url) self.getPage(url) expected = ntou('Героям Слава!', 'utf-8') self.assertInBody(expected)
[docs]def error_page_404(status, message, traceback, version): path = os.path.join(curdir, 'static', '404.html') return static.serve_file(path, content_type='text/html')