Lib/test/test_httpservers.py

   1 """Unittests for the various HTTPServer modules.
   2
   3 Written by Cody A.W. Somerville <cody-somerville@ubuntu.com>,
   4 Josip Dzolonga, and Michael Otteneder for the 2007/08 GHOP contest.
   5 """
   6
   7 from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
   8 from SimpleHTTPServer import SimpleHTTPRequestHandler
   9 from CGIHTTPServer import CGIHTTPRequestHandler
  10 import CGIHTTPServer
  11
  12 import os
  13 import sys
  14 import base64
  15 import shutil
  16 import urllib
  17 import httplib
  18 import tempfile
  19
  20 import unittest
  21 from test import test_support
  22 threading = test_support.import_module('threading')
  23
  24
  25 class NoLogRequestHandler:
  26     def log_message(self, *args):
  27         # don't write log messages to stderr
  28         pass
  29
  30
  31 class TestServerThread(threading.Thread):
  32     def __init__(self, test_object, request_handler):
  33         threading.Thread.__init__(self)
  34         self.request_handler = request_handler
  35         self.test_object = test_object
  36
  37     def run(self):
  38         self.server = HTTPServer(('', 0), self.request_handler)
  39         self.test_object.PORT = self.server.socket.getsockname()[1]
  40         self.test_object.server_started.set()
  41         self.test_object = None
  42         try:
  43             self.server.serve_forever(0.05)
  44         finally:
  45             self.server.server_close()
  46
  47     def stop(self):
  48         self.server.shutdown()
  49
  50
  51 class BaseTestCase(unittest.TestCase):
  52     def setUp(self):
  53         self._threads = test_support.threading_setup()
  54         os.environ = test_support.EnvironmentVarGuard()
  55         self.server_started = threading.Event()
  56         self.thread = TestServerThread(self, self.request_handler)
  57         self.thread.start()
  58         self.server_started.wait()
  59
  60     def tearDown(self):
  61         self.thread.stop()
  62         os.environ.__exit__()
  63         test_support.threading_cleanup(*self._threads)
  64
  65     def request(self, uri, method='GET', body=None, headers={}):
  66         self.connection = httplib.HTTPConnection('localhost', self.PORT)
  67         self.connection.request(method, uri, body, headers)
  68         return self.connection.getresponse()
  69
  70
  71 class BaseHTTPServerTestCase(BaseTestCase):
  72     class request_handler(NoLogRequestHandler, BaseHTTPRequestHandler):
  73         protocol_version = 'HTTP/1.1'
  74         default_request_version = 'HTTP/1.1'
  75
  76         def do_TEST(self):
  77             self.send_response(204)
  78             self.send_header('Content-Type', 'text/html')
  79             self.send_header('Connection', 'close')
  80             self.end_headers()
  81
  82         def do_KEEP(self):
  83             self.send_response(204)
  84             self.send_header('Content-Type', 'text/html')
  85             self.send_header('Connection', 'keep-alive')
  86             self.end_headers()
  87
  88         def do_KEYERROR(self):
  89             self.send_error(999)
  90
  91         def do_CUSTOM(self):
  92             self.send_response(999)
  93             self.send_header('Content-Type', 'text/html')
  94             self.send_header('Connection', 'close')
  95             self.end_headers()
  96
  97     def setUp(self):
  98         BaseTestCase.setUp(self)
  99         self.con = httplib.HTTPConnection('localhost', self.PORT)
 100         self.con.connect()
 101
 102     def test_command(self):
 103         self.con.request('GET', '/')
 104         res = self.con.getresponse()
 105         self.assertEqual(res.status, 501)
 106
 107     def test_request_line_trimming(self):
 108         self.con._http_vsn_str = 'HTTP/1.1\n'
 109         self.con.putrequest('GET', '/')
 110         self.con.endheaders()
 111         res = self.con.getresponse()
 112         self.assertEqual(res.status, 501)
 113
 114     def test_version_bogus(self):
 115         self.con._http_vsn_str = 'FUBAR'
 116         self.con.putrequest('GET', '/')
 117         self.con.endheaders()
 118         res = self.con.getresponse()
 119         self.assertEqual(res.status, 400)
 120
 121     def test_version_digits(self):
 122         self.con._http_vsn_str = 'HTTP/9.9.9'
 123         self.con.putrequest('GET', '/')
 124         self.con.endheaders()
 125         res = self.con.getresponse()
 126         self.assertEqual(res.status, 400)
 127
 128     def test_version_none_get(self):
 129         self.con._http_vsn_str = ''
 130         self.con.putrequest('GET', '/')
 131         self.con.endheaders()
 132         res = self.con.getresponse()
 133         self.assertEqual(res.status, 501)
 134
 135     def test_version_none(self):
 136         self.con._http_vsn_str = ''
 137         self.con.putrequest('PUT', '/')
 138         self.con.endheaders()
 139         res = self.con.getresponse()
 140         self.assertEqual(res.status, 400)
 141
 142     def test_version_invalid(self):
 143         self.con._http_vsn = 99
 144         self.con._http_vsn_str = 'HTTP/9.9'
 145         self.con.putrequest('GET', '/')
 146         self.con.endheaders()
 147         res = self.con.getresponse()
 148         self.assertEqual(res.status, 505)
 149
 150     def test_send_blank(self):
 151         self.con._http_vsn_str = ''
 152         self.con.putrequest('', '')
 153         self.con.endheaders()
 154         res = self.con.getresponse()
 155         self.assertEqual(res.status, 400)
 156
 157     def test_header_close(self):
 158         self.con.putrequest('GET', '/')
 159         self.con.putheader('Connection', 'close')
 160         self.con.endheaders()
 161         res = self.con.getresponse()
 162         self.assertEqual(res.status, 501)
 163
 164     def test_head_keep_alive(self):
 165         self.con._http_vsn_str = 'HTTP/1.1'
 166         self.con.putrequest('GET', '/')
 167         self.con.putheader('Connection', 'keep-alive')
 168         self.con.endheaders()
 169         res = self.con.getresponse()
 170         self.assertEqual(res.status, 501)
 171
 172     def test_handler(self):
 173         self.con.request('TEST', '/')
 174         res = self.con.getresponse()
 175         self.assertEqual(res.status, 204)
 176
 177     def test_return_header_keep_alive(self):
 178         self.con.request('KEEP', '/')
 179         res = self.con.getresponse()
 180         self.assertEqual(res.getheader('Connection'), 'keep-alive')
 181         self.con.request('TEST', '/')
 182
 183     def test_internal_key_error(self):
 184         self.con.request('KEYERROR', '/')
 185         res = self.con.getresponse()
 186         self.assertEqual(res.status, 999)
 187
 188     def test_return_custom_status(self):
 189         self.con.request('CUSTOM', '/')
 190         res = self.con.getresponse()
 191         self.assertEqual(res.status, 999)
 192
 193
 194 class SimpleHTTPServerTestCase(BaseTestCase):
 195     class request_handler(NoLogRequestHandler, SimpleHTTPRequestHandler):
 196         pass
 197
 198     def setUp(self):
 199         BaseTestCase.setUp(self)
 200         self.cwd = os.getcwd()
 201         basetempdir = tempfile.gettempdir()
 202         os.chdir(basetempdir)
 203         self.data = 'We are the knights who say Ni!'
 204         self.tempdir = tempfile.mkdtemp(dir=basetempdir)
 205         self.tempdir_name = os.path.basename(self.tempdir)
 206         temp = open(os.path.join(self.tempdir, 'test'), 'wb')
 207         temp.write(self.data)
 208         temp.close()
 209
 210     def tearDown(self):
 211         try:
 212             os.chdir(self.cwd)
 213             try:
 214                 shutil.rmtree(self.tempdir)
 215             except:
 216                 pass
 217         finally:
 218             BaseTestCase.tearDown(self)
 219
 220     def check_status_and_reason(self, response, status, data=None):
 221         body = response.read()
 222         self.assertTrue(response)
 223         self.assertEqual(response.status, status)
 224         self.assertIsNotNone(response.reason)
 225         if data:
 226             self.assertEqual(data, body)
 227
 228     def test_get(self):
 229         #constructs the path relative to the root directory of the HTTPServer
 230         response = self.request(self.tempdir_name + '/test')
 231         self.check_status_and_reason(response, 200, data=self.data)
 232         response = self.request(self.tempdir_name + '/')
 233         self.check_status_and_reason(response, 200)
 234         response = self.request(self.tempdir_name)
 235         self.check_status_and_reason(response, 301)
 236         response = self.request('/ThisDoesNotExist')
 237         self.check_status_and_reason(response, 404)
 238         response = self.request('/' + 'ThisDoesNotExist' + '/')
 239         self.check_status_and_reason(response, 404)
 240         f = open(os.path.join(self.tempdir_name, 'index.html'), 'w')
 241         response = self.request('/' + self.tempdir_name + '/')
 242         self.check_status_and_reason(response, 200)
 243         if os.name == 'posix':
 244             # chmod won't work as expected on Windows platforms
 245             os.chmod(self.tempdir, 0)
 246             response = self.request(self.tempdir_name + '/')
 247             self.check_status_and_reason(response, 404)
 248             os.chmod(self.tempdir, 0755)
 249
 250     def test_head(self):
 251         response = self.request(
 252             self.tempdir_name + '/test', method='HEAD')
 253         self.check_status_and_reason(response, 200)
 254         self.assertEqual(response.getheader('content-length'),
 255                          str(len(self.data)))
 256         self.assertEqual(response.getheader('content-type'),
 257                          'application/octet-stream')
 258
 259     def test_invalid_requests(self):
 260         response = self.request('/', method='FOO')
 261         self.check_status_and_reason(response, 501)
 262         # requests must be case sensitive,so this should fail too
 263         response = self.request('/', method='get')
 264         self.check_status_and_reason(response, 501)
 265         response = self.request('/', method='GETs')
 266         self.check_status_and_reason(response, 501)
 267
 268
 269 cgi_file1 = """\
 270 #!%s
 271
 272 print "Content-type: text/html"
 273 print
 274 print "Hello World"
 275 """
 276
 277 cgi_file2 = """\
 278 #!%s
 279 import cgi
 280
 281 print "Content-type: text/html"
 282 print
 283
 284 form = cgi.FieldStorage()
 285 print "%%s, %%s, %%s" %% (form.getfirst("spam"), form.getfirst("eggs"),
 286                           form.getfirst("bacon"))
 287 """
 288
 289 class CGIHTTPServerTestCase(BaseTestCase):
 290     class request_handler(NoLogRequestHandler, CGIHTTPRequestHandler):
 291         pass
 292
 293     def setUp(self):
 294         BaseTestCase.setUp(self)
 295         self.parent_dir = tempfile.mkdtemp()
 296         self.cgi_dir = os.path.join(self.parent_dir, 'cgi-bin')
 297         os.mkdir(self.cgi_dir)
 298
 299         # The shebang line should be pure ASCII: use symlink if possible.
 300         # See issue #7668.
 301         if hasattr(os, 'symlink'):
 302             self.pythonexe = os.path.join(self.parent_dir, 'python')
 303             os.symlink(sys.executable, self.pythonexe)
 304         else:
 305             self.pythonexe = sys.executable
 306
 307         self.file1_path = os.path.join(self.cgi_dir, 'file1.py')
 308         with open(self.file1_path, 'w') as file1:
 309             file1.write(cgi_file1 % self.pythonexe)
 310         os.chmod(self.file1_path, 0777)
 311
 312         self.file2_path = os.path.join(self.cgi_dir, 'file2.py')
 313         with open(self.file2_path, 'w') as file2:
 314             file2.write(cgi_file2 % self.pythonexe)
 315         os.chmod(self.file2_path, 0777)
 316
 317         self.cwd = os.getcwd()
 318         os.chdir(self.parent_dir)
 319
 320     def tearDown(self):
 321         try:
 322             os.chdir(self.cwd)
 323             if self.pythonexe != sys.executable:
 324                 os.remove(self.pythonexe)
 325             os.remove(self.file1_path)
 326             os.remove(self.file2_path)
 327             os.rmdir(self.cgi_dir)
 328             os.rmdir(self.parent_dir)
 329         finally:
 330             BaseTestCase.tearDown(self)
 331
 332     def test_url_collapse_path_split(self):
 333         test_vectors = {
 334             '': ('/', ''),
 335             '..': IndexError,
 336             '/.//..': IndexError,
 337             '/': ('/', ''),
 338             '//': ('/', ''),
 339             '/\\': ('/', '\\'),
 340             '/.//': ('/', ''),
 341             'cgi-bin/file1.py': ('/cgi-bin', 'file1.py'),
 342             '/cgi-bin/file1.py': ('/cgi-bin', 'file1.py'),
 343             'a': ('/', 'a'),
 344             '/a': ('/', 'a'),
 345             '//a': ('/', 'a'),
 346             './a': ('/', 'a'),
 347             './C:/': ('/C:', ''),
 348             '/a/b': ('/a', 'b'),
 349             '/a/b/': ('/a/b', ''),
 350             '/a/b/c/..': ('/a/b', ''),
 351             '/a/b/c/../d': ('/a/b', 'd'),
 352             '/a/b/c/../d/e/../f': ('/a/b/d', 'f'),
 353             '/a/b/c/../d/e/../../f': ('/a/b', 'f'),
 354             '/a/b/c/../d/e/.././././..//f': ('/a/b', 'f'),
 355             '../a/b/c/../d/e/.././././..//f': IndexError,
 356             '/a/b/c/../d/e/../../../f': ('/a', 'f'),
 357             '/a/b/c/../d/e/../../../../f': ('/', 'f'),
 358             '/a/b/c/../d/e/../../../../../f': IndexError,
 359             '/a/b/c/../d/e/../../../../f/..': ('/', ''),
 360         }
 361         for path, expected in test_vectors.iteritems():
 362             if isinstance(expected, type) and issubclass(expected, Exception):
 363                 self.assertRaises(expected,
 364                                   CGIHTTPServer._url_collapse_path_split, path)
 365             else:
 366                 actual = CGIHTTPServer._url_collapse_path_split(path)
 367                 self.assertEqual(expected, actual,
 368                                  msg='path = %r\nGot:    %r\nWanted: %r' %
 369                                  (path, actual, expected))
 370
 371     def test_headers_and_content(self):
 372         res = self.request('/cgi-bin/file1.py')
 373         self.assertEqual(('Hello World\n', 'text/html', 200),
 374             (res.read(), res.getheader('Content-type'), res.status))
 375
 376     def test_post(self):
 377         params = urllib.urlencode({'spam' : 1, 'eggs' : 'python', 'bacon' : 123456})
 378         headers = {'Content-type' : 'application/x-www-form-urlencoded'}
 379         res = self.request('/cgi-bin/file2.py', 'POST', params, headers)
 380
 381         self.assertEqual(res.read(), '1, python, 123456\n')
 382
 383     def test_invaliduri(self):
 384         res = self.request('/cgi-bin/invalid')
 385         res.read()
 386         self.assertEqual(res.status, 404)
 387
 388     def test_authorization(self):
 389         headers = {'Authorization' : 'Basic %s' %
 390                    base64.b64encode('username:pass')}
 391         res = self.request('/cgi-bin/file1.py', 'GET', headers=headers)
 392         self.assertEqual(('Hello World\n', 'text/html', 200),
 393                 (res.read(), res.getheader('Content-type'), res.status))
 394
 395     def test_no_leading_slash(self):
 396         # http://bugs.python.org/issue2254
 397         res = self.request('cgi-bin/file1.py')
 398         self.assertEqual(('Hello World\n', 'text/html', 200),
 399              (res.read(), res.getheader('Content-type'), res.status))
 400
 401
 402 def test_main(verbose=None):
 403     try:
 404         cwd = os.getcwd()
 405         test_support.run_unittest(BaseHTTPServerTestCase,
 406                                 SimpleHTTPServerTestCase,
 407                                 CGIHTTPServerTestCase
 408                                 )
 409     finally:
 410         os.chdir(cwd)
 411
 412 if __name__ == '__main__':
 413     test_main()