Lib/test/test_httplib.py

   1 import array
   2 import httplib
   3 import StringIO
   4 import socket
   5
   6 import unittest
   7 TestCase = unittest.TestCase
   8
   9 from test import test_support
  10
  11 HOST = test_support.HOST
  12
  13 class FakeSocket:
  14     def __init__(self, text, fileclass=StringIO.StringIO):
  15         self.text = text
  16         self.fileclass = fileclass
  17         self.data = ''
  18
  19     def sendall(self, data):
  20         self.data += ''.join(data)
  21
  22     def makefile(self, mode, bufsize=None):
  23         if mode != 'r' and mode != 'rb':
  24             raise httplib.UnimplementedFileMode()
  25         return self.fileclass(self.text)
  26
  27 class NoEOFStringIO(StringIO.StringIO):
  28     """Like StringIO, but raises AssertionError on EOF.
  29
  30     This is used below to test that httplib doesn't try to read
  31     more from the underlying file than it should.
  32     """
  33     def read(self, n=-1):
  34         data = StringIO.StringIO.read(self, n)
  35         if data == '':
  36             raise AssertionError('caller tried to read past EOF')
  37         return data
  38
  39     def readline(self, length=None):
  40         data = StringIO.StringIO.readline(self, length)
  41         if data == '':
  42             raise AssertionError('caller tried to read past EOF')
  43         return data
  44
  45
  46 class HeaderTests(TestCase):
  47     def test_auto_headers(self):
  48         # Some headers are added automatically, but should not be added by
  49         # .request() if they are explicitly set.
  50
  51         import httplib
  52
  53         class HeaderCountingBuffer(list):
  54             def __init__(self):
  55                 self.count = {}
  56             def append(self, item):
  57                 kv = item.split(':')
  58                 if len(kv) > 1:
  59                     # item is a 'Key: Value' header string
  60                     lcKey = kv[0].lower()
  61                     self.count.setdefault(lcKey, 0)
  62                     self.count[lcKey] += 1
  63                 list.append(self, item)
  64
  65         for explicit_header in True, False:
  66             for header in 'Content-length', 'Host', 'Accept-encoding':
  67                 conn = httplib.HTTPConnection('example.com')
  68                 conn.sock = FakeSocket('blahblahblah')
  69                 conn._buffer = HeaderCountingBuffer()
  70
  71                 body = 'spamspamspam'
  72                 headers = {}
  73                 if explicit_header:
  74                     headers[header] = str(len(body))
  75                 conn.request('POST', '/', body, headers)
  76                 self.assertEqual(conn._buffer.count[header.lower()], 1)
  77
  78 class BasicTest(TestCase):
  79     def test_status_lines(self):
  80         # Test HTTP status lines
  81
  82         body = "HTTP/1.1 200 Ok\r\n\r\nText"
  83         sock = FakeSocket(body)
  84         resp = httplib.HTTPResponse(sock)
  85         resp.begin()
  86         self.assertEqual(resp.read(), 'Text')
  87         self.assertTrue(resp.isclosed())
  88
  89         body = "HTTP/1.1 400.100 Not Ok\r\n\r\nText"
  90         sock = FakeSocket(body)
  91         resp = httplib.HTTPResponse(sock)
  92         self.assertRaises(httplib.BadStatusLine, resp.begin)
  93
  94     def test_bad_status_repr(self):
  95         exc = httplib.BadStatusLine('')
  96         self.assertEquals(repr(exc), '''BadStatusLine("\'\'",)''')
  97
  98     def test_partial_reads(self):
  99         # if we have a lenght, the system knows when to close itself
 100         # same behaviour than when we read the whole thing with read()
 101         body = "HTTP/1.1 200 Ok\r\nContent-Length: 4\r\n\r\nText"
 102         sock = FakeSocket(body)
 103         resp = httplib.HTTPResponse(sock)
 104         resp.begin()
 105         self.assertEqual(resp.read(2), 'Te')
 106         self.assertFalse(resp.isclosed())
 107         self.assertEqual(resp.read(2), 'xt')
 108         self.assertTrue(resp.isclosed())
 109
 110     def test_host_port(self):
 111         # Check invalid host_port
 112
 113         for hp in ("www.python.org:abc", "www.python.org:"):
 114             self.assertRaises(httplib.InvalidURL, httplib.HTTP, hp)
 115
 116         for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000", "fe80::207:e9ff:fe9b",
 117                           8000),
 118                          ("www.python.org:80", "www.python.org", 80),
 119                          ("www.python.org", "www.python.org", 80),
 120                          ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 80)):
 121             http = httplib.HTTP(hp)
 122             c = http._conn
 123             if h != c.host:
 124                 self.fail("Host incorrectly parsed: %s != %s" % (h, c.host))
 125             if p != c.port:
 126                 self.fail("Port incorrectly parsed: %s != %s" % (p, c.host))
 127
 128     def test_response_headers(self):
 129         # test response with multiple message headers with the same field name.
 130         text = ('HTTP/1.1 200 OK\r\n'
 131                 'Set-Cookie: Customer="WILE_E_COYOTE";'
 132                 ' Version="1"; Path="/acme"\r\n'
 133                 'Set-Cookie: Part_Number="Rocket_Launcher_0001"; Version="1";'
 134                 ' Path="/acme"\r\n'
 135                 '\r\n'
 136                 'No body\r\n')
 137         hdr = ('Customer="WILE_E_COYOTE"; Version="1"; Path="/acme"'
 138                ', '
 139                'Part_Number="Rocket_Launcher_0001"; Version="1"; Path="/acme"')
 140         s = FakeSocket(text)
 141         r = httplib.HTTPResponse(s)
 142         r.begin()
 143         cookies = r.getheader("Set-Cookie")
 144         if cookies != hdr:
 145             self.fail("multiple headers not combined properly")
 146
 147     def test_read_head(self):
 148         # Test that the library doesn't attempt to read any data
 149         # from a HEAD request.  (Tickles SF bug #622042.)
 150         sock = FakeSocket(
 151             'HTTP/1.1 200 OK\r\n'
 152             'Content-Length: 14432\r\n'
 153             '\r\n',
 154             NoEOFStringIO)
 155         resp = httplib.HTTPResponse(sock, method="HEAD")
 156         resp.begin()
 157         if resp.read() != "":
 158             self.fail("Did not expect response from HEAD request")
 159
 160     def test_send_file(self):
 161         expected = 'GET /foo HTTP/1.1\r\nHost: example.com\r\n' \
 162                    'Accept-Encoding: identity\r\nContent-Length:'
 163
 164         body = open(__file__, 'rb')
 165         conn = httplib.HTTPConnection('example.com')
 166         sock = FakeSocket(body)
 167         conn.sock = sock
 168         conn.request('GET', '/foo', body)
 169         self.assertTrue(sock.data.startswith(expected))
 170
 171     def test_send(self):
 172         expected = 'this is a test this is only a test'
 173         conn = httplib.HTTPConnection('example.com')
 174         sock = FakeSocket(None)
 175         conn.sock = sock
 176         conn.send(expected)
 177         self.assertEquals(expected, sock.data)
 178         sock.data = ''
 179         conn.send(array.array('c', expected))
 180         self.assertEquals(expected, sock.data)
 181         sock.data = ''
 182         conn.send(StringIO.StringIO(expected))
 183         self.assertEquals(expected, sock.data)
 184
 185     def test_chunked(self):
 186         chunked_start = (
 187             'HTTP/1.1 200 OK\r\n'
 188             'Transfer-Encoding: chunked\r\n\r\n'
 189             'a\r\n'
 190             'hello worl\r\n'
 191             '1\r\n'
 192             'd\r\n'
 193         )
 194         sock = FakeSocket(chunked_start + '0\r\n')
 195         resp = httplib.HTTPResponse(sock, method="GET")
 196         resp.begin()
 197         self.assertEquals(resp.read(), 'hello world')
 198         resp.close()
 199
 200         for x in ('', 'foo\r\n'):
 201             sock = FakeSocket(chunked_start + x)
 202             resp = httplib.HTTPResponse(sock, method="GET")
 203             resp.begin()
 204             try:
 205                 resp.read()
 206             except httplib.IncompleteRead, i:
 207                 self.assertEquals(i.partial, 'hello world')
 208                 self.assertEqual(repr(i),'IncompleteRead(11 bytes read)')
 209                 self.assertEqual(str(i),'IncompleteRead(11 bytes read)')
 210             else:
 211                 self.fail('IncompleteRead expected')
 212             finally:
 213                 resp.close()
 214
 215     def test_chunked_head(self):
 216         chunked_start = (
 217             'HTTP/1.1 200 OK\r\n'
 218             'Transfer-Encoding: chunked\r\n\r\n'
 219             'a\r\n'
 220             'hello world\r\n'
 221             '1\r\n'
 222             'd\r\n'
 223         )
 224         sock = FakeSocket(chunked_start + '0\r\n')
 225         resp = httplib.HTTPResponse(sock, method="HEAD")
 226         resp.begin()
 227         self.assertEquals(resp.read(), '')
 228         self.assertEquals(resp.status, 200)
 229         self.assertEquals(resp.reason, 'OK')
 230         self.assertTrue(resp.isclosed())
 231
 232     def test_negative_content_length(self):
 233         sock = FakeSocket('HTTP/1.1 200 OK\r\n'
 234                           'Content-Length: -1\r\n\r\nHello\r\n')
 235         resp = httplib.HTTPResponse(sock, method="GET")
 236         resp.begin()
 237         self.assertEquals(resp.read(), 'Hello\r\n')
 238         resp.close()
 239
 240     def test_incomplete_read(self):
 241         sock = FakeSocket('HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\nHello\r\n')
 242         resp = httplib.HTTPResponse(sock, method="GET")
 243         resp.begin()
 244         try:
 245             resp.read()
 246         except httplib.IncompleteRead as i:
 247             self.assertEquals(i.partial, 'Hello\r\n')
 248             self.assertEqual(repr(i),
 249                              "IncompleteRead(7 bytes read, 3 more expected)")
 250             self.assertEqual(str(i),
 251                              "IncompleteRead(7 bytes read, 3 more expected)")
 252         else:
 253             self.fail('IncompleteRead expected')
 254         finally:
 255             resp.close()
 256
 257
 258 class OfflineTest(TestCase):
 259     def test_responses(self):
 260         self.assertEquals(httplib.responses[httplib.NOT_FOUND], "Not Found")
 261
 262
 263 class SourceAddressTest(TestCase):
 264     def setUp(self):
 265         self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 266         self.port = test_support.bind_port(self.serv)
 267         self.source_port = test_support.find_unused_port()
 268         self.serv.listen(5)
 269         self.conn = None
 270
 271     def tearDown(self):
 272         if self.conn:
 273             self.conn.close()
 274             self.conn = None
 275         self.serv.close()
 276         self.serv = None
 277
 278     def testHTTPConnectionSourceAddress(self):
 279         self.conn = httplib.HTTPConnection(HOST, self.port,
 280                 source_address=('', self.source_port))
 281         self.conn.connect()
 282         self.assertEqual(self.conn.sock.getsockname()[1], self.source_port)
 283
 284     @unittest.skipIf(not hasattr(httplib, 'HTTPSConnection'),
 285                      'httplib.HTTPSConnection not defined')
 286     def testHTTPSConnectionSourceAddress(self):
 287         self.conn = httplib.HTTPSConnection(HOST, self.port,
 288                 source_address=('', self.source_port))
 289         # We don't test anything here other the constructor not barfing as
 290         # this code doesn't deal with setting up an active running SSL server
 291         # for an ssl_wrapped connect() to actually return from.
 292
 293
 294 class TimeoutTest(TestCase):
 295     PORT = None
 296
 297     def setUp(self):
 298         self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 299         TimeoutTest.PORT = test_support.bind_port(self.serv)
 300         self.serv.listen(5)
 301
 302     def tearDown(self):
 303         self.serv.close()
 304         self.serv = None
 305
 306     def testTimeoutAttribute(self):
 307         '''This will prove that the timeout gets through
 308         HTTPConnection and into the socket.
 309         '''
 310         # default -- use global socket timeout
 311         self.assertTrue(socket.getdefaulttimeout() is None)
 312         socket.setdefaulttimeout(30)
 313         try:
 314             httpConn = httplib.HTTPConnection(HOST, TimeoutTest.PORT)
 315             httpConn.connect()
 316         finally:
 317             socket.setdefaulttimeout(None)
 318         self.assertEqual(httpConn.sock.gettimeout(), 30)
 319         httpConn.close()
 320
 321         # no timeout -- do not use global socket default
 322         self.assertTrue(socket.getdefaulttimeout() is None)
 323         socket.setdefaulttimeout(30)
 324         try:
 325             httpConn = httplib.HTTPConnection(HOST, TimeoutTest.PORT,
 326                                               timeout=None)
 327             httpConn.connect()
 328         finally:
 329             socket.setdefaulttimeout(None)
 330         self.assertEqual(httpConn.sock.gettimeout(), None)
 331         httpConn.close()
 332
 333         # a value
 334         httpConn = httplib.HTTPConnection(HOST, TimeoutTest.PORT, timeout=30)
 335         httpConn.connect()
 336         self.assertEqual(httpConn.sock.gettimeout(), 30)
 337         httpConn.close()
 338
 339
 340 class HTTPSTimeoutTest(TestCase):
 341 # XXX Here should be tests for HTTPS, there isn't any right now!
 342
 343     def test_attributes(self):
 344         # simple test to check it's storing it
 345         if hasattr(httplib, 'HTTPSConnection'):
 346             h = httplib.HTTPSConnection(HOST, TimeoutTest.PORT, timeout=30)
 347             self.assertEqual(h.timeout, 30)
 348
 349 def test_main(verbose=None):
 350     test_support.run_unittest(HeaderTests, OfflineTest, BasicTest, TimeoutTest,
 351                               HTTPSTimeoutTest, SourceAddressTest)
 352
 353 if __name__ == '__main__':
 354     test_main()