diff options
Diffstat (limited to 'Lib/test/test_ftplib.py')
| -rw-r--r-- | Lib/test/test_ftplib.py | 54 |
1 files changed, 33 insertions, 21 deletions
diff --git a/Lib/test/test_ftplib.py b/Lib/test/test_ftplib.py index dcf1fd9..f93a621 100644 --- a/Lib/test/test_ftplib.py +++ b/Lib/test/test_ftplib.py @@ -21,6 +21,7 @@ from test import support from test.support import HOST threading = support.import_module('threading') +TIMEOUT = 3 # the dummy data returned by server over the data channel when # RETR, LIST, NLST, MLSD commands are issued RETR_DATA = 'abcde12345\r\n' * 1000 @@ -125,7 +126,7 @@ class DummyFTPHandler(asynchat.async_chat): addr = list(map(int, arg.split(','))) ip = '%d.%d.%d.%d' %tuple(addr[:4]) port = (addr[4] * 256) + addr[5] - s = socket.create_connection((ip, port), timeout=2) + s = socket.create_connection((ip, port), timeout=TIMEOUT) self.dtp = self.dtp_handler(s, baseclass=self) self.push('200 active data connection established') @@ -133,7 +134,7 @@ class DummyFTPHandler(asynchat.async_chat): with socket.socket() as sock: sock.bind((self.socket.getsockname()[0], 0)) sock.listen(5) - sock.settimeout(10) + sock.settimeout(TIMEOUT) ip, port = sock.getsockname()[:2] ip = ip.replace('.', ','); p1 = port / 256; p2 = port % 256 self.push('227 entering passive mode (%s,%d,%d)' %(ip, p1, p2)) @@ -143,7 +144,7 @@ class DummyFTPHandler(asynchat.async_chat): def cmd_eprt(self, arg): af, ip, port = arg.split(arg[0])[1:-1] port = int(port) - s = socket.create_connection((ip, port), timeout=2) + s = socket.create_connection((ip, port), timeout=TIMEOUT) self.dtp = self.dtp_handler(s, baseclass=self) self.push('200 active data connection established') @@ -151,7 +152,7 @@ class DummyFTPHandler(asynchat.async_chat): with socket.socket(socket.AF_INET6) as sock: sock.bind((self.socket.getsockname()[0], 0)) sock.listen(5) - sock.settimeout(10) + sock.settimeout(TIMEOUT) port = sock.getsockname()[1] self.push('229 entering extended passive mode (|||%d|)' %port) conn, addr = sock.accept() @@ -321,7 +322,7 @@ if ssl is not None: elif err.args[0] == ssl.SSL_ERROR_EOF: return self.handle_close() raise - except socket.error as err: + except OSError as err: if err.args[0] == errno.ECONNABORTED: return self.handle_close() else: @@ -335,7 +336,7 @@ if ssl is not None: if err.args[0] in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE): return - except socket.error as err: + except OSError as err: # Any "socket error" corresponds to a SSL_ERROR_SYSCALL return # from OpenSSL's SSL_shutdown(), corresponding to a # closed socket condition. See also: @@ -454,7 +455,7 @@ class TestFTPClass(TestCase): def setUp(self): self.server = DummyFTPServer((HOST, 0)) self.server.start() - self.client = ftplib.FTP(timeout=2) + self.client = ftplib.FTP(timeout=TIMEOUT) self.client.connect(self.server.host, self.server.port) def tearDown(self): @@ -482,7 +483,7 @@ class TestFTPClass(TestCase): def test_all_errors(self): exceptions = (ftplib.error_reply, ftplib.error_temp, ftplib.error_perm, - ftplib.error_proto, ftplib.Error, IOError, EOFError) + ftplib.error_proto, ftplib.Error, OSError, EOFError) for x in exceptions: try: raise x('exception not included in all_errors set') @@ -668,7 +669,7 @@ class TestFTPClass(TestCase): def test_makepasv(self): host, port = self.client.makepasv() - conn = socket.create_connection((host, port), 10) + conn = socket.create_connection((host, port), timeout=TIMEOUT) conn.close() # IPv4 is in use, just make sure send_epsv has not been used self.assertEqual(self.server.handler_instance.last_received_cmd, 'pasv') @@ -681,12 +682,12 @@ class TestFTPClass(TestCase): return False try: self.client.sendcmd('noop') - except (socket.error, EOFError): + except (OSError, EOFError): return False return True # base test - with ftplib.FTP(timeout=2) as self.client: + with ftplib.FTP(timeout=TIMEOUT) as self.client: self.client.connect(self.server.host, self.server.port) self.client.sendcmd('noop') self.assertTrue(is_client_connected()) @@ -694,7 +695,7 @@ class TestFTPClass(TestCase): self.assertFalse(is_client_connected()) # QUIT sent inside the with block - with ftplib.FTP(timeout=2) as self.client: + with ftplib.FTP(timeout=TIMEOUT) as self.client: self.client.connect(self.server.host, self.server.port) self.client.sendcmd('noop') self.client.quit() @@ -704,7 +705,7 @@ class TestFTPClass(TestCase): # force a wrong response code to be sent on QUIT: error_perm # is expected and the connection is supposed to be closed try: - with ftplib.FTP(timeout=2) as self.client: + with ftplib.FTP(timeout=TIMEOUT) as self.client: self.client.connect(self.server.host, self.server.port) self.client.sendcmd('noop') self.server.handler_instance.next_response = '550 error on quit' @@ -726,7 +727,7 @@ class TestFTPClass(TestCase): source_address=(HOST, port)) self.assertEqual(self.client.sock.getsockname()[1], port) self.client.quit() - except IOError as e: + except OSError as e: if e.errno == errno.EADDRINUSE: self.skipTest("couldn't bind to port %d" % port) raise @@ -737,7 +738,7 @@ class TestFTPClass(TestCase): try: with self.client.transfercmd('list') as sock: self.assertEqual(sock.getsockname()[1], port) - except IOError as e: + except OSError as e: if e.errno == errno.EADDRINUSE: self.skipTest("couldn't bind to port %d" % port) raise @@ -760,7 +761,7 @@ class TestIPv6Environment(TestCase): def setUp(self): self.server = DummyFTPServer(('::1', 0), af=socket.AF_INET6) self.server.start() - self.client = ftplib.FTP() + self.client = ftplib.FTP(timeout=TIMEOUT) self.client.connect(self.server.host, self.server.port) def tearDown(self): @@ -777,7 +778,7 @@ class TestIPv6Environment(TestCase): def test_makepasv(self): host, port = self.client.makepasv() - conn = socket.create_connection((host, port), 10) + conn = socket.create_connection((host, port), timeout=TIMEOUT) conn.close() self.assertEqual(self.server.handler_instance.last_received_cmd, 'epsv') @@ -803,7 +804,7 @@ class TestTLS_FTPClassMixin(TestFTPClass): def setUp(self): self.server = DummyTLS_FTPServer((HOST, 0)) self.server.start() - self.client = ftplib.FTP_TLS(timeout=2) + self.client = ftplib.FTP_TLS(timeout=TIMEOUT) self.client.connect(self.server.host, self.server.port) # enable TLS self.client.auth() @@ -816,7 +817,7 @@ class TestTLS_FTPClass(TestCase): def setUp(self): self.server = DummyTLS_FTPServer((HOST, 0)) self.server.start() - self.client = ftplib.FTP_TLS(timeout=2) + self.client = ftplib.FTP_TLS(timeout=TIMEOUT) self.client.connect(self.server.host, self.server.port) def tearDown(self): @@ -876,7 +877,7 @@ class TestTLS_FTPClass(TestCase): self.assertRaises(ValueError, ftplib.FTP_TLS, certfile=CERTFILE, keyfile=CERTFILE, context=ctx) - self.client = ftplib.FTP_TLS(context=ctx, timeout=2) + self.client = ftplib.FTP_TLS(context=ctx, timeout=TIMEOUT) self.client.connect(self.server.host, self.server.port) self.assertNotIsInstance(self.client.sock, ssl.SSLSocket) self.client.auth() @@ -990,8 +991,19 @@ class TestTimeouts(TestCase): ftp.close() +class TestNetrcDeprecation(TestCase): + + def test_deprecation(self): + with support.temp_cwd(), support.EnvironmentVarGuard() as env: + env['HOME'] = os.getcwd() + open('.netrc', 'w').close() + with self.assertWarns(DeprecationWarning): + ftplib.Netrc() + + + def test_main(): - tests = [TestFTPClass, TestTimeouts] + tests = [TestFTPClass, TestTimeouts, TestNetrcDeprecation] if support.IPV6_ENABLED: tests.append(TestIPv6Environment) |