diff options
author | Facundo Batista <facundobatista@gmail.com> | 2007-05-24 17:50:54 +0000 |
---|---|---|
committer | Facundo Batista <facundobatista@gmail.com> | 2007-05-24 17:50:54 +0000 |
commit | 987465000d8c7b4b8010bec75ba32941255176f7 (patch) | |
tree | 74ea186179a59a5cb785ca390e92a87abe42736d /Lib/test/test_urllib.py | |
parent | 6c0e39249128e66fa763a6849bc0418d00cf5063 (diff) | |
download | cpython-987465000d8c7b4b8010bec75ba32941255176f7.tar.gz |
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...
Diffstat (limited to 'Lib/test/test_urllib.py')
-rw-r--r-- | Lib/test/test_urllib.py | 67 |
1 files changed, 66 insertions, 1 deletions
diff --git a/Lib/test/test_urllib.py b/Lib/test/test_urllib.py index 294ed5e06a..93e4d6040e 100644 --- a/Lib/test/test_urllib.py +++ b/Lib/test/test_urllib.py @@ -8,6 +8,10 @@ import os 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 @@ class Pathname_Tests(unittest.TestCase): "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 @@ def test_main(): QuotingTests, UnquotingTests, urlencode_Tests, - Pathname_Tests + Pathname_Tests, + FTPWrapperTests, ) |