(self)
| 1475 | resp.close() |
| 1476 | |
| 1477 | def test_response_fileno(self): |
| 1478 | # Make sure fd returned by fileno is valid. |
| 1479 | serv = socket.create_server((HOST, 0)) |
| 1480 | self.addCleanup(serv.close) |
| 1481 | |
| 1482 | result = None |
| 1483 | def run_server(): |
| 1484 | [conn, address] = serv.accept() |
| 1485 | with conn, conn.makefile("rb") as reader: |
| 1486 | # Read the request header until a blank line |
| 1487 | while True: |
| 1488 | line = reader.readline() |
| 1489 | if not line.rstrip(b"\r\n"): |
| 1490 | break |
| 1491 | conn.sendall(b"HTTP/1.1 200 Connection established\r\n\r\n") |
| 1492 | nonlocal result |
| 1493 | result = reader.read() |
| 1494 | |
| 1495 | thread = threading.Thread(target=run_server) |
| 1496 | thread.start() |
| 1497 | self.addCleanup(thread.join, float(1)) |
| 1498 | conn = client.HTTPConnection(*serv.getsockname()) |
| 1499 | conn.request("CONNECT", "dummy:1234") |
| 1500 | response = conn.getresponse() |
| 1501 | try: |
| 1502 | self.assertEqual(response.status, client.OK) |
| 1503 | s = socket.socket(fileno=response.fileno()) |
| 1504 | try: |
| 1505 | s.sendall(b"proxied data\n") |
| 1506 | finally: |
| 1507 | s.detach() |
| 1508 | finally: |
| 1509 | response.close() |
| 1510 | conn.close() |
| 1511 | thread.join() |
| 1512 | self.assertEqual(result, b"proxied data\n") |
| 1513 | |
| 1514 | def test_large_content_length(self): |
| 1515 | serv = socket.create_server((HOST, 0)) |
nothing calls this directly
no test coverage detected