Changeset 391 for python/trunk/Lib/test/test_httplib.py

Timestamp:

Mar 19, 2014, 11:31:01 PM (11 years ago)

Author:

dmik

Message:

python: Merge vendor 2.7.6 to trunk.

Location:

python/trunk

Files:

• . (modified) (1 prop)
• Lib/test/test_httplib.py (modified) (16 diffs)

python/trunk/Lib/test/test_httplib.py

@@ +1 @@
+import httplib
 import array
 import httplib
 import StringIO
 import socket
-
-from unittest import TestCase
+import errno
+
+import unittest
+TestCase = unittest.TestCase
 
 from test import test_support
@@ -23 +26 @@
             raise httplib.UnimplementedFileMode()
         return self.fileclass(self.text)
+
+class EPipeSocket(FakeSocket):
+
+    def __init__(self, text, pipe_trigger):
+        # When sendall() is called with pipe_trigger, raise EPIPE.
+        FakeSocket.__init__(self, text)
+        self.pipe_trigger = pipe_trigger
+
+    def sendall(self, data):
+        if self.pipe_trigger in data:
+            raise socket.error(errno.EPIPE, "gotcha")
+        self.data += data
+
+    def close(self):
+        pass
 
 class NoEOFStringIO(StringIO.StringIO):
@@ -47 +65 @@
         # Some headers are added automatically, but should not be added by
         # .request() if they are explicitly set.
-
-        import httplib
 
         class HeaderCountingBuffer(list):
@@ -75 +91 @@
                 self.assertEqual(conn._buffer.count[header.lower()], 1)
 
+    def test_content_length_0(self):
+
+        class ContentLengthChecker(list):
+            def __init__(self):
+                list.__init__(self)
+                self.content_length = None
+            def append(self, item):
+                kv = item.split(':', 1)
+                if len(kv) > 1 and kv[0].lower() == 'content-length':
+                    self.content_length = kv[1].strip()
+                list.append(self, item)
+
+        # POST with empty body
+        conn = httplib.HTTPConnection('example.com')
+        conn.sock = FakeSocket(None)
+        conn._buffer = ContentLengthChecker()
+        conn.request('POST', '/', '')
+        self.assertEqual(conn._buffer.content_length, '0',
+                        'Header Content-Length not set')
+
+        # PUT request with empty body
+        conn = httplib.HTTPConnection('example.com')
+        conn.sock = FakeSocket(None)
+        conn._buffer = ContentLengthChecker()
+        conn.request('PUT', '/', '')
+        self.assertEqual(conn._buffer.content_length, '0',
+                        'Header Content-Length not set')
+
+    def test_putheader(self):
+        conn = httplib.HTTPConnection('example.com')
+        conn.sock = FakeSocket(None)
+        conn.putrequest('GET','/')
+        conn.putheader('Content-length',42)
+        self.assertTrue('Content-length: 42' in conn._buffer)
+
+    def test_ipv6host_header(self):
+        # Default host header on IPv6 transaction should wrapped by [] if
+        # its actual IPv6 address
+        expected = 'GET /foo HTTP/1.1\r\nHost: [2001::]:81\r\n' \
+                   'Accept-Encoding: identity\r\n\r\n'
+        conn = httplib.HTTPConnection('[2001::]:81')
+        sock = FakeSocket('')
+        conn.sock = sock
+        conn.request('GET', '/foo')
+        self.assertTrue(sock.data.startswith(expected))
+
+        expected = 'GET /foo HTTP/1.1\r\nHost: [2001:102A::]\r\n' \
+                   'Accept-Encoding: identity\r\n\r\n'
+        conn = httplib.HTTPConnection('[2001:102A::]')
+        sock = FakeSocket('')
+        conn.sock = sock
+        conn.request('GET', '/foo')
+        self.assertTrue(sock.data.startswith(expected))
+
+
 class BasicTest(TestCase):
     def test_status_lines(self):
@@ -91 +162 @@
         self.assertRaises(httplib.BadStatusLine, resp.begin)
 
+    def test_bad_status_repr(self):
+        exc = httplib.BadStatusLine('')
+        self.assertEqual(repr(exc), '''BadStatusLine("\'\'",)''')
+
     def test_partial_reads(self):
-        # if we have a lenght, the system knows when to close itself
+        # if we have a length, the system knows when to close itself
         # same behaviour than when we read the whole thing with read()
         body = "HTTP/1.1 200 Ok\r\nContent-Length: 4\r\n\r\nText"
@@ -103 +178 @@
         self.assertTrue(resp.isclosed())
 
