monitor/cherrypy/test/test_static.py

   1 from cherrypy._cpcompat import HTTPConnection, HTTPSConnection, ntob
   2 from cherrypy._cpcompat import BytesIO
   3
   4 import os
   5 curdir = os.path.join(os.getcwd(), os.path.dirname(__file__))
   6 has_space_filepath = os.path.join(curdir, 'static', 'has space.html')
   7 bigfile_filepath = os.path.join(curdir, "static", "bigfile.log")
   8 BIGFILE_SIZE = 1024 * 1024
   9 import threading
  10
  11 import cherrypy
  12 from cherrypy.lib import static
  13 from cherrypy.test import helper
  14
  15
  16 class StaticTest(helper.CPWebCase):
  17
  18     def setup_server():
  19         if not os.path.exists(has_space_filepath):
  20             open(has_space_filepath, 'wb').write(ntob('Hello, world\r\n'))
  21         if not os.path.exists(bigfile_filepath):
  22             open(bigfile_filepath, 'wb').write(ntob("x" * BIGFILE_SIZE))
  23
  24         class Root:
  25
  26             def bigfile(self):
  27                 from cherrypy.lib import static
  28                 self.f = static.serve_file(bigfile_filepath)
  29                 return self.f
  30             bigfile.exposed = True
  31             bigfile._cp_config = {'response.stream': True}
  32
  33             def tell(self):
  34                 if self.f.input.closed:
  35                     return ''
  36                 return repr(self.f.input.tell()).rstrip('L')
  37             tell.exposed = True
  38
  39             def fileobj(self):
  40                 f = open(os.path.join(curdir, 'style.css'), 'rb')
  41                 return static.serve_fileobj(f, content_type='text/css')
  42             fileobj.exposed = True
  43
  44             def bytesio(self):
  45                 f = BytesIO(ntob('Fee\nfie\nfo\nfum'))
  46                 return static.serve_fileobj(f, content_type='text/plain')
  47             bytesio.exposed = True
  48
  49         class Static:
  50
  51             def index(self):
  52                 return 'You want the Baron? You can have the Baron!'
  53             index.exposed = True
  54
  55             def dynamic(self):
  56                 return "This is a DYNAMIC page"
  57             dynamic.exposed = True
  58
  59
  60         root = Root()
  61         root.static = Static()
  62
  63         rootconf = {
  64             '/static': {
  65                 'tools.staticdir.on': True,
  66                 'tools.staticdir.dir': 'static',
  67                 'tools.staticdir.root': curdir,
  68             },
  69             '/style.css': {
  70                 'tools.staticfile.on': True,
  71                 'tools.staticfile.filename': os.path.join(curdir, 'style.css'),
  72             },
  73             '/docroot': {
  74                 'tools.staticdir.on': True,
  75                 'tools.staticdir.root': curdir,
  76                 'tools.staticdir.dir': 'static',
  77                 'tools.staticdir.index': 'index.html',
  78             },
  79             '/error': {
  80                 'tools.staticdir.on': True,
  81                 'request.show_tracebacks': True,
  82             },
  83             }
  84         rootApp = cherrypy.Application(root)
  85         rootApp.merge(rootconf)
  86
  87         test_app_conf = {
  88             '/test': {
  89                 'tools.staticdir.index': 'index.html',
  90                 'tools.staticdir.on': True,
  91                 'tools.staticdir.root': curdir,
  92                 'tools.staticdir.dir': 'static',
  93                 },
  94             }
  95         testApp = cherrypy.Application(Static())
  96         testApp.merge(test_app_conf)
  97
  98         vhost = cherrypy._cpwsgi.VirtualHost(rootApp, {'virt.net': testApp})
  99         cherrypy.tree.graft(vhost)
 100     setup_server = staticmethod(setup_server)
 101
 102
 103     def teardown_server():
 104         for f in (has_space_filepath, bigfile_filepath):
 105             if os.path.exists(f):
 106                 try:
 107                     os.unlink(f)
 108                 except:
 109                     pass
 110     teardown_server = staticmethod(teardown_server)
 111
 112
 113     def testStatic(self):
 114         self.getPage("/static/index.html")
 115         self.assertStatus('200 OK')
 116         self.assertHeader('Content-Type', 'text/html')
 117         self.assertBody('Hello, world\r\n')
 118
 119         # Using a staticdir.root value in a subdir...
 120         self.getPage("/docroot/index.html")
 121         self.assertStatus('200 OK')
 122         self.assertHeader('Content-Type', 'text/html')
 123         self.assertBody('Hello, world\r\n')
 124
 125         # Check a filename with spaces in it
 126         self.getPage("/static/has%20space.html")
 127         self.assertStatus('200 OK')
 128         self.assertHeader('Content-Type', 'text/html')
 129         self.assertBody('Hello, world\r\n')
 130
 131         self.getPage("/style.css")
 132         self.assertStatus('200 OK')
 133         self.assertHeader('Content-Type', 'text/css')
 134         # Note: The body should be exactly 'Dummy stylesheet\n', but
 135         #   unfortunately some tools such as WinZip sometimes turn \n
 136         #   into \r\n on Windows when extracting the CherryPy tarball so
 137         #   we just check the content
 138         self.assertMatchesBody('^Dummy stylesheet')
 139
 140     def test_fallthrough(self):
 141         # Test that NotFound will then try dynamic handlers (see [878]).
 142         self.getPage("/static/dynamic")
 143         self.assertBody("This is a DYNAMIC page")
 144
 145         # Check a directory via fall-through to dynamic handler.
 146         self.getPage("/static/")
 147         self.assertStatus('200 OK')
 148         self.assertHeader('Content-Type', 'text/html;charset=utf-8')
 149         self.assertBody('You want the Baron? You can have the Baron!')
 150
 151     def test_index(self):
 152         # Check a directory via "staticdir.index".
 153         self.getPage("/docroot/")
 154         self.assertStatus('200 OK')
 155         self.assertHeader('Content-Type', 'text/html')
 156         self.assertBody('Hello, world\r\n')
 157         # The same page should be returned even if redirected.
 158         self.getPage("/docroot")
 159         self.assertStatus(301)
 160         self.assertHeader('Location', '%s/docroot/' % self.base())
 161         self.assertMatchesBody("This resource .* <a href='%s/docroot/'>"
 162                                "%s/docroot/</a>." % (self.base(), self.base()))
 163
 164     def test_config_errors(self):
 165         # Check that we get an error if no .file or .dir
 166         self.getPage("/error/thing.html")
 167         self.assertErrorPage(500)
 168         self.assertMatchesBody(ntob("TypeError: staticdir\(\) takes at least 2 "
 169                                     "(positional )?arguments \(0 given\)"))
 170
 171     def test_security(self):
 172         # Test up-level security
 173         self.getPage("/static/../../test/style.css")
 174         self.assertStatus((400, 403))
 175
 176     def test_modif(self):
 177         # Test modified-since on a reasonably-large file
 178         self.getPage("/static/dirback.jpg")
 179         self.assertStatus("200 OK")
 180         lastmod = ""
 181         for k, v in self.headers:
 182             if k == 'Last-Modified':
 183                 lastmod = v
 184         ims = ("If-Modified-Since", lastmod)
 185         self.getPage("/static/dirback.jpg", headers=[ims])
 186         self.assertStatus(304)
 187         self.assertNoHeader("Content-Type")
 188         self.assertNoHeader("Content-Length")
 189         self.assertNoHeader("Content-Disposition")
 190         self.assertBody("")
 191
 192     def test_755_vhost(self):
 193         self.getPage("/test/", [('Host', 'virt.net')])
 194         self.assertStatus(200)
 195         self.getPage("/test", [('Host', 'virt.net')])
 196         self.assertStatus(301)
 197         self.assertHeader('Location', self.scheme + '://virt.net/test/')
 198
 199     def test_serve_fileobj(self):
 200         self.getPage("/fileobj")
 201         self.assertStatus('200 OK')
 202         self.assertHeader('Content-Type', 'text/css;charset=utf-8')
 203         self.assertMatchesBody('^Dummy stylesheet')
 204
 205     def test_serve_bytesio(self):
 206         self.getPage("/bytesio")
 207         self.assertStatus('200 OK')
 208         self.assertHeader('Content-Type', 'text/plain;charset=utf-8')
 209         self.assertHeader('Content-Length', 14)
 210         self.assertMatchesBody('Fee\nfie\nfo\nfum')
 211
 212     def test_file_stream(self):
 213         if cherrypy.server.protocol_version != "HTTP/1.1":
 214             return self.skip()
 215
 216         self.PROTOCOL = "HTTP/1.1"
 217
 218         # Make an initial request
 219         self.persistent = True
 220         conn = self.HTTP_CONN
 221         conn.putrequest("GET", "/bigfile", skip_host=True)
 222         conn.putheader("Host", self.HOST)
 223         conn.endheaders()
 224         response = conn.response_class(conn.sock, method="GET")
 225         response.begin()
 226         self.assertEqual(response.status, 200)
 227
 228         body = ntob('')
 229         remaining = BIGFILE_SIZE
 230         while remaining > 0:
 231             data = response.fp.read(65536)
 232             if not data:
 233                 break
 234             body += data
 235             remaining -= len(data)
 236
 237             if self.scheme == "https":
 238                 newconn = HTTPSConnection
 239             else:
 240                 newconn = HTTPConnection
 241             s, h, b = helper.webtest.openURL(
 242                 ntob("/tell"), headers=[], host=self.HOST, port=self.PORT,
 243                 http_conn=newconn)
 244             if not b:
 245                 # The file was closed on the server.
 246                 tell_position = BIGFILE_SIZE
 247             else:
 248                 tell_position = int(b)
 249
 250             expected = len(body)
 251             if tell_position >= BIGFILE_SIZE:
 252                 # We can't exactly control how much content the server asks for.
 253                 # Fudge it by only checking the first half of the reads.
 254                 if expected < (BIGFILE_SIZE / 2):
 255                     self.fail(
 256                         "The file should have advanced to position %r, but has "
 257                         "already advanced to the end of the file. It may not be "
 258                         "streamed as intended, or at the wrong chunk size (64k)" %
 259                         expected)
 260             elif tell_position < expected:
 261                 self.fail(
 262                     "The file should have advanced to position %r, but has "
 263                     "only advanced to position %r. It may not be streamed "
 264                     "as intended, or at the wrong chunk size (65536)" %
 265                     (expected, tell_position))
 266
 267         if body != ntob("x" * BIGFILE_SIZE):
 268             self.fail("Body != 'x' * %d. Got %r instead (%d bytes)." %
 269                       (BIGFILE_SIZE, body[:50], len(body)))
 270         conn.close()
 271
 272     def test_file_stream_deadlock(self):
 273         if cherrypy.server.protocol_version != "HTTP/1.1":
 274             return self.skip()
 275
 276         self.PROTOCOL = "HTTP/1.1"
 277
 278         # Make an initial request but abort early.
 279         self.persistent = True
 280         conn = self.HTTP_CONN
 281         conn.putrequest("GET", "/bigfile", skip_host=True)
 282         conn.putheader("Host", self.HOST)
 283         conn.endheaders()
 284         response = conn.response_class(conn.sock, method="GET")
 285         response.begin()
 286         self.assertEqual(response.status, 200)
 287         body = response.fp.read(65536)
 288         if body != ntob("x" * len(body)):
 289             self.fail("Body != 'x' * %d. Got %r instead (%d bytes)." %
 290                       (65536, body[:50], len(body)))
 291         response.close()
 292         conn.close()
 293
 294         # Make a second request, which should fetch the whole file.
 295         self.persistent = False
 296         self.getPage("/bigfile")
 297         if self.body != ntob("x" * BIGFILE_SIZE):
 298             self.fail("Body != 'x' * %d. Got %r instead (%d bytes)." %
 299                       (BIGFILE_SIZE, self.body[:50], len(body)))
 300