Lib/test/test_httplib.py

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