(self)
| 1542 | thread.join(1.0) |
| 1543 | |
| 1544 | def test_large_content_length_truncated(self): |
| 1545 | serv = socket.create_server((HOST, 0)) |
| 1546 | self.addCleanup(serv.close) |
| 1547 | |
| 1548 | def run_server(): |
| 1549 | while True: |
| 1550 | [conn, address] = serv.accept() |
| 1551 | with conn: |
| 1552 | conn.recv(1024) |
| 1553 | if not size: |
| 1554 | break |
| 1555 | conn.sendall( |
| 1556 | b"HTTP/1.1 200 Ok\r\n" |
| 1557 | b"Content-Length: %d\r\n" |
| 1558 | b"\r\n" |
| 1559 | b"Text" % size) |
| 1560 | |
| 1561 | thread = threading.Thread(target=run_server) |
| 1562 | thread.start() |
| 1563 | self.addCleanup(thread.join, 1.0) |
| 1564 | |
| 1565 | conn = client.HTTPConnection(*serv.getsockname()) |
| 1566 | try: |
| 1567 | for w in range(18, 65): |
| 1568 | size = 1 << w |
| 1569 | conn.request("GET", "/") |
| 1570 | with conn.getresponse() as response: |
| 1571 | self.assertRaises(client.IncompleteRead, response.read) |
| 1572 | conn.close() |
| 1573 | finally: |
| 1574 | conn.close() |
| 1575 | size = 0 |
| 1576 | conn.request("GET", "/") |
| 1577 | conn.close() |
| 1578 | thread.join(1.0) |
| 1579 | |
| 1580 | def test_putrequest_override_domain_validation(self): |
| 1581 | """ |
nothing calls this directly
no test coverage detected