OLD | NEW |
1 import errno | 1 import errno |
2 from http import client | 2 from http import client |
3 import io | 3 import io |
4 import os | 4 import os |
5 import array | 5 import array |
6 import socket | 6 import socket |
7 | 7 |
8 import unittest | 8 import unittest |
9 TestCase = unittest.TestCase | 9 TestCase = unittest.TestCase |
10 | 10 |
(...skipping 126 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
137 self.assertEqual(resp.read(), b"Text") | 137 self.assertEqual(resp.read(), b"Text") |
138 self.assertTrue(resp.isclosed()) | 138 self.assertTrue(resp.isclosed()) |
139 | 139 |
140 body = "HTTP/1.1 400.100 Not Ok\r\n\r\nText" | 140 body = "HTTP/1.1 400.100 Not Ok\r\n\r\nText" |
141 sock = FakeSocket(body) | 141 sock = FakeSocket(body) |
142 resp = client.HTTPResponse(sock) | 142 resp = client.HTTPResponse(sock) |
143 self.assertRaises(client.BadStatusLine, resp.begin) | 143 self.assertRaises(client.BadStatusLine, resp.begin) |
144 | 144 |
145 def test_bad_status_repr(self): | 145 def test_bad_status_repr(self): |
146 exc = client.BadStatusLine('') | 146 exc = client.BadStatusLine('') |
147 self.assertEquals(repr(exc), '''BadStatusLine("\'\'",)''') | 147 self.assertEqual(repr(exc), '''BadStatusLine("\'\'",)''') |
148 | 148 |
149 def test_partial_reads(self): | 149 def test_partial_reads(self): |
150 # if we have a lenght, the system knows when to close itself | 150 # if we have a lenght, the system knows when to close itself |
151 # same behaviour than when we read the whole thing with read() | 151 # same behaviour than when we read the whole thing with read() |
152 body = "HTTP/1.1 200 Ok\r\nContent-Length: 4\r\n\r\nText" | 152 body = "HTTP/1.1 200 Ok\r\nContent-Length: 4\r\n\r\nText" |
153 sock = FakeSocket(body) | 153 sock = FakeSocket(body) |
154 resp = client.HTTPResponse(sock) | 154 resp = client.HTTPResponse(sock) |
155 resp.begin() | 155 resp.begin() |
156 self.assertEqual(resp.read(2), b'Te') | 156 self.assertEqual(resp.read(2), b'Te') |
157 self.assertFalse(resp.isclosed()) | 157 self.assertFalse(resp.isclosed()) |
(...skipping 57 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
215 conn.request('GET', '/foo', body) | 215 conn.request('GET', '/foo', body) |
216 self.assertTrue(sock.data.startswith(expected), '%r != %r' % | 216 self.assertTrue(sock.data.startswith(expected), '%r != %r' % |
217 (sock.data[:len(expected)], expected)) | 217 (sock.data[:len(expected)], expected)) |
218 | 218 |
219 def test_send(self): | 219 def test_send(self): |
220 expected = b'this is a test this is only a test' | 220 expected = b'this is a test this is only a test' |
221 conn = client.HTTPConnection('example.com') | 221 conn = client.HTTPConnection('example.com') |
222 sock = FakeSocket(None) | 222 sock = FakeSocket(None) |
223 conn.sock = sock | 223 conn.sock = sock |
224 conn.send(expected) | 224 conn.send(expected) |
225 self.assertEquals(expected, sock.data) | 225 self.assertEqual(expected, sock.data) |
226 sock.data = b'' | 226 sock.data = b'' |
227 conn.send(array.array('b', expected)) | 227 conn.send(array.array('b', expected)) |
228 self.assertEquals(expected, sock.data) | 228 self.assertEqual(expected, sock.data) |
229 sock.data = b'' | 229 sock.data = b'' |
230 conn.send(io.BytesIO(expected)) | 230 conn.send(io.BytesIO(expected)) |
231 self.assertEquals(expected, sock.data) | 231 self.assertEqual(expected, sock.data) |
232 | 232 |
233 def test_chunked(self): | 233 def test_chunked(self): |
234 chunked_start = ( | 234 chunked_start = ( |
235 'HTTP/1.1 200 OK\r\n' | 235 'HTTP/1.1 200 OK\r\n' |
236 'Transfer-Encoding: chunked\r\n\r\n' | 236 'Transfer-Encoding: chunked\r\n\r\n' |
237 'a\r\n' | 237 'a\r\n' |
238 'hello worl\r\n' | 238 'hello worl\r\n' |
239 '1\r\n' | 239 '1\r\n' |
240 'd\r\n' | 240 'd\r\n' |
241 ) | 241 ) |
242 sock = FakeSocket(chunked_start + '0\r\n') | 242 sock = FakeSocket(chunked_start + '0\r\n') |
243 resp = client.HTTPResponse(sock, method="GET") | 243 resp = client.HTTPResponse(sock, method="GET") |
244 resp.begin() | 244 resp.begin() |
245 self.assertEquals(resp.read(), b'hello world') | 245 self.assertEqual(resp.read(), b'hello world') |
246 resp.close() | 246 resp.close() |
247 | 247 |
248 for x in ('', 'foo\r\n'): | 248 for x in ('', 'foo\r\n'): |
249 sock = FakeSocket(chunked_start + x) | 249 sock = FakeSocket(chunked_start + x) |
250 resp = client.HTTPResponse(sock, method="GET") | 250 resp = client.HTTPResponse(sock, method="GET") |
251 resp.begin() | 251 resp.begin() |
252 try: | 252 try: |
253 resp.read() | 253 resp.read() |
254 except client.IncompleteRead as i: | 254 except client.IncompleteRead as i: |
255 self.assertEquals(i.partial, b'hello world') | 255 self.assertEqual(i.partial, b'hello world') |
256 self.assertEqual(repr(i),'IncompleteRead(11 bytes read)') | 256 self.assertEqual(repr(i),'IncompleteRead(11 bytes read)') |
257 self.assertEqual(str(i),'IncompleteRead(11 bytes read)') | 257 self.assertEqual(str(i),'IncompleteRead(11 bytes read)') |
258 else: | 258 else: |
259 self.fail('IncompleteRead expected') | 259 self.fail('IncompleteRead expected') |
260 finally: | 260 finally: |
261 resp.close() | 261 resp.close() |
262 | 262 |
263 def test_chunked_head(self): | 263 def test_chunked_head(self): |
264 chunked_start = ( | 264 chunked_start = ( |
265 'HTTP/1.1 200 OK\r\n' | 265 'HTTP/1.1 200 OK\r\n' |
266 'Transfer-Encoding: chunked\r\n\r\n' | 266 'Transfer-Encoding: chunked\r\n\r\n' |
267 'a\r\n' | 267 'a\r\n' |
268 'hello world\r\n' | 268 'hello world\r\n' |
269 '1\r\n' | 269 '1\r\n' |
270 'd\r\n' | 270 'd\r\n' |
271 ) | 271 ) |
272 sock = FakeSocket(chunked_start + '0\r\n') | 272 sock = FakeSocket(chunked_start + '0\r\n') |
273 resp = client.HTTPResponse(sock, method="HEAD") | 273 resp = client.HTTPResponse(sock, method="HEAD") |
274 resp.begin() | 274 resp.begin() |
275 self.assertEquals(resp.read(), b'') | 275 self.assertEqual(resp.read(), b'') |
276 self.assertEquals(resp.status, 200) | 276 self.assertEqual(resp.status, 200) |
277 self.assertEquals(resp.reason, 'OK') | 277 self.assertEqual(resp.reason, 'OK') |
278 self.assertTrue(resp.isclosed()) | 278 self.assertTrue(resp.isclosed()) |
279 | 279 |
280 def test_negative_content_length(self): | 280 def test_negative_content_length(self): |
281 sock = FakeSocket( | 281 sock = FakeSocket( |
282 'HTTP/1.1 200 OK\r\nContent-Length: -1\r\n\r\nHello\r\n') | 282 'HTTP/1.1 200 OK\r\nContent-Length: -1\r\n\r\nHello\r\n') |
283 resp = client.HTTPResponse(sock, method="GET") | 283 resp = client.HTTPResponse(sock, method="GET") |
284 resp.begin() | 284 resp.begin() |
285 self.assertEquals(resp.read(), b'Hello\r\n') | 285 self.assertEqual(resp.read(), b'Hello\r\n') |
286 resp.close() | 286 resp.close() |
287 | 287 |
288 def test_incomplete_read(self): | 288 def test_incomplete_read(self): |
289 sock = FakeSocket('HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\nHello\r\
n') | 289 sock = FakeSocket('HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\nHello\r\
n') |
290 resp = client.HTTPResponse(sock, method="GET") | 290 resp = client.HTTPResponse(sock, method="GET") |
291 resp.begin() | 291 resp.begin() |
292 try: | 292 try: |
293 resp.read() | 293 resp.read() |
294 except client.IncompleteRead as i: | 294 except client.IncompleteRead as i: |
295 self.assertEquals(i.partial, b'Hello\r\n') | 295 self.assertEqual(i.partial, b'Hello\r\n') |
296 self.assertEqual(repr(i), | 296 self.assertEqual(repr(i), |
297 "IncompleteRead(7 bytes read, 3 more expected)") | 297 "IncompleteRead(7 bytes read, 3 more expected)") |
298 self.assertEqual(str(i), | 298 self.assertEqual(str(i), |
299 "IncompleteRead(7 bytes read, 3 more expected)") | 299 "IncompleteRead(7 bytes read, 3 more expected)") |
300 else: | 300 else: |
301 self.fail('IncompleteRead expected') | 301 self.fail('IncompleteRead expected') |
302 finally: | 302 finally: |
303 resp.close() | 303 resp.close() |
304 | 304 |
305 def test_epipe(self): | 305 def test_epipe(self): |
306 sock = EPipeSocket( | 306 sock = EPipeSocket( |
307 "HTTP/1.0 401 Authorization Required\r\n" | 307 "HTTP/1.0 401 Authorization Required\r\n" |
308 "Content-type: text/html\r\n" | 308 "Content-type: text/html\r\n" |
309 "WWW-Authenticate: Basic realm=\"example\"\r\n", | 309 "WWW-Authenticate: Basic realm=\"example\"\r\n", |
310 b"Content-Length") | 310 b"Content-Length") |
311 conn = client.HTTPConnection("example.com") | 311 conn = client.HTTPConnection("example.com") |
312 conn.sock = sock | 312 conn.sock = sock |
313 self.assertRaises(socket.error, | 313 self.assertRaises(socket.error, |
314 lambda: conn.request("PUT", "/url", "body")) | 314 lambda: conn.request("PUT", "/url", "body")) |
315 resp = conn.getresponse() | 315 resp = conn.getresponse() |
316 self.assertEqual(401, resp.status) | 316 self.assertEqual(401, resp.status) |
317 self.assertEqual("Basic realm=\"example\"", | 317 self.assertEqual("Basic realm=\"example\"", |
318 resp.getheader("www-authenticate")) | 318 resp.getheader("www-authenticate")) |
319 | 319 |
320 class OfflineTest(TestCase): | 320 class OfflineTest(TestCase): |
321 def test_responses(self): | 321 def test_responses(self): |
322 self.assertEquals(client.responses[client.NOT_FOUND], "Not Found") | 322 self.assertEqual(client.responses[client.NOT_FOUND], "Not Found") |
323 | 323 |
324 | 324 |
325 class SourceAddressTest(TestCase): | 325 class SourceAddressTest(TestCase): |
326 def setUp(self): | 326 def setUp(self): |
327 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | 327 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
328 self.port = support.bind_port(self.serv) | 328 self.port = support.bind_port(self.serv) |
329 self.source_port = support.find_unused_port() | 329 self.source_port = support.find_unused_port() |
330 self.serv.listen(5) | 330 self.serv.listen(5) |
331 self.conn = None | 331 self.conn = None |
332 | 332 |
(...skipping 262 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
595 header = self.resp.getheader('No-Such-Header',default=42) | 595 header = self.resp.getheader('No-Such-Header',default=42) |
596 self.assertEqual(header, 42) | 596 self.assertEqual(header, 42) |
597 | 597 |
598 def test_main(verbose=None): | 598 def test_main(verbose=None): |
599 support.run_unittest(HeaderTests, OfflineTest, BasicTest, TimeoutTest, | 599 support.run_unittest(HeaderTests, OfflineTest, BasicTest, TimeoutTest, |
600 HTTPSTest, RequestBodyTest, SourceAddressTest, | 600 HTTPSTest, RequestBodyTest, SourceAddressTest, |
601 HTTPResponseTest) | 601 HTTPResponseTest) |
602 | 602 |
603 if __name__ == '__main__': | 603 if __name__ == '__main__': |
604 test_main() | 604 test_main() |
OLD | NEW |