Lib/test/test_httplib.py

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