(self)
| 1512 | self.assertEqual(result, b"proxied data\n") |
| 1513 | |
| 1514 | def test_large_content_length(self): |
| 1515 | serv = socket.create_server((HOST, 0)) |
| 1516 | self.addCleanup(serv.close) |
| 1517 | |
| 1518 | def run_server(): |
| 1519 | [conn, address] = serv.accept() |
| 1520 | with conn: |
| 1521 | while conn.recv(1024): |
| 1522 | conn.sendall( |
| 1523 | b"HTTP/1.1 200 Ok\r\n" |
| 1524 | b"Content-Length: %d\r\n" |
| 1525 | b"\r\n" % size) |
| 1526 | conn.sendall(b'A' * (size//3)) |
| 1527 | conn.sendall(b'B' * (size - size//3)) |
| 1528 | |
| 1529 | thread = threading.Thread(target=run_server) |
| 1530 | thread.start() |
| 1531 | self.addCleanup(thread.join, 1.0) |
| 1532 | |
| 1533 | conn = client.HTTPConnection(*serv.getsockname()) |
| 1534 | try: |
| 1535 | for w in range(15, 27): |
| 1536 | size = 1 << w |
| 1537 | conn.request("GET", "/") |
| 1538 | with conn.getresponse() as response: |
| 1539 | self.assertEqual(len(response.read()), size) |
| 1540 | finally: |
| 1541 | conn.close() |
| 1542 | thread.join(1.0) |
| 1543 | |
| 1544 | def test_large_content_length_truncated(self): |
| 1545 | serv = socket.create_server((HOST, 0)) |
nothing calls this directly
no test coverage detected