summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_httplib.py
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/test/test_httplib.py')
-rw-r--r--Lib/test/test_httplib.py94
1 files changed, 90 insertions, 4 deletions
diff --git a/Lib/test/test_httplib.py b/Lib/test/test_httplib.py
index ff03321..425c716 100644
--- a/Lib/test/test_httplib.py
+++ b/Lib/test/test_httplib.py
@@ -158,6 +158,23 @@ class BasicTest(TestCase):
self.assertEqual(resp.read(2), b'xt')
self.assertTrue(resp.isclosed())
+ def test_partial_readintos(self):
+ # if we have a lenght, 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"
+ sock = FakeSocket(body)
+ resp = client.HTTPResponse(sock)
+ resp.begin()
+ b = bytearray(2)
+ n = resp.readinto(b)
+ self.assertEqual(n, 2)
+ self.assertEqual(bytes(b), b'Te')
+ self.assertFalse(resp.isclosed())
+ n = resp.readinto(b)
+ self.assertEqual(n, 2)
+ self.assertEqual(bytes(b), b'xt')
+ self.assertTrue(resp.isclosed())
+
def test_host_port(self):
# Check invalid host_port
@@ -206,6 +223,21 @@ class BasicTest(TestCase):
if resp.read():
self.fail("Did not expect response from HEAD request")
+ def test_readinto_head(self):
+ # Test that the library doesn't attempt to read any data
+ # from a HEAD request. (Tickles SF bug #622042.)
+ sock = FakeSocket(
+ 'HTTP/1.1 200 OK\r\n'
+ 'Content-Length: 14432\r\n'
+ '\r\n',
+ NoEOFStringIO)
+ resp = client.HTTPResponse(sock, method="HEAD")
+ resp.begin()
+ b = bytearray(5)
+ if resp.readinto(b) != 0:
+ self.fail("Did not expect response from HEAD request")
+ self.assertEqual(bytes(b), b'\x00'*5)
+
def test_send_file(self):
expected = (b'GET /foo HTTP/1.1\r\nHost: example.com\r\n'
b'Accept-Encoding: identity\r\nContent-Length:')
@@ -285,6 +317,40 @@ class BasicTest(TestCase):
finally:
resp.close()
+ def test_readinto_chunked(self):
+ chunked_start = (
+ 'HTTP/1.1 200 OK\r\n'
+ 'Transfer-Encoding: chunked\r\n\r\n'
+ 'a\r\n'
+ 'hello worl\r\n'
+ '1\r\n'
+ 'd\r\n'
+ )
+ sock = FakeSocket(chunked_start + '0\r\n')
+ resp = client.HTTPResponse(sock, method="GET")
+ resp.begin()
+ b = bytearray(16)
+ n = resp.readinto(b)
+ self.assertEqual(b[:11], b'hello world')
+ self.assertEqual(n, 11)
+ resp.close()
+
+ for x in ('', 'foo\r\n'):
+ sock = FakeSocket(chunked_start + x)
+ resp = client.HTTPResponse(sock, method="GET")
+ resp.begin()
+ try:
+ b = bytearray(16)
+ n = resp.readinto(b)
+ except client.IncompleteRead as i:
+ self.assertEqual(i.partial, b'hello world')
+ self.assertEqual(repr(i),'IncompleteRead(11 bytes read)')
+ self.assertEqual(str(i),'IncompleteRead(11 bytes read)')
+ else:
+ self.fail('IncompleteRead expected')
+ finally:
+ resp.close()
+
def test_chunked_head(self):
chunked_start = (
'HTTP/1.1 200 OK\r\n'
@@ -302,6 +368,26 @@ class BasicTest(TestCase):
self.assertEqual(resp.reason, 'OK')
self.assertTrue(resp.isclosed())
+ def test_readinto_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 = client.HTTPResponse(sock, method="HEAD")
+ resp.begin()
+ b = bytearray(5)
+ n = resp.readinto(b)
+ self.assertEqual(n, 0)
+ self.assertEqual(bytes(b), b'\x00'*5)
+ 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\nContent-Length: -1\r\n\r\nHello\r\n')
@@ -508,8 +594,7 @@ class HTTPSTest(TestCase):
def test_local_good_hostname(self):
# The (valid) cert validates the HTTP hostname
import ssl
- from test.ssl_servers import make_https_server
- server = make_https_server(self, CERT_localhost)
+ server = self.make_server(CERT_localhost)
context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
context.verify_mode = ssl.CERT_REQUIRED
context.load_verify_locations(CERT_localhost)
@@ -517,12 +602,12 @@ class HTTPSTest(TestCase):
h.request('GET', '/nonexistent')
resp = h.getresponse()
self.assertEqual(resp.status, 404)
+ del server
def test_local_bad_hostname(self):
# The (valid) cert doesn't validate the HTTP hostname
import ssl
- from test.ssl_servers import make_https_server
- server = make_https_server(self, CERT_fakehostname)
+ server = self.make_server(CERT_fakehostname)
context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
context.verify_mode = ssl.CERT_REQUIRED
context.load_verify_locations(CERT_fakehostname)
@@ -540,6 +625,7 @@ class HTTPSTest(TestCase):
h.request('GET', '/nonexistent')
resp = h.getresponse()
self.assertEqual(resp.status, 404)
+ del server
@unittest.skipIf(not hasattr(client, 'HTTPSConnection'),
'http.client.HTTPSConnection not available')