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