+    def test_partial_reads_no_content_length(self):
+        # when no length is present, the socket should be gracefully closed when
+        # all data was read
+        body = "HTTP/1.1 200 Ok\r\n\r\nText"
+        sock = FakeSocket(body)
+        resp = httplib.HTTPResponse(sock)
+        resp.begin()
+        self.assertEqual(resp.read(2), 'Te')
+        self.assertFalse(resp.isclosed())
+        self.assertEqual(resp.read(2), 'xt')
+        self.assertEqual(resp.read(1), '')
+        self.assertTrue(resp.isclosed())
+
+    def test_partial_reads_incomplete_body(self):
+        # if the server shuts down the connection before the whole
+        # content-length is delivered, the socket is gracefully closed
+        body = "HTTP/1.1 200 Ok\r\nContent-Length: 10\r\n\r\nText"
+        sock = FakeSocket(body)
+        resp = httplib.HTTPResponse(sock)
+        resp.begin()
+        self.assertEqual(resp.read(2), 'Te')
+        self.assertFalse(resp.isclosed())
+        self.assertEqual(resp.read(2), 'xt')
+        self.assertEqual(resp.read(1), '')
+        self.assertTrue(resp.isclosed())
+
     def test_host_port(self):
         # Check invalid host_port
 
-        for hp in ("www.python.org:abc", "www.python.org:"):
+        # Note that httplib does not accept user:password@ in the host-port.
+        for hp in ("www.python.org:abc", "user:password@www.python.org"):
             self.assertRaises(httplib.InvalidURL, httplib.HTTP, hp)
 
@@ -113 +215 @@
                          ("www.python.org:80", "www.python.org", 80),
                          ("www.python.org", "www.python.org", 80),
+                         ("www.python.org:", "www.python.org", 80),
                          ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 80)):
             http = httplib.HTTP(hp)
@@ -170 +273 @@
         conn.sock = sock
         conn.send(expected)
-        self.assertEquals(expected, sock.data)
+        self.assertEqual(expected, sock.data)
         sock.data = ''
         conn.send(array.array('c', expected))
-        self.assertEquals(expected, sock.data)
+        self.assertEqual(expected, sock.data)
         sock.data = ''
         conn.send(StringIO.StringIO(expected))
-        self.assertEquals(expected, sock.data)
+        self.assertEqual(expected, sock.data)
 
     def test_chunked(self):
@@ -190 +293 @@
         resp = httplib.HTTPResponse(sock, method="GET")
         resp.begin()
-        self.assertEquals(resp.read(), 'hello world')
+        self.assertEqual(resp.read(), 'hello world')
         resp.close()
 
@@ -200 +303 @@
                 resp.read()
             except httplib.IncompleteRead, i:
-                self.assertEquals(i.partial, 'hello world')
+                self.assertEqual(i.partial, 'hello world')
                 self.assertEqual(repr(i),'IncompleteRead(11 bytes read)')
                 self.assertEqual(str(i),'IncompleteRead(11 bytes read)')
@@ -208 +311 @@
                 resp.close()
 
