(self)
| 1466 | self.assertEqual(result, b"proxied data\n") |
| 1467 | |
| 1468 | def test_large_content_length(self): |
| 1469 | serv = socket.create_server((HOST, 0)) |
| 1470 | self.addCleanup(serv.close) |
| 1471 | |
| 1472 | def run_server(): |
| 1473 | [conn, address] = serv.accept() |
| 1474 | with conn: |
| 1475 | while conn.recv(1024): |
| 1476 | conn.sendall( |
| 1477 | b"HTTP/1.1 200 Ok\r\n" |
| 1478 | b"Content-Length: %d\r\n" |
| 1479 | b"\r\n" % size) |
| 1480 | conn.sendall(b'A' * (size//3)) |
| 1481 | conn.sendall(b'B' * (size - size//3)) |
| 1482 | |
| 1483 | thread = threading.Thread(target=run_server) |
| 1484 | thread.start() |
| 1485 | self.addCleanup(thread.join, 1.0) |
| 1486 | |
| 1487 | conn = client.HTTPConnection(*serv.getsockname()) |
| 1488 | try: |
| 1489 | for w in range(15, 27): |
| 1490 | size = 1 << w |
| 1491 | conn.request("GET", "/") |
| 1492 | with conn.getresponse() as response: |
| 1493 | self.assertEqual(len(response.read()), size) |
| 1494 | finally: |
| 1495 | conn.close() |
| 1496 | thread.join(1.0) |
| 1497 | |
| 1498 | def test_large_content_length_truncated(self): |
| 1499 | serv = socket.create_server((HOST, 0)) |
nothing calls this directly
no test coverage detected