mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-25 04:34:37 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 11

Issue #16298: In HTTPResponse.read(), close the socket when there is no Content-Length and the incoming stream is finished.

Browse source 
Patch by Eran Rundstein.
This commit is contained in:
Antoine Pitrou 2012-12-15 19:22:30 +01:00
commit d20e7745ee
4 changed files with 42 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

36
Lib/test/test_httplib.py
View file

@ -175,7 +175,7 @@ class BasicTest(TestCase):
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"
sock = FakeSocket(body)
@ -187,7 +187,7 @@ class BasicTest(TestCase):
self.assertTrue(resp.isclosed())
def test_partial_readintos(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"
sock = FakeSocket(body)
@ -203,6 +203,38 @@ class BasicTest(TestCase):
self.assertEqual(bytes(b), b'xt')
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 = client.HTTPResponse(sock)
resp.begin()
self.assertEqual(resp.read(2), b'Te')
self.assertFalse(resp.isclosed())
self.assertEqual(resp.read(2), b'xt')
self.assertEqual(resp.read(1), b'')
self.assertTrue(resp.isclosed())
def test_partial_readintos_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 = 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')
n = resp.readinto(b)
self.assertEqual(n, 0)
self.assertTrue(resp.isclosed())
def test_host_port(self):
# Check invalid host_port