Rietveld Code Review Tool
Help | Bug tracker | Discussion group | Source code | Sign in
(23081)

Delta Between Two Patch Sets: Lib/test/test_httplib.py

Issue 16037: httplib: header parsing is not delimited
Left Patch Set: Created 6 years ago
Right Patch Set: Created 5 years, 1 month ago
Left:
Right:
Use n/p to move between diff chunks; N/P to move between comments. Please Sign in to add in-line comments.
Jump to:
Left: Side by side diff | Download
Right: Side by side diff | Download
« no previous file with change/comment | « Lib/httplib.py ('k') | no next file » | no next file with change/comment »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
LEFTRIGHT
1 import httplib
2 import array
3 import httplib
4 import StringIO
5 import socket
1 import errno 6 import errno
2 from http import client
3 import io
4 import os
5 import array
6 import socket
7 7
8 import unittest 8 import unittest
9 TestCase = unittest.TestCase 9 TestCase = unittest.TestCase
10 10
11 from test import support 11 from test import test_support
12 12
13 here = os.path.dirname(__file__) 13 HOST = test_support.HOST
14 # Self-signed cert file for 'localhost'
15 CERT_localhost = os.path.join(here, 'keycert.pem')
16 # Self-signed cert file for 'fakehostname'
17 CERT_fakehostname = os.path.join(here, 'keycert2.pem')
18 # Root cert file (CA) for svn.python.org's cert
19 CACERT_svn_python_org = os.path.join(here, 'https_svn_python_org_root.pem')
20
21 HOST = support.HOST
22 14
23 class FakeSocket: 15 class FakeSocket:
24 def __init__(self, text, fileclass=io.BytesIO): 16 def __init__(self, text, fileclass=StringIO.StringIO, host=None, port=None):
25 if isinstance(text, str):
26 text = text.encode("ascii")
27 self.text = text 17 self.text = text
28 self.fileclass = fileclass 18 self.fileclass = fileclass
29 self.data = b'' 19 self.data = ''
20 self.host = host
21 self.port = port
30 22
31 def sendall(self, data): 23 def sendall(self, data):
32 self.data += data 24 self.data += ''.join(data)
33 25
34 def makefile(self, mode, bufsize=None): 26 def makefile(self, mode, bufsize=None):
35 if mode != 'r' and mode != 'rb': 27 if mode != 'r' and mode != 'rb':
36 raise client.UnimplementedFileMode() 28 raise httplib.UnimplementedFileMode()
37 return self.fileclass(self.text) 29 return self.fileclass(self.text)
30
31 def close(self):
32 pass
38 33
39 class EPipeSocket(FakeSocket): 34 class EPipeSocket(FakeSocket):
40 35
41 def __init__(self, text, pipe_trigger): 36 def __init__(self, text, pipe_trigger):
42 # When sendall() is called with pipe_trigger, raise EPIPE. 37 # When sendall() is called with pipe_trigger, raise EPIPE.
43 FakeSocket.__init__(self, text) 38 FakeSocket.__init__(self, text)
44 self.pipe_trigger = pipe_trigger 39 self.pipe_trigger = pipe_trigger
45 40
46 def sendall(self, data): 41 def sendall(self, data):
47 if self.pipe_trigger in data: 42 if self.pipe_trigger in data:
48 raise socket.error(errno.EPIPE, "gotcha") 43 raise socket.error(errno.EPIPE, "gotcha")
49 self.data += data 44 self.data += data
50 45
51 def close(self): 46 def close(self):
52 pass 47 pass
53 48
54 class NoEOFStringIO(io.BytesIO): 49 class NoEOFStringIO(StringIO.StringIO):
55 """Like StringIO, but raises AssertionError on EOF. 50 """Like StringIO, but raises AssertionError on EOF.
56 51
57 This is used below to test that http.client doesn't try to read 52 This is used below to test that httplib doesn't try to read
58 more from the underlying file than it should. 53 more from the underlying file than it should.
59 """ 54 """
60 def read(self, n=-1): 55 def read(self, n=-1):
61 data = io.BytesIO.read(self, n) 56 data = StringIO.StringIO.read(self, n)
62 if data == b'': 57 if data == '':
63 raise AssertionError('caller tried to read past EOF') 58 raise AssertionError('caller tried to read past EOF')
64 return data 59 return data
65 60
66 def readline(self, length=None): 61 def readline(self, length=None):
67 data = io.BytesIO.readline(self, length) 62 data = StringIO.StringIO.readline(self, length)
68 if data == b'': 63 if data == '':
69 raise AssertionError('caller tried to read past EOF') 64 raise AssertionError('caller tried to read past EOF')
70 return data 65 return data
66
71 67
72 class HeaderTests(TestCase): 68 class HeaderTests(TestCase):
73 def test_auto_headers(self): 69 def test_auto_headers(self):
74 # Some headers are added automatically, but should not be added by 70 # Some headers are added automatically, but should not be added by
75 # .request() if they are explicitly set. 71 # .request() if they are explicitly set.
76 72
77 class HeaderCountingBuffer(list): 73 class HeaderCountingBuffer(list):
78 def __init__(self): 74 def __init__(self):
79 self.count = {} 75 self.count = {}
80 def append(self, item): 76 def append(self, item):
81 kv = item.split(b':') 77 kv = item.split(':')
82 if len(kv) > 1: 78 if len(kv) > 1:
83 # item is a 'Key: Value' header string 79 # item is a 'Key: Value' header string
84 lcKey = kv[0].decode('ascii').lower() 80 lcKey = kv[0].lower()
85 self.count.setdefault(lcKey, 0) 81 self.count.setdefault(lcKey, 0)
86 self.count[lcKey] += 1 82 self.count[lcKey] += 1
87 list.append(self, item) 83 list.append(self, item)
88 84
89 for explicit_header in True, False: 85 for explicit_header in True, False:
90 for header in 'Content-length', 'Host', 'Accept-encoding': 86 for header in 'Content-length', 'Host', 'Accept-encoding':
91 conn = client.HTTPConnection('example.com') 87 conn = httplib.HTTPConnection('example.com')
92 conn.sock = FakeSocket('blahblahblah') 88 conn.sock = FakeSocket('blahblahblah')
93 conn._buffer = HeaderCountingBuffer() 89 conn._buffer = HeaderCountingBuffer()
94 90
95 body = 'spamspamspam' 91 body = 'spamspamspam'
96 headers = {} 92 headers = {}
97 if explicit_header: 93 if explicit_header:
98 headers[header] = str(len(body)) 94 headers[header] = str(len(body))
99 conn.request('POST', '/', body, headers) 95 conn.request('POST', '/', body, headers)
100 self.assertEqual(conn._buffer.count[header.lower()], 1) 96 self.assertEqual(conn._buffer.count[header.lower()], 1)
101 97
102 def test_content_length_0(self): 98 def test_content_length_0(self):
103 99
104 class ContentLengthChecker(list): 100 class ContentLengthChecker(list):
105 def __init__(self): 101 def __init__(self):
106 list.__init__(self) 102 list.__init__(self)
107 self.content_length = None 103 self.content_length = None
108 def append(self, item): 104 def append(self, item):
109 kv = item.split(b':', 1) 105 kv = item.split(':', 1)
110 if len(kv) > 1 and kv[0].lower() == b'content-length': 106 if len(kv) > 1 and kv[0].lower() == 'content-length':
111 self.content_length = kv[1].strip() 107 self.content_length = kv[1].strip()
112 list.append(self, item) 108 list.append(self, item)
113 109
114 # POST with empty body 110 # POST with empty body
115 conn = client.HTTPConnection('example.com') 111 conn = httplib.HTTPConnection('example.com')
116 conn.sock = FakeSocket(None) 112 conn.sock = FakeSocket(None)
117 conn._buffer = ContentLengthChecker() 113 conn._buffer = ContentLengthChecker()
118 conn.request('POST', '/', '') 114 conn.request('POST', '/', '')
119 self.assertEqual(conn._buffer.content_length, b'0', 115 self.assertEqual(conn._buffer.content_length, '0',
120 'Header Content-Length not set') 116 'Header Content-Length not set')
121 117
122 # PUT request with empty body 118 # PUT request with empty body
123 conn = client.HTTPConnection('example.com') 119 conn = httplib.HTTPConnection('example.com')
124 conn.sock = FakeSocket(None) 120 conn.sock = FakeSocket(None)
125 conn._buffer = ContentLengthChecker() 121 conn._buffer = ContentLengthChecker()
126 conn.request('PUT', '/', '') 122 conn.request('PUT', '/', '')
127 self.assertEqual(conn._buffer.content_length, b'0', 123 self.assertEqual(conn._buffer.content_length, '0',
128 'Header Content-Length not set') 124 'Header Content-Length not set')
129 125
130 def test_putheader(self): 126 def test_putheader(self):
131 conn = client.HTTPConnection('example.com') 127 conn = httplib.HTTPConnection('example.com')
132 conn.sock = FakeSocket(None) 128 conn.sock = FakeSocket(None)
133 conn.putrequest('GET','/') 129 conn.putrequest('GET','/')
134 conn.putheader('Content-length', 42) 130 conn.putheader('Content-length',42)
135 self.assertTrue(b'Content-length: 42' in conn._buffer) 131 self.assertIn('Content-length: 42', conn._buffer)
136 132
137 def test_ipv6host_header(self): 133 def test_ipv6host_header(self):
138 # Default host header on IPv6 transaction should wrapped by [] if 134 # Default host header on IPv6 transaction should wrapped by [] if
139 # its actual IPv6 address 135 # its actual IPv6 address
140 expected = b'GET /foo HTTP/1.1\r\nHost: [2001::]:81\r\n' \ 136 expected = 'GET /foo HTTP/1.1\r\nHost: [2001::]:81\r\n' \
141 b'Accept-Encoding: identity\r\n\r\n' 137 'Accept-Encoding: identity\r\n\r\n'
142 conn = client.HTTPConnection('[2001::]:81') 138 conn = httplib.HTTPConnection('[2001::]:81')
143 sock = FakeSocket('') 139 sock = FakeSocket('')
144 conn.sock = sock 140 conn.sock = sock
145 conn.request('GET', '/foo') 141 conn.request('GET', '/foo')
146 self.assertTrue(sock.data.startswith(expected)) 142 self.assertTrue(sock.data.startswith(expected))
147 143
148 expected = b'GET /foo HTTP/1.1\r\nHost: [2001:102A::]\r\n' \ 144 expected = 'GET /foo HTTP/1.1\r\nHost: [2001:102A::]\r\n' \
149 b'Accept-Encoding: identity\r\n\r\n' 145 'Accept-Encoding: identity\r\n\r\n'
150 conn = client.HTTPConnection('[2001:102A::]') 146 conn = httplib.HTTPConnection('[2001:102A::]')
151 sock = FakeSocket('') 147 sock = FakeSocket('')
152 conn.sock = sock 148 conn.sock = sock
153 conn.request('GET', '/foo') 149 conn.request('GET', '/foo')
154 self.assertTrue(sock.data.startswith(expected)) 150 self.assertTrue(sock.data.startswith(expected))
155 151
156 152
157 class BasicTest(TestCase): 153 class BasicTest(TestCase):
158 def test_status_lines(self): 154 def test_status_lines(self):
159 # Test HTTP status lines 155 # Test HTTP status lines
160 156
161 body = "HTTP/1.1 200 Ok\r\n\r\nText" 157 body = "HTTP/1.1 200 Ok\r\n\r\nText"
162 sock = FakeSocket(body) 158 sock = FakeSocket(body)
163 resp = client.HTTPResponse(sock) 159 resp = httplib.HTTPResponse(sock)
164 resp.begin() 160 resp.begin()
165 self.assertEqual(resp.read(), b"Text") 161 self.assertEqual(resp.read(0), '') # Issue #20007
166 self.assertTrue(resp.isclosed()) 162 self.assertFalse(resp.isclosed())
167 self.assertFalse(resp.closed) 163 self.assertEqual(resp.read(), 'Text')
168 resp.close() 164 self.assertTrue(resp.isclosed())
169 self.assertTrue(resp.closed)
170 165
171 body = "HTTP/1.1 400.100 Not Ok\r\n\r\nText" 166 body = "HTTP/1.1 400.100 Not Ok\r\n\r\nText"
172 sock = FakeSocket(body) 167 sock = FakeSocket(body)
173 resp = client.HTTPResponse(sock) 168 resp = httplib.HTTPResponse(sock)
174 self.assertRaises(client.BadStatusLine, resp.begin) 169 self.assertRaises(httplib.BadStatusLine, resp.begin)
175 170
176 def test_bad_status_repr(self): 171 def test_bad_status_repr(self):
177 exc = client.BadStatusLine('') 172 exc = httplib.BadStatusLine('')
178 self.assertEqual(repr(exc), '''BadStatusLine("\'\'",)''') 173 self.assertEqual(repr(exc), '''BadStatusLine("\'\'",)''')
179 174
180 def test_partial_reads(self): 175 def test_partial_reads(self):
181 # if we have a length, the system knows when to close itself 176 # if we have a length, the system knows when to close itself
182 # same behaviour than when we read the whole thing with read() 177 # same behaviour than when we read the whole thing with read()
183 body = "HTTP/1.1 200 Ok\r\nContent-Length: 4\r\n\r\nText" 178 body = "HTTP/1.1 200 Ok\r\nContent-Length: 4\r\n\r\nText"
184 sock = FakeSocket(body) 179 sock = FakeSocket(body)
185 resp = client.HTTPResponse(sock) 180 resp = httplib.HTTPResponse(sock)
186 resp.begin() 181 resp.begin()
187 self.assertEqual(resp.read(2), b'Te') 182 self.assertEqual(resp.read(2), 'Te')
188 self.assertFalse(resp.isclosed()) 183 self.assertFalse(resp.isclosed())
189 self.assertEqual(resp.read(2), b'xt') 184 self.assertEqual(resp.read(2), 'xt')
190 self.assertTrue(resp.isclosed()) 185 self.assertTrue(resp.isclosed())
191 self.assertFalse(resp.closed)
192 resp.close()
193 self.assertTrue(resp.closed)
194 186
195 def test_partial_reads_no_content_length(self): 187 def test_partial_reads_no_content_length(self):
196 # when no length is present, the socket should be gracefully closed when 188 # when no length is present, the socket should be gracefully closed when
197 # all data was read 189 # all data was read
198 body = "HTTP/1.1 200 Ok\r\n\r\nText" 190 body = "HTTP/1.1 200 Ok\r\n\r\nText"
199 sock = FakeSocket(body) 191 sock = FakeSocket(body)
200 resp = client.HTTPResponse(sock) 192 resp = httplib.HTTPResponse(sock)
201 resp.begin() 193 resp.begin()
202 self.assertEqual(resp.read(2), b'Te') 194 self.assertEqual(resp.read(2), 'Te')
203 self.assertFalse(resp.isclosed()) 195 self.assertFalse(resp.isclosed())
204 self.assertEqual(resp.read(2), b'xt') 196 self.assertEqual(resp.read(2), 'xt')
205 self.assertEqual(resp.read(1), b'') 197 self.assertEqual(resp.read(1), '')
206 self.assertTrue(resp.isclosed()) 198 self.assertTrue(resp.isclosed())
207 self.assertFalse(resp.closed)
208 resp.close()
209 self.assertTrue(resp.closed)
210 199
211 def test_partial_reads_incomplete_body(self): 200 def test_partial_reads_incomplete_body(self):
212 # if the server shuts down the connection before the whole 201 # if the server shuts down the connection before the whole
213 # content-length is delivered, the socket is gracefully closed 202 # content-length is delivered, the socket is gracefully closed
214 body = "HTTP/1.1 200 Ok\r\nContent-Length: 10\r\n\r\nText" 203 body = "HTTP/1.1 200 Ok\r\nContent-Length: 10\r\n\r\nText"
215 sock = FakeSocket(body) 204 sock = FakeSocket(body)
216 resp = client.HTTPResponse(sock) 205 resp = httplib.HTTPResponse(sock)
217 resp.begin() 206 resp.begin()
218 self.assertEqual(resp.read(2), b'Te') 207 self.assertEqual(resp.read(2), 'Te')
219 self.assertFalse(resp.isclosed()) 208 self.assertFalse(resp.isclosed())
220 self.assertEqual(resp.read(2), b'xt') 209 self.assertEqual(resp.read(2), 'xt')
221 self.assertEqual(resp.read(1), b'') 210 self.assertEqual(resp.read(1), '')
222 self.assertTrue(resp.isclosed()) 211 self.assertTrue(resp.isclosed())
223 self.assertFalse(resp.closed)
224 resp.close()
225 self.assertTrue(resp.closed)
226 212
227 def test_host_port(self): 213 def test_host_port(self):
228 # Check invalid host_port 214 # Check invalid host_port
229 215
216 # Note that httplib does not accept user:password@ in the host-port.
230 for hp in ("www.python.org:abc", "user:password@www.python.org"): 217 for hp in ("www.python.org:abc", "user:password@www.python.org"):
231 self.assertRaises(client.InvalidURL, client.HTTPConnection, hp) 218 self.assertRaises(httplib.InvalidURL, httplib.HTTP, hp)
232 219
233 for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000", 220 for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000", "fe80::207:e9ff:fe9b",
234 "fe80::207:e9ff:fe9b", 8000), 221 8000),
235 ("www.python.org:80", "www.python.org", 80), 222 ("www.python.org:80", "www.python.org", 80),
223 ("www.python.org", "www.python.org", 80),
236 ("www.python.org:", "www.python.org", 80), 224 ("www.python.org:", "www.python.org", 80),
237 ("www.python.org", "www.python.org", 80), 225 ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 80)):
238 ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 80), 226 http = httplib.HTTP(hp)
239 ("[fe80::207:e9ff:fe9b]:", "fe80::207:e9ff:fe9b", 80)): 227 c = http._conn
240 c = client.HTTPConnection(hp) 228 if h != c.host:
241 self.assertEqual(h, c.host) 229 self.fail("Host incorrectly parsed: %s != %s" % (h, c.host))
242 self.assertEqual(p, c.port) 230 if p != c.port:
231 self.fail("Port incorrectly parsed: %s != %s" % (p, c.host))
243 232
244 def test_response_headers(self): 233 def test_response_headers(self):
245 # test response with multiple message headers with the same field name. 234 # test response with multiple message headers with the same field name.
246 text = ('HTTP/1.1 200 OK\r\n' 235 text = ('HTTP/1.1 200 OK\r\n'
247 'Set-Cookie: Customer="WILE_E_COYOTE"; ' 236 'Set-Cookie: Customer="WILE_E_COYOTE";'
248 'Version="1"; Path="/acme"\r\n' 237 ' Version="1"; Path="/acme"\r\n'
249 'Set-Cookie: Part_Number="Rocket_Launcher_0001"; Version="1";' 238 'Set-Cookie: Part_Number="Rocket_Launcher_0001"; Version="1";'
250 ' Path="/acme"\r\n' 239 ' Path="/acme"\r\n'
251 '\r\n' 240 '\r\n'
252 'No body\r\n') 241 'No body\r\n')
253 hdr = ('Customer="WILE_E_COYOTE"; Version="1"; Path="/acme"' 242 hdr = ('Customer="WILE_E_COYOTE"; Version="1"; Path="/acme"'
254 ', ' 243 ', '
255 'Part_Number="Rocket_Launcher_0001"; Version="1"; Path="/acme"') 244 'Part_Number="Rocket_Launcher_0001"; Version="1"; Path="/acme"')
256 s = FakeSocket(text) 245 s = FakeSocket(text)
257 r = client.HTTPResponse(s) 246 r = httplib.HTTPResponse(s)
258 r.begin() 247 r.begin()
259 cookies = r.getheader("Set-Cookie") 248 cookies = r.getheader("Set-Cookie")
260 self.assertEqual(cookies, hdr) 249 if cookies != hdr:
250 self.fail("multiple headers not combined properly")
261 251
262 def test_read_head(self): 252 def test_read_head(self):
263 # Test that the library doesn't attempt to read any data 253 # Test that the library doesn't attempt to read any data
264 # from a HEAD request. (Tickles SF bug #622042.) 254 # from a HEAD request. (Tickles SF bug #622042.)
265 sock = FakeSocket( 255 sock = FakeSocket(
266 'HTTP/1.1 200 OK\r\n' 256 'HTTP/1.1 200 OK\r\n'
267 'Content-Length: 14432\r\n' 257 'Content-Length: 14432\r\n'
268 '\r\n', 258 '\r\n',
269 NoEOFStringIO) 259 NoEOFStringIO)
270 resp = client.HTTPResponse(sock, method="HEAD") 260 resp = httplib.HTTPResponse(sock, method="HEAD")
271 resp.begin() 261 resp.begin()
272 if resp.read(): 262 if resp.read() != "":
273 self.fail("Did not expect response from HEAD request") 263 self.fail("Did not expect response from HEAD request")
274 264
275 def test_too_many_headers(self): 265 def test_too_many_headers(self):
276 headers = '\r\n'.join('Header%d: foo' % i for i in range(200)) + '\r\n' 266 headers = '\r\n'.join('Header%d: foo' % i for i in xrange(200)) + '\r\n'
277 text = ('HTTP/1.1 200 OK\r\n' + headers) 267 text = ('HTTP/1.1 200 OK\r\n' + headers)
278 s = FakeSocket(text) 268 s = FakeSocket(text)
279 r = client.HTTPResponse(s) 269 r = httplib.HTTPResponse(s)
280 self.assertRaises(client.TooManyHeaders, r.begin) 270 self.assertRaises(httplib.HTTPException, r.begin)
281 271
282 def test_send_file(self): 272 def test_send_file(self):
283 expected = (b'GET /foo HTTP/1.1\r\nHost: example.com\r\n' 273 expected = 'GET /foo HTTP/1.1\r\nHost: example.com\r\n' \
284 b'Accept-Encoding: identity\r\nContent-Length:') 274 'Accept-Encoding: identity\r\nContent-Length:'
285 275
286 with open(__file__, 'rb') as body: 276 body = open(__file__, 'rb')
287 conn = client.HTTPConnection('example.com') 277 conn = httplib.HTTPConnection('example.com')
288 sock = FakeSocket(body) 278 sock = FakeSocket(body)
289 conn.sock = sock 279 conn.sock = sock
290 conn.request('GET', '/foo', body) 280 conn.request('GET', '/foo', body)
291 self.assertTrue(sock.data.startswith(expected), '%r != %r' % 281 self.assertTrue(sock.data.startswith(expected))
292 (sock.data[:len(expected)], expected))
293 282
294 def test_send(self): 283 def test_send(self):
295 expected = b'this is a test this is only a test' 284 expected = 'this is a test this is only a test'
296 conn = client.HTTPConnection('example.com') 285 conn = httplib.HTTPConnection('example.com')
297 sock = FakeSocket(None) 286 sock = FakeSocket(None)
298 conn.sock = sock 287 conn.sock = sock
299 conn.send(expected) 288 conn.send(expected)
300 self.assertEqual(expected, sock.data) 289 self.assertEqual(expected, sock.data)
301 sock.data = b'' 290 sock.data = ''
302 conn.send(array.array('b', expected)) 291 conn.send(array.array('c', expected))
303 self.assertEqual(expected, sock.data) 292 self.assertEqual(expected, sock.data)
304 sock.data = b'' 293 sock.data = ''
305 conn.send(io.BytesIO(expected)) 294 conn.send(StringIO.StringIO(expected))
306 self.assertEqual(expected, sock.data) 295 self.assertEqual(expected, sock.data)
307
308 def test_send_iter(self):
309 expected = b'GET /foo HTTP/1.1\r\nHost: example.com\r\n' \
310 b'Accept-Encoding: identity\r\nContent-Length: 11\r\n' \
311 b'\r\nonetwothree'
312
313 def body():
314 yield b"one"
315 yield b"two"
316 yield b"three"
317
318 conn = client.HTTPConnection('example.com')
319 sock = FakeSocket("")
320 conn.sock = sock
321 conn.request('GET', '/foo', body(), {'Content-Length': '11'})
322 self.assertEqual(sock.data, expected)
323
324 def test_send_type_error(self):
325 # See: Issue #12676
326 conn = client.HTTPConnection('example.com')
327 conn.sock = FakeSocket('')
328 with self.assertRaises(TypeError):
329 conn.request('POST', 'test', conn)
330 296
331 def test_chunked(self): 297 def test_chunked(self):
332 chunked_start = ( 298 chunked_start = (
333 'HTTP/1.1 200 OK\r\n' 299 'HTTP/1.1 200 OK\r\n'
334 'Transfer-Encoding: chunked\r\n\r\n' 300 'Transfer-Encoding: chunked\r\n\r\n'
335 'a\r\n' 301 'a\r\n'
336 'hello worl\r\n' 302 'hello worl\r\n'
337 '1\r\n' 303 '1\r\n'
338 'd\r\n' 304 'd\r\n'
339 ) 305 )
340 sock = FakeSocket(chunked_start + '0\r\n') 306 sock = FakeSocket(chunked_start + '0\r\n')
341 resp = client.HTTPResponse(sock, method="GET") 307 resp = httplib.HTTPResponse(sock, method="GET")
342 resp.begin() 308 resp.begin()
343 self.assertEqual(resp.read(), b'hello world') 309 self.assertEqual(resp.read(), 'hello world')
344 resp.close() 310 resp.close()
345 311
346 for x in ('', 'foo\r\n'): 312 for x in ('', 'foo\r\n'):
347 sock = FakeSocket(chunked_start + x) 313 sock = FakeSocket(chunked_start + x)
348 resp = client.HTTPResponse(sock, method="GET") 314 resp = httplib.HTTPResponse(sock, method="GET")
349 resp.begin() 315 resp.begin()
350 try: 316 try:
351 resp.read() 317 resp.read()
352 except client.IncompleteRead as i: 318 except httplib.IncompleteRead, i:
353 self.assertEqual(i.partial, b'hello world') 319 self.assertEqual(i.partial, 'hello world')
354 self.assertEqual(repr(i),'IncompleteRead(11 bytes read)') 320 self.assertEqual(repr(i),'IncompleteRead(11 bytes read)')
355 self.assertEqual(str(i),'IncompleteRead(11 bytes read)') 321 self.assertEqual(str(i),'IncompleteRead(11 bytes read)')
356 else: 322 else:
357 self.fail('IncompleteRead expected') 323 self.fail('IncompleteRead expected')
358 finally: 324 finally:
359 resp.close() 325 resp.close()
360 326
361 def test_chunked_head(self): 327 def test_chunked_head(self):
362 chunked_start = ( 328 chunked_start = (
363 'HTTP/1.1 200 OK\r\n' 329 'HTTP/1.1 200 OK\r\n'
364 'Transfer-Encoding: chunked\r\n\r\n' 330 'Transfer-Encoding: chunked\r\n\r\n'
365 'a\r\n' 331 'a\r\n'
366 'hello world\r\n' 332 'hello world\r\n'
367 '1\r\n' 333 '1\r\n'
368 'd\r\n' 334 'd\r\n'
369 ) 335 )
370 sock = FakeSocket(chunked_start + '0\r\n') 336 sock = FakeSocket(chunked_start + '0\r\n')
371 resp = client.HTTPResponse(sock, method="HEAD") 337 resp = httplib.HTTPResponse(sock, method="HEAD")
372 resp.begin() 338 resp.begin()
373 self.assertEqual(resp.read(), b'') 339 self.assertEqual(resp.read(), '')
374 self.assertEqual(resp.status, 200) 340 self.assertEqual(resp.status, 200)
375 self.assertEqual(resp.reason, 'OK') 341 self.assertEqual(resp.reason, 'OK')
376 self.assertTrue(resp.isclosed()) 342 self.assertTrue(resp.isclosed())
377 self.assertFalse(resp.closed)
378 resp.close()
379 self.assertTrue(resp.closed)
380 343
381 def test_negative_content_length(self): 344 def test_negative_content_length(self):
382 sock = FakeSocket( 345 sock = FakeSocket('HTTP/1.1 200 OK\r\n'
383 'HTTP/1.1 200 OK\r\nContent-Length: -1\r\n\r\nHello\r\n') 346 'Content-Length: -1\r\n\r\nHello\r\n')
384 resp = client.HTTPResponse(sock, method="GET") 347 resp = httplib.HTTPResponse(sock, method="GET")
385 resp.begin() 348 resp.begin()
386 self.assertEqual(resp.read(), b'Hello\r\n') 349 self.assertEqual(resp.read(), 'Hello\r\n')
387 self.assertTrue(resp.isclosed()) 350 self.assertTrue(resp.isclosed())
388 351
389 def test_incomplete_read(self): 352 def test_incomplete_read(self):
390 sock = FakeSocket('HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\nHello\r\ n') 353 sock = FakeSocket('HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\nHello\r\ n')
391 resp = client.HTTPResponse(sock, method="GET") 354 resp = httplib.HTTPResponse(sock, method="GET")
392 resp.begin() 355 resp.begin()
393 try: 356 try:
394 resp.read() 357 resp.read()
395 except client.IncompleteRead as i: 358 except httplib.IncompleteRead as i:
396 self.assertEqual(i.partial, b'Hello\r\n') 359 self.assertEqual(i.partial, 'Hello\r\n')
397 self.assertEqual(repr(i), 360 self.assertEqual(repr(i),
398 "IncompleteRead(7 bytes read, 3 more expected)") 361 "IncompleteRead(7 bytes read, 3 more expected)")
399 self.assertEqual(str(i), 362 self.assertEqual(str(i),
400 "IncompleteRead(7 bytes read, 3 more expected)") 363 "IncompleteRead(7 bytes read, 3 more expected)")
401 self.assertTrue(resp.isclosed()) 364 self.assertTrue(resp.isclosed())
402 else: 365 else:
403 self.fail('IncompleteRead expected') 366 self.fail('IncompleteRead expected')
404 367
405 def test_epipe(self): 368 def test_epipe(self):
406 sock = EPipeSocket( 369 sock = EPipeSocket(
407 "HTTP/1.0 401 Authorization Required\r\n" 370 "HTTP/1.0 401 Authorization Required\r\n"
408 "Content-type: text/html\r\n" 371 "Content-type: text/html\r\n"
409 "WWW-Authenticate: Basic realm=\"example\"\r\n", 372 "WWW-Authenticate: Basic realm=\"example\"\r\n",
410 b"Content-Length") 373 b"Content-Length")
411 conn = client.HTTPConnection("example.com") 374 conn = httplib.HTTPConnection("example.com")
412 conn.sock = sock 375 conn.sock = sock
413 self.assertRaises(socket.error, 376 self.assertRaises(socket.error,
414 lambda: conn.request("PUT", "/url", "body")) 377 lambda: conn.request("PUT", "/url", "body"))
415 resp = conn.getresponse() 378 resp = conn.getresponse()
416 self.assertEqual(401, resp.status) 379 self.assertEqual(401, resp.status)
417 self.assertEqual("Basic realm=\"example\"", 380 self.assertEqual("Basic realm=\"example\"",
418 resp.getheader("www-authenticate")) 381 resp.getheader("www-authenticate"))
419 382
383 def test_filenoattr(self):
384 # Just test the fileno attribute in the HTTPResponse Object.
385 body = "HTTP/1.1 200 Ok\r\n\r\nText"
386 sock = FakeSocket(body)
387 resp = httplib.HTTPResponse(sock)
388 self.assertTrue(hasattr(resp,'fileno'),
389 'HTTPResponse should expose a fileno attribute')
390
420 # Test lines overflowing the max line size (_MAXLINE in http.client) 391 # Test lines overflowing the max line size (_MAXLINE in http.client)
421 392
422 def test_overflowing_status_line(self): 393 def test_overflowing_status_line(self):
394 self.skipTest("disabled for HTTP 0.9 support")
423 body = "HTTP/1.1 200 Ok" + "k" * 65536 + "\r\n" 395 body = "HTTP/1.1 200 Ok" + "k" * 65536 + "\r\n"
424 resp = client.HTTPResponse(FakeSocket(body)) 396 resp = httplib.HTTPResponse(FakeSocket(body))
425 self.assertRaises((client.LineTooLong, client.BadStatusLine), resp.begin ) 397 self.assertRaises((httplib.LineTooLong, httplib.BadStatusLine), resp.beg in)
426 398
427 def test_overflowing_header_line(self): 399 def test_overflowing_header_line(self):
428 body = ( 400 body = (
429 'HTTP/1.1 200 OK\r\n' 401 'HTTP/1.1 200 OK\r\n'
430 'X-Foo: bar' + 'r' * 65536 + '\r\n\r\n' 402 'X-Foo: bar' + 'r' * 65536 + '\r\n\r\n'
431 ) 403 )
432 resp = client.HTTPResponse(FakeSocket(body)) 404 resp = httplib.HTTPResponse(FakeSocket(body))
433 self.assertRaises(client.LineTooLong, resp.begin) 405 self.assertRaises(httplib.LineTooLong, resp.begin)
434 406
435 def test_overflowing_chunked_line(self): 407 def test_overflowing_chunked_line(self):
436 body = ( 408 body = (
437 'HTTP/1.1 200 OK\r\n' 409 'HTTP/1.1 200 OK\r\n'
438 'Transfer-Encoding: chunked\r\n\r\n' 410 'Transfer-Encoding: chunked\r\n\r\n'
439 + '0' * 65536 + 'a\r\n' 411 + '0' * 65536 + 'a\r\n'
440 'hello world\r\n' 412 'hello world\r\n'
441 '0\r\n' 413 '0\r\n'
442 ) 414 )
443 resp = client.HTTPResponse(FakeSocket(body)) 415 resp = httplib.HTTPResponse(FakeSocket(body))
444 resp.begin() 416 resp.begin()
445 self.assertRaises(client.LineTooLong, resp.read) 417 self.assertRaises(httplib.LineTooLong, resp.read)
446 418
447 def test_early_eof(self): 419 def test_early_eof(self):
448 # Test httpresponse with no \r\n termination, 420 # Test httpresponse with no \r\n termination,
449 body = "HTTP/1.1 200 Ok" 421 body = "HTTP/1.1 200 Ok"
450 sock = FakeSocket(body) 422 sock = FakeSocket(body)
451 resp = client.HTTPResponse(sock) 423 resp = httplib.HTTPResponse(sock)
452 resp.begin() 424 resp.begin()
453 self.assertEqual(resp.read(), b'') 425 self.assertEqual(resp.read(), '')
454 self.assertTrue(resp.isclosed()) 426 self.assertTrue(resp.isclosed())
455 self.assertFalse(resp.closed)
456 resp.close()
457 self.assertTrue(resp.closed)
458 427
459 class OfflineTest(TestCase): 428 class OfflineTest(TestCase):
460 def test_responses(self): 429 def test_responses(self):
461 self.assertEqual(client.responses[client.NOT_FOUND], "Not Found") 430 self.assertEqual(httplib.responses[httplib.NOT_FOUND], "Not Found")
462 431
463 432
464 class SourceAddressTest(TestCase): 433 class SourceAddressTest(TestCase):
465 def setUp(self): 434 def setUp(self):
466 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 435 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
467 self.port = support.bind_port(self.serv) 436 self.port = test_support.bind_port(self.serv)
468 self.source_port = support.find_unused_port() 437 self.source_port = test_support.find_unused_port()
469 self.serv.listen(5) 438 self.serv.listen(5)
470 self.conn = None 439 self.conn = None
471 440
472 def tearDown(self): 441 def tearDown(self):
473 if self.conn: 442 if self.conn:
474 self.conn.close() 443 self.conn.close()
475 self.conn = None 444 self.conn = None
476 self.serv.close() 445 self.serv.close()
477 self.serv = None 446 self.serv = None
478 447
479 def testHTTPConnectionSourceAddress(self): 448 def testHTTPConnectionSourceAddress(self):
480 self.conn = client.HTTPConnection(HOST, self.port, 449 self.conn = httplib.HTTPConnection(HOST, self.port,
481 source_address=('', self.source_port)) 450 source_address=('', self.source_port))
482 self.conn.connect() 451 self.conn.connect()
483 self.assertEqual(self.conn.sock.getsockname()[1], self.source_port) 452 self.assertEqual(self.conn.sock.getsockname()[1], self.source_port)
484 453
485 @unittest.skipIf(not hasattr(client, 'HTTPSConnection'), 454 @unittest.skipIf(not hasattr(httplib, 'HTTPSConnection'),
486 'http.client.HTTPSConnection not defined') 455 'httplib.HTTPSConnection not defined')
487 def testHTTPSConnectionSourceAddress(self): 456 def testHTTPSConnectionSourceAddress(self):
488 self.conn = client.HTTPSConnection(HOST, self.port, 457 self.conn = httplib.HTTPSConnection(HOST, self.port,
489 source_address=('', self.source_port)) 458 source_address=('', self.source_port))
490 # We don't test anything here other the constructor not barfing as 459 # We don't test anything here other the constructor not barfing as
491 # this code doesn't deal with setting up an active running SSL server 460 # this code doesn't deal with setting up an active running SSL server
492 # for an ssl_wrapped connect() to actually return from. 461 # for an ssl_wrapped connect() to actually return from.
493 462
494 463
495 class TimeoutTest(TestCase): 464 class TimeoutTest(TestCase):
496 PORT = None 465 PORT = None
497 466
498 def setUp(self): 467 def setUp(self):
499 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 468 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
500 TimeoutTest.PORT = support.bind_port(self.serv) 469 TimeoutTest.PORT = test_support.bind_port(self.serv)
501 self.serv.listen(5) 470 self.serv.listen(5)
502 471
503 def tearDown(self): 472 def tearDown(self):
504 self.serv.close() 473 self.serv.close()
505 self.serv = None 474 self.serv = None
506 475
507 def testTimeoutAttribute(self): 476 def testTimeoutAttribute(self):
508 # This will prove that the timeout gets through HTTPConnection 477 '''This will prove that the timeout gets through
509 # and into the socket. 478 HTTPConnection and into the socket.
510 479 '''
511 # default -- use global socket timeout 480 # default -- use global socket timeout
512 self.assertTrue(socket.getdefaulttimeout() is None) 481 self.assertIsNone(socket.getdefaulttimeout())
513 socket.setdefaulttimeout(30) 482 socket.setdefaulttimeout(30)
514 try: 483 try:
515 httpConn = client.HTTPConnection(HOST, TimeoutTest.PORT) 484 httpConn = httplib.HTTPConnection(HOST, TimeoutTest.PORT)
516 httpConn.connect() 485 httpConn.connect()
517 finally: 486 finally:
518 socket.setdefaulttimeout(None) 487 socket.setdefaulttimeout(None)
519 self.assertEqual(httpConn.sock.gettimeout(), 30) 488 self.assertEqual(httpConn.sock.gettimeout(), 30)
520 httpConn.close() 489 httpConn.close()
521 490
522 # no timeout -- do not use global socket default 491 # no timeout -- do not use global socket default
523 self.assertTrue(socket.getdefaulttimeout() is None) 492 self.assertIsNone(socket.getdefaulttimeout())
524 socket.setdefaulttimeout(30) 493 socket.setdefaulttimeout(30)
525 try: 494 try:
526 httpConn = client.HTTPConnection(HOST, TimeoutTest.PORT, 495 httpConn = httplib.HTTPConnection(HOST, TimeoutTest.PORT,
527 timeout=None) 496 timeout=None)
528 httpConn.connect() 497 httpConn.connect()
529 finally: 498 finally:
530 socket.setdefaulttimeout(None) 499 socket.setdefaulttimeout(None)
531 self.assertEqual(httpConn.sock.gettimeout(), None) 500 self.assertEqual(httpConn.sock.gettimeout(), None)
532 httpConn.close() 501 httpConn.close()
533 502
534 # a value 503 # a value
535 httpConn = client.HTTPConnection(HOST, TimeoutTest.PORT, timeout=30) 504 httpConn = httplib.HTTPConnection(HOST, TimeoutTest.PORT, timeout=30)
536 httpConn.connect() 505 httpConn.connect()
537 self.assertEqual(httpConn.sock.gettimeout(), 30) 506 self.assertEqual(httpConn.sock.gettimeout(), 30)
538 httpConn.close() 507 httpConn.close()
539 508
540 509
541 class HTTPSTest(TestCase): 510 class HTTPSTimeoutTest(TestCase):
542 511 # XXX Here should be tests for HTTPS, there isn't any right now!
543 def setUp(self):
544 if not hasattr(client, 'HTTPSConnection'):
545 self.skipTest('ssl support required')
546
547 def make_server(self, certfile):
548 from test.ssl_servers import make_https_server
549 return make_https_server(self, certfile)
550 512
551 def test_attributes(self): 513 def test_attributes(self):
552 # simple test to check it's storing the timeout 514 # simple test to check it's storing it
553 h = client.HTTPSConnection(HOST, TimeoutTest.PORT, timeout=30) 515 if hasattr(httplib, 'HTTPSConnection'):
554 self.assertEqual(h.timeout, 30) 516 h = httplib.HTTPSConnection(HOST, TimeoutTest.PORT, timeout=30)
555 517 self.assertEqual(h.timeout, 30)
556 def _check_svn_python_org(self, resp): 518
557 # Just a simple check that everything went fine 519 @unittest.skipIf(not hasattr(httplib, 'HTTPS'), 'httplib.HTTPS not available ')
558 server_string = resp.getheader('server')
559 self.assertIn('Apache', server_string)
560
561 def test_networked(self):
562 # Default settings: no cert verification is done
563 support.requires('network')
564 with support.transient_internet('svn.python.org'):
565 h = client.HTTPSConnection('svn.python.org', 443)
566 h.request('GET', '/')
567 resp = h.getresponse()
568 self._check_svn_python_org(resp)
569
570 def test_networked_good_cert(self):
571 # We feed a CA cert that validates the server's cert
572 import ssl
573 support.requires('network')
574 with support.transient_internet('svn.python.org'):
575 context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
576 context.verify_mode = ssl.CERT_REQUIRED
577 context.load_verify_locations(CACERT_svn_python_org)
578 h = client.HTTPSConnection('svn.python.org', 443, context=context)
579 h.request('GET', '/')
580 resp = h.getresponse()
581 self._check_svn_python_org(resp)
582
583 def test_networked_bad_cert(self):
584 # We feed a "CA" cert that is unrelated to the server's cert
585 import ssl
586 support.requires('network')
587 with support.transient_internet('svn.python.org'):
588 context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
589 context.verify_mode = ssl.CERT_REQUIRED
590 context.load_verify_locations(CERT_localhost)
591 h = client.HTTPSConnection('svn.python.org', 443, context=context)
592 with self.assertRaises(ssl.SSLError):
593 h.request('GET', '/')
594
595 def test_local_good_hostname(self):
596 # The (valid) cert validates the HTTP hostname
597 import ssl
598 from test.ssl_servers import make_https_server
599 server = make_https_server(self, CERT_localhost)
600 context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
601 context.verify_mode = ssl.CERT_REQUIRED
602 context.load_verify_locations(CERT_localhost)
603 h = client.HTTPSConnection('localhost', server.port, context=context)
604 h.request('GET', '/nonexistent')
605 resp = h.getresponse()
606 self.assertEqual(resp.status, 404)
607
608 def test_local_bad_hostname(self):
609 # The (valid) cert doesn't validate the HTTP hostname
610 import ssl
611 from test.ssl_servers import make_https_server
612 server = make_https_server(self, CERT_fakehostname)
613 context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
614 context.verify_mode = ssl.CERT_REQUIRED
615 context.load_verify_locations(CERT_fakehostname)
616 h = client.HTTPSConnection('localhost', server.port, context=context)
617 with self.assertRaises(ssl.CertificateError):
618 h.request('GET', '/')
619 # Same with explicit check_hostname=True
620 h = client.HTTPSConnection('localhost', server.port, context=context,
621 check_hostname=True)
622 with self.assertRaises(ssl.CertificateError):
623 h.request('GET', '/')
624 # With check_hostname=False, the mismatching is ignored
625 h = client.HTTPSConnection('localhost', server.port, context=context,
626 check_hostname=False)
627 h.request('GET', '/nonexistent')
628 resp = h.getresponse()
629 self.assertEqual(resp.status, 404)
630
631 @unittest.skipIf(not hasattr(client, 'HTTPSConnection'),
632 'http.client.HTTPSConnection not available')
633 def test_host_port(self): 520 def test_host_port(self):
634 # Check invalid host_port 521 # Check invalid host_port
635 522
523 # Note that httplib does not accept user:password@ in the host-port.
636 for hp in ("www.python.org:abc", "user:password@www.python.org"): 524 for hp in ("www.python.org:abc", "user:password@www.python.org"):
637 self.assertRaises(client.InvalidURL, client.HTTPSConnection, hp) 525 self.assertRaises(httplib.InvalidURL, httplib.HTTP, hp)
638 526
639 for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000", 527 for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000", "fe80::207:e9ff:fe9b",
640 "fe80::207:e9ff:fe9b", 8000), 528 8000),
641 ("www.python.org:443", "www.python.org", 443), 529 ("pypi.python.org:443", "pypi.python.org", 443),
642 ("www.python.org:", "www.python.org", 443), 530 ("pypi.python.org", "pypi.python.org", 443),
643 ("www.python.org", "www.python.org", 443), 531 ("pypi.python.org:", "pypi.python.org", 443),
644 ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 443), 532 ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 443)):
645 ("[fe80::207:e9ff:fe9b]:", "fe80::207:e9ff:fe9b", 533 http = httplib.HTTPS(hp)
646 443)): 534 c = http._conn
647 c = client.HTTPSConnection(hp) 535 if h != c.host:
648 self.assertEqual(h, c.host) 536 self.fail("Host incorrectly parsed: %s != %s" % (h, c.host))
649 self.assertEqual(p, c.port) 537 if p != c.port:
650 538 self.fail("Port incorrectly parsed: %s != %s" % (p, c.host))
651 539
652 class RequestBodyTest(TestCase): 540
653 """Test cases where a request includes a message body.""" 541 class TunnelTests(TestCase):
654 542 def test_connect(self):
655 def setUp(self): 543 response_text = (
656 self.conn = client.HTTPConnection('example.com') 544 'HTTP/1.0 200 OK\r\n\r\n' # Reply to CONNECT
657 self.conn.sock = self.sock = FakeSocket("") 545 'HTTP/1.1 200 OK\r\n' # Reply to HEAD
658 self.conn.sock = self.sock 546 'Content-Length: 42\r\n\r\n'
659 547 )
660 def get_headers_and_fp(self): 548
661 f = io.BytesIO(self.sock.data) 549 def create_connection(address, timeout=None, source_address=None):
662 f.readline() # read the request line 550 return FakeSocket(response_text, host=address[0], port=address[1])
663 message = client.parse_headers(f) 551
664 return message, f 552 conn = httplib.HTTPConnection('proxy.com')
665 553 conn._create_connection = create_connection
666 def test_manual_content_length(self): 554
667 # Set an incorrect content-length so that we can verify that 555 # Once connected, we should not be able to tunnel anymore
668 # it will not be over-ridden by the library. 556 conn.connect()
669 self.conn.request("PUT", "/url", "body", 557 self.assertRaises(RuntimeError, conn.set_tunnel, 'destination.com')
670 {"Content-Length": "42"}) 558
671 message, f = self.get_headers_and_fp() 559 # But if close the connection, we are good.
672 self.assertEqual("42", message.get("content-length")) 560 conn.close()
673 self.assertEqual(4, len(f.read())) 561 conn.set_tunnel('destination.com')
674 562 conn.request('HEAD', '/', '')
675 def test_ascii_body(self): 563
676 self.conn.request("PUT", "/url", "body") 564 self.assertEqual(conn.sock.host, 'proxy.com')
677 message, f = self.get_headers_and_fp() 565 self.assertEqual(conn.sock.port, 80)
678 self.assertEqual("text/plain", message.get_content_type()) 566 self.assertTrue('CONNECT destination.com' in conn.sock.data)
679 self.assertIsNone(message.get_charset()) 567 self.assertTrue('Host: destination.com' in conn.sock.data)
680 self.assertEqual("4", message.get("content-length")) 568
681 self.assertEqual(b'body', f.read()) 569 self.assertTrue('Host: proxy.com' not in conn.sock.data)
682 570
683 def test_latin1_body(self): 571 conn.close()
684 self.conn.request("PUT", "/url", "body\xc1") 572
685 message, f = self.get_headers_and_fp() 573 conn.request('PUT', '/', '')
686 self.assertEqual("text/plain", message.get_content_type()) 574 self.assertEqual(conn.sock.host, 'proxy.com')
687 self.assertIsNone(message.get_charset()) 575 self.assertEqual(conn.sock.port, 80)
688 self.assertEqual("5", message.get("content-length")) 576 self.assertTrue('CONNECT destination.com' in conn.sock.data)
689 self.assertEqual(b'body\xc1', f.read()) 577 self.assertTrue('Host: destination.com' in conn.sock.data)
690 578
691 def test_bytes_body(self):
692 self.conn.request("PUT", "/url", b"body\xc1")
693 message, f = self.get_headers_and_fp()
694 self.assertEqual("text/plain", message.get_content_type())
695 self.assertIsNone(message.get_charset())
696 self.assertEqual("5", message.get("content-length"))
697 self.assertEqual(b'body\xc1', f.read())
698
699 def test_file_body(self):
700 self.addCleanup(support.unlink, support.TESTFN)
701 with open(support.TESTFN, "w") as f:
702 f.write("body")
703 with open(support.TESTFN) as f:
704 self.conn.request("PUT", "/url", f)
705 message, f = self.get_headers_and_fp()
706 self.assertEqual("text/plain", message.get_content_type())
707 self.assertIsNone(message.get_charset())
708 self.assertEqual("4", message.get("content-length"))
709 self.assertEqual(b'body', f.read())
710
711 def test_binary_file_body(self):
712 self.addCleanup(support.unlink, support.TESTFN)
713 with open(support.TESTFN, "wb") as f:
714 f.write(b"body\xc1")
715 with open(support.TESTFN, "rb") as f:
716 self.conn.request("PUT", "/url", f)
717 message, f = self.get_headers_and_fp()
718 self.assertEqual("text/plain", message.get_content_type())
719 self.assertIsNone(message.get_charset())
720 self.assertEqual("5", message.get("content-length"))
721 self.assertEqual(b'body\xc1', f.read())
722
723
724 class HTTPResponseTest(TestCase):
725
726 def setUp(self):
727 body = "HTTP/1.1 200 Ok\r\nMy-Header: first-value\r\nMy-Header: \
728 second-value\r\n\r\nText"
729 sock = FakeSocket(body)
730 self.resp = client.HTTPResponse(sock)
731 self.resp.begin()
732
733 def test_getting_header(self):
734 header = self.resp.getheader('My-Header')
735 self.assertEqual(header, 'first-value, second-value')
736
737 header = self.resp.getheader('My-Header', 'some default')
738 self.assertEqual(header, 'first-value, second-value')
739
740 def test_getting_nonexistent_header_with_string_default(self):
741 header = self.resp.getheader('No-Such-Header', 'default-value')
742 self.assertEqual(header, 'default-value')
743
744 def test_getting_nonexistent_header_with_iterable_default(self):
745 header = self.resp.getheader('No-Such-Header', ['default', 'values'])
746 self.assertEqual(header, 'default, values')
747
748 header = self.resp.getheader('No-Such-Header', ('default', 'values'))
749 self.assertEqual(header, 'default, values')
750
751 def test_getting_nonexistent_header_without_default(self):
752 header = self.resp.getheader('No-Such-Header')
753 self.assertEqual(header, None)
754
755 def test_getting_header_defaultint(self):
756 header = self.resp.getheader('No-Such-Header',default=42)
757 self.assertEqual(header, 42)
758 579
759 def test_main(verbose=None): 580 def test_main(verbose=None):
760 support.run_unittest(HeaderTests, OfflineTest, BasicTest, TimeoutTest, 581 test_support.run_unittest(HeaderTests, OfflineTest, BasicTest, TimeoutTest,
761 HTTPSTest, RequestBodyTest, SourceAddressTest, 582 HTTPSTimeoutTest, SourceAddressTest, TunnelTests)
762 HTTPResponseTest)
763 583
764 if __name__ == '__main__': 584 if __name__ == '__main__':
765 test_main() 585 test_main()
LEFTRIGHT

RSS Feeds Recent Issues | This issue
This is Rietveld 894c83f36cb7+