+    def test_chunked_head(self):
+        chunked_start = (
+            'HTTP/1.1 200 OK\r\n'
+            'Transfer-Encoding: chunked\r\n\r\n'
+            'a\r\n'
+            'hello world\r\n'
+            '1\r\n'
+            'd\r\n'
+        )
+        sock = FakeSocket(chunked_start + '0\r\n')
+        resp = httplib.HTTPResponse(sock, method="HEAD")
+        resp.begin()
+        self.assertEqual(resp.read(), '')
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(resp.reason, 'OK')
+        self.assertTrue(resp.isclosed())
+
     def test_negative_content_length(self):
         sock = FakeSocket('HTTP/1.1 200 OK\r\n'
@@ -213 +333 @@
         resp = httplib.HTTPResponse(sock, method="GET")
         resp.begin()
-        self.assertEquals(resp.read(), 'Hello\r\n')
-        resp.close()
+        self.assertEqual(resp.read(), 'Hello\r\n')
+        self.assertTrue(resp.isclosed())
 
     def test_incomplete_read(self):
@@ -223 +343 @@
             resp.read()
         except httplib.IncompleteRead as i:
-            self.assertEquals(i.partial, 'Hello\r\n')
+            self.assertEqual(i.partial, 'Hello\r\n')
             self.assertEqual(repr(i),
                              "IncompleteRead(7 bytes read, 3 more expected)")
             self.assertEqual(str(i),
                              "IncompleteRead(7 bytes read, 3 more expected)")
+            self.assertTrue(resp.isclosed())
         else:
             self.fail('IncompleteRead expected')
-        finally:
-            resp.close()
-
+
+    def test_epipe(self):
+        sock = EPipeSocket(
+            "HTTP/1.0 401 Authorization Required\r\n"
+            "Content-type: text/html\r\n"
+            "WWW-Authenticate: Basic realm=\"example\"\r\n",
+            b"Content-Length")
+        conn = httplib.HTTPConnection("example.com")
+        conn.sock = sock
+        self.assertRaises(socket.error,
+                          lambda: conn.request("PUT", "/url", "body"))
+        resp = conn.getresponse()
+        self.assertEqual(401, resp.status)
+        self.assertEqual("Basic realm=\"example\"",
+                         resp.getheader("www-authenticate"))
+
+    def test_filenoattr(self):
+        # Just test the fileno attribute in the HTTPResponse Object.
+        body = "HTTP/1.1 200 Ok\r\n\r\nText"
+        sock = FakeSocket(body)
+        resp = httplib.HTTPResponse(sock)
+        self.assertTrue(hasattr(resp,'fileno'),
+                'HTTPResponse should expose a fileno attribute')
+
+    # Test lines overflowing the max line size (_MAXLINE in http.client)
+
+    def test_overflowing_status_line(self):
+        self.skipTest("disabled for HTTP 0.9 support")
+        body = "HTTP/1.1 200 Ok" + "k" * 65536 + "\r\n"
+        resp = httplib.HTTPResponse(FakeSocket(body))
+        self.assertRaises((httplib.LineTooLong, httplib.BadStatusLine), resp.begin)
+
+    def test_overflowing_header_line(self):
+        body = (
+            'HTTP/1.1 200 OK\r\n'
+            'X-Foo: bar' + 'r' * 65536 + '\r\n\r\n'
+        )
+        resp = httplib.HTTPResponse(FakeSocket(body))
+        self.assertRaises(httplib.LineTooLong, resp.begin)
+
+    def test_overflowing_chunked_line(self):
+        body = (
+            'HTTP/1.1 200 OK\r\n'
+            'Transfer-Encoding: chunked\r\n\r\n'
+            + '0' * 65536 + 'a\r\n'
+            'hello world\r\n'
+            '0\r\n'
+        )
+        resp = httplib.HTTPResponse(FakeSocket(body))
+        resp.begin()
+        self.assertRaises(httplib.LineTooLong, resp.read)
+
+    def test_early_eof(self):
+        # Test httpresponse with no \r\n termination,
+        body = "HTTP/1.1 200 Ok"
+        sock = FakeSocket(body)
+        resp = httplib.HTTPResponse(sock)
+        resp.begin()
+        self.assertEqual(resp.read(), '')
+        self.assertTrue(resp.isclosed())
 
 class OfflineTest(TestCase):
     def test_responses(self):
-        self.assertEquals(httplib.responses[httplib.NOT_FOUND], "Not Found")
+        self.assertEqual(httplib.responses[httplib.NOT_FOUND], "Not Found")
+
+
+class SourceAddressTest(TestCase):
+    def setUp(self):
+        self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.port = test_support.bind_port(self.serv)
+        self.source_port = test_support.find_unused_port()
+        self.serv.listen(5)
+        self.conn = None
+
+    def tearDown(self):
+        if self.conn:
+            self.conn.close()
+            self.conn = None
+        self.serv.close()
+        self.serv = None
+
+    def testHTTPConnectionSourceAddress(self):
+        self.conn = httplib.HTTPConnection(HOST, self.port,
+                source_address=('', self.source_port))
+        self.conn.connect()
+        self.assertEqual(self.conn.sock.getsockname()[1], self.source_port)
+
+    @unittest.skipIf(not hasattr(httplib, 'HTTPSConnection'),
+                     'httplib.HTTPSConnection not defined')
+    def testHTTPSConnectionSourceAddress(self):
+        self.conn = httplib.HTTPSConnection(HOST, self.port,
+                source_address=('', self.source_port))
+        # We don't test anything here other the constructor not barfing as
+        # this code doesn't deal with setting up an active running SSL server
+        # for an ssl_wrapped connect() to actually return from.
+
 
 class TimeoutTest(TestCase):
@@ -255 +465 @@
         '''
         # default -- use global socket timeout
-        self.assert_(socket.getdefaulttimeout() is None)
+        self.assertTrue(socket.getdefaulttimeout() is None)
         socket.setdefaulttimeout(30)
         try:
@@ -266 +476 @@
 
         # no timeout -- do not use global socket default
-        self.assert_(socket.getdefaulttimeout() is None)
+        self.assertTrue(socket.getdefaulttimeout() is None)
         socket.setdefaulttimeout(30)
         try:
@@ -293 +503 @@
             self.assertEqual(h.timeout, 30)
 
+    @unittest.skipIf(not hasattr(httplib, 'HTTPS'), 'httplib.HTTPS not available')
+    def test_host_port(self):
+        # Check invalid host_port
+
+        # Note that httplib does not accept user:password@ in the host-port.
+        for hp in ("www.python.org:abc", "user:password@www.python.org"):
+            self.assertRaises(httplib.InvalidURL, httplib.HTTP, hp)
+
+        for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000", "fe80::207:e9ff:fe9b",
+                          8000),
+                         ("pypi.python.org:443", "pypi.python.org", 443),
+                         ("pypi.python.org", "pypi.python.org", 443),
+                         ("pypi.python.org:", "pypi.python.org", 443),
+                         ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 443)):
+            http = httplib.HTTPS(hp)
+            c = http._conn
+            if h != c.host:
+                self.fail("Host incorrectly parsed: %s != %s" % (h, c.host))
+            if p != c.port:
+                self.fail("Port incorrectly parsed: %s != %s" % (p, c.host))
+
+
 def test_main(verbose=None):
     test_support.run_unittest(HeaderTests, OfflineTest, BasicTest, TimeoutTest,
-                              HTTPSTimeoutTest)
+                              HTTPSTimeoutTest, SourceAddressTest)
 
 if __name__ == '__main__':