Added an optional timeout parameter to urllib.ftpwrapper, with tests
(for this and a basic one, because there weren't any). Changed also
NEWS, but didn't find documentation for this function, assumed it
wasn't public...
diff --git a/Lib/test/test_urllib.py b/Lib/test/test_urllib.py
index 294ed5e..93e4d60 100644
--- a/Lib/test/test_urllib.py
+++ b/Lib/test/test_urllib.py
@@ -8,6 +8,10 @@
import mimetools
import tempfile
import StringIO
+import ftplib
+import threading
+import socket
+import time
def hexescape(char):
"""Escape char as RFC 2396 specifies"""
@@ -541,6 +545,66 @@
"url2pathname() failed; %s != %s" %
(expect, result))
+def server(evt):
+ serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ serv.settimeout(3)
+ serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ serv.bind(("", 9091))
+ serv.listen(5)
+ try:
+ conn, addr = serv.accept()
+ except socket.timeout:
+ pass
+ else:
+ conn.send("1 Hola mundo\n")
+ conn.recv(200)
+ conn.send("2 No more lines\n")
+ conn.close()
+ finally:
+ serv.close()
+ evt.set()
+
+class FTPWrapperTests(unittest.TestCase):
+
+ def setUp(self):
+ ftplib.FTP.port = 9091
+ self.evt = threading.Event()
+ threading.Thread(target=server, args=(self.evt,)).start()
+ time.sleep(.1)
+
+ def tearDown(self):
+ self.evt.wait()
+
+ def testBasic(self):
+ # connects
+ ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9091, [])
+ ftp.ftp.sock.close()
+
+ def testTimeoutDefault(self):
+ # default
+ ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9091, [])
+ self.assertTrue(ftp.ftp.sock.gettimeout() is None)
+ ftp.ftp.sock.close()
+
+ def testTimeoutValue(self):
+ # a value
+ ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9091, [], timeout=30)
+ self.assertEqual(ftp.ftp.sock.gettimeout(), 30)
+ ftp.ftp.sock.close()
+
+
+ def testTimeoutNone(self):
+ # None, having other default
+ previous = socket.getdefaulttimeout()
+ socket.setdefaulttimeout(30)
+ try:
+ ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9091, [], timeout=30)
+ finally:
+ socket.setdefaulttimeout(previous)
+ self.assertEqual(ftp.ftp.sock.gettimeout(), 30)
+ ftp.ftp.close()
+
+
def test_main():
@@ -551,7 +615,8 @@
QuotingTests,
UnquotingTests,
urlencode_Tests,
- Pathname_Tests
+ Pathname_Tests,
+ FTPWrapperTests,
)