(self)
| 1429 | resp.close() |
| 1430 | |
| 1431 | def test_response_fileno(self): |
| 1432 | # Make sure fd returned by fileno is valid. |
| 1433 | serv = socket.create_server((HOST, 0)) |
| 1434 | self.addCleanup(serv.close) |
| 1435 | |
| 1436 | result = None |
| 1437 | def run_server(): |
| 1438 | [conn, address] = serv.accept() |
| 1439 | with conn, conn.makefile("rb") as reader: |
| 1440 | # Read the request header until a blank line |
| 1441 | while True: |
| 1442 | line = reader.readline() |
| 1443 | if not line.rstrip(b"\r\n"): |
| 1444 | break |
| 1445 | conn.sendall(b"HTTP/1.1 200 Connection established\r\n\r\n") |
| 1446 | nonlocal result |
| 1447 | result = reader.read() |
| 1448 | |
| 1449 | thread = threading.Thread(target=run_server) |
| 1450 | thread.start() |
| 1451 | self.addCleanup(thread.join, float(1)) |
| 1452 | conn = client.HTTPConnection(*serv.getsockname()) |
| 1453 | conn.request("CONNECT", "dummy:1234") |
| 1454 | response = conn.getresponse() |
| 1455 | try: |
| 1456 | self.assertEqual(response.status, client.OK) |
| 1457 | s = socket.socket(fileno=response.fileno()) |
| 1458 | try: |
| 1459 | s.sendall(b"proxied data\n") |
| 1460 | finally: |
| 1461 | s.detach() |
| 1462 | finally: |
| 1463 | response.close() |
| 1464 | conn.close() |
| 1465 | thread.join() |
| 1466 | self.assertEqual(result, b"proxied data\n") |
| 1467 | |
| 1468 | def test_large_content_length(self): |
| 1469 | serv = socket.create_server((HOST, 0)) |
nothing calls this directly
no test coverage detected