diff options
| -rw-r--r-- | Lib/test/test_httplib.py | 41 | ||||
| -rw-r--r-- | Lib/test/test_urllibnet.py | 11 |
2 files changed, 41 insertions, 11 deletions
diff --git a/Lib/test/test_httplib.py b/Lib/test/test_httplib.py index 6d2ed39..7d997b0 100644 --- a/Lib/test/test_httplib.py +++ b/Lib/test/test_httplib.py @@ -915,6 +915,47 @@ class BasicTest(TestCase): self.assertEqual(sock.file.read(), extradata) #we read to the end resp.close() + def test_response_fileno(self): + # Make sure fd returned by fileno is valid. + threading = support.import_module("threading") + + serv = socket.socket( + socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP) + self.addCleanup(serv.close) + serv.bind((HOST, 0)) + serv.listen() + + result = None + def run_server(): + [conn, address] = serv.accept() + with conn, conn.makefile("rb") as reader: + # Read the request header until a blank line + while True: + line = reader.readline() + if not line.rstrip(b"\r\n"): + break + conn.sendall(b"HTTP/1.1 200 Connection established\r\n\r\n") + nonlocal result + result = reader.read() + + thread = threading.Thread(target=run_server) + thread.start() + conn = client.HTTPConnection(*serv.getsockname()) + conn.request("CONNECT", "dummy:1234") + response = conn.getresponse() + try: + self.assertEqual(response.status, client.OK) + s = socket.socket(fileno=response.fileno()) + try: + s.sendall(b"proxied data\n") + finally: + s.detach() + finally: + response.close() + conn.close() + thread.join() + self.assertEqual(result, b"proxied data\n") + class ExtendedReadTest(TestCase): """ Test peek(), read1(), readline() diff --git a/Lib/test/test_urllibnet.py b/Lib/test/test_urllibnet.py index d5110a6..f6df54e 100644 --- a/Lib/test/test_urllibnet.py +++ b/Lib/test/test_urllibnet.py @@ -99,17 +99,6 @@ class urlopenNetworkTests(unittest.TestCase): open_url.close() self.assertEqual(code, 404) - # On Windows, socket handles are not file descriptors; this - # test can't pass on Windows. - @unittest.skipIf(sys.platform in ('win32',), 'not appropriate for Windows') - def test_fileno(self): - # Make sure fd returned by fileno is valid. - with self.urlopen("http://www.google.com/") as open_url: - fd = open_url.fileno() - with os.fdopen(fd, 'rb') as f: - self.assertTrue(f.read(), "reading from file created using fd " - "returned by fileno failed") - def test_bad_address(self): # Make sure proper exception is raised when connecting to a bogus # address. |