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