37
37
InvalidHttpResponse.
41
import http.client as http_client
42
parse_headers = http_client.parse_headers
43
except ImportError: # python < 3
44
import httplib as http_client
45
parse_headers = http_client.HTTPMessage
40
from io import BytesIO
42
import http.client as http_client
44
parse_headers = http_client.parse_headers
51
from ..sixish import (
54
50
from ..transport.http import (
58
54
from .file_utils import (
70
66
return self.readfile
73
class FakeHTTPConnection(_urllib2_wrappers.HTTPConnection):
69
class FakeHTTPConnection(HTTPConnection):
75
71
def __init__(self, sock):
76
_urllib2_wrappers.HTTPConnection.__init__(self, 'localhost')
72
HTTPConnection.__init__(self, 'localhost')
77
73
# Set the socket to bypass the connection
92
88
f = response.ResponseFile('many', BytesIO(b'0\n1\nboo!\n'))
93
89
self.assertEqual([b'0\n', b'1\n', b'boo!\n'], list(f))
91
def test_readlines(self):
92
f = response.ResponseFile('many', BytesIO(b'0\n1\nboo!\n'))
93
self.assertEqual([b'0\n', b'1\n', b'boo!\n'], f.readlines())
96
96
class TestHTTPConnection(tests.TestCase):
722
722
# Get rid of the status line
723
723
status_and_headers.readline()
724
724
msg = parse_headers(status_and_headers)
727
727
def get_response(self, a_response):
728
728
"""Process a supplied response, and return the result."""
729
729
code, raw_headers, body = a_response
730
msg = self._build_HTTPMessage(raw_headers)
731
return response.handle_response('http://foo', code, msg,
732
BytesIO(a_response[2]))
730
getheader = self._build_HTTPMessage(raw_headers)
731
return response.handle_response(
732
'http://foo', code, getheader, BytesIO(a_response[2]))
734
734
def test_full_text(self):
735
735
out = self.get_response(_full_text_response)
780
780
def test_full_text_no_content_type(self):
781
781
# We should not require Content-Type for a full response
782
782
code, raw_headers, body = _full_text_response_no_content_type
783
msg = self._build_HTTPMessage(raw_headers)
784
out = response.handle_response('http://foo', code, msg, BytesIO(body))
783
getheader = self._build_HTTPMessage(raw_headers)
784
out = response.handle_response(
785
'http://foo', code, getheader, BytesIO(body))
785
786
self.assertEqual(body, out.read())
787
788
def test_full_text_no_content_length(self):
788
789
code, raw_headers, body = _full_text_response_no_content_length
789
msg = self._build_HTTPMessage(raw_headers)
790
out = response.handle_response('http://foo', code, msg, BytesIO(body))
790
getheader = self._build_HTTPMessage(raw_headers)
791
out = response.handle_response(
792
'http://foo', code, getheader, BytesIO(body))
791
793
self.assertEqual(body, out.read())
793
795
def test_missing_content_range(self):
794
796
code, raw_headers, body = _single_range_no_content_range
795
msg = self._build_HTTPMessage(raw_headers)
797
getheader = self._build_HTTPMessage(raw_headers)
796
798
self.assertRaises(errors.InvalidHttpResponse,
797
799
response.handle_response,
798
'http://bogus', code, msg, BytesIO(body))
800
'http://bogus', code, getheader, BytesIO(body))
800
802
def test_multipart_no_content_range(self):
801
803
code, raw_headers, body = _multipart_no_content_range
802
msg = self._build_HTTPMessage(raw_headers)
804
getheader = self._build_HTTPMessage(raw_headers)
803
805
self.assertRaises(errors.InvalidHttpResponse,
804
806
response.handle_response,
805
'http://bogus', code, msg, BytesIO(body))
807
'http://bogus', code, getheader, BytesIO(body))
807
809
def test_multipart_no_boundary(self):
808
810
out = self.get_response(_multipart_no_boundary)