Update ftplib and test_ftplib to 3.12 (#5196)

This commit is contained in:
Kirill Podoprigora
2024-03-13 00:35:16 +02:00
committed by GitHub
parent 4e7b3bc8f2
commit 855fa1411f
2 changed files with 15 additions and 43 deletions

36
Lib/ftplib.py vendored
View File

@@ -434,10 +434,7 @@ class FTP:
"""
self.voidcmd('TYPE I')
with self.transfercmd(cmd, rest) as conn:
while 1:
data = conn.recv(blocksize)
if not data:
break
while data := conn.recv(blocksize):
callback(data)
# shutdown ssl layer
if _SSLSocket is not None and isinstance(conn, _SSLSocket):
@@ -496,10 +493,7 @@ class FTP:
"""
self.voidcmd('TYPE I')
with self.transfercmd(cmd, rest) as conn:
while 1:
buf = fp.read(blocksize)
if not buf:
break
while buf := fp.read(blocksize):
conn.sendall(buf)
if callback:
callback(buf)
@@ -561,7 +555,7 @@ class FTP:
LIST command. (This *should* only be used for a pathname.)'''
cmd = 'LIST'
func = None
if args[-1:] and type(args[-1]) != type(''):
if args[-1:] and not isinstance(args[-1], str):
args, func = args[:-1], args[-1]
for arg in args:
if arg:
@@ -713,28 +707,12 @@ else:
'221 Goodbye.'
>>>
'''
ssl_version = ssl.PROTOCOL_TLS_CLIENT
def __init__(self, host='', user='', passwd='', acct='',
keyfile=None, certfile=None, context=None,
timeout=_GLOBAL_DEFAULT_TIMEOUT, source_address=None, *,
encoding='utf-8'):
if context is not None and keyfile is not None:
raise ValueError("context and keyfile arguments are mutually "
"exclusive")
if context is not None and certfile is not None:
raise ValueError("context and certfile arguments are mutually "
"exclusive")
if keyfile is not None or certfile is not None:
import warnings
warnings.warn("keyfile and certfile are deprecated, use a "
"custom context instead", DeprecationWarning, 2)
self.keyfile = keyfile
self.certfile = certfile
*, context=None, timeout=_GLOBAL_DEFAULT_TIMEOUT,
source_address=None, encoding='utf-8'):
if context is None:
context = ssl._create_stdlib_context(self.ssl_version,
certfile=certfile,
keyfile=keyfile)
context = ssl._create_stdlib_context()
self.context = context
self._prot_p = False
super().__init__(host, user, passwd, acct,
@@ -749,7 +727,7 @@ else:
'''Set up secure control connection by using TLS/SSL.'''
if isinstance(self.sock, ssl.SSLSocket):
raise ValueError("Already using TLS")
if self.ssl_version >= ssl.PROTOCOL_TLS:
if self.context.protocol >= ssl.PROTOCOL_TLS:
resp = self.voidcmd('AUTH TLS')
else:
resp = self.voidcmd('AUTH SSL')

View File

@@ -21,6 +21,8 @@ from test import support
from test.support import threading_helper
from test.support import socket_helper
from test.support import warnings_helper
from test.support import asynchat
from test.support import asyncore
from test.support.socket_helper import HOST, HOSTv6
import sys
@@ -992,11 +994,11 @@ class TestTLS_FTPClass(TestCase):
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
self.assertRaises(ValueError, ftplib.FTP_TLS, keyfile=CERTFILE,
self.assertRaises(TypeError, ftplib.FTP_TLS, keyfile=CERTFILE,
context=ctx)
self.assertRaises(ValueError, ftplib.FTP_TLS, certfile=CERTFILE,
self.assertRaises(TypeError, ftplib.FTP_TLS, certfile=CERTFILE,
context=ctx)
self.assertRaises(ValueError, ftplib.FTP_TLS, certfile=CERTFILE,
self.assertRaises(TypeError, ftplib.FTP_TLS, certfile=CERTFILE,
keyfile=CERTFILE, context=ctx)
self.client = ftplib.FTP_TLS(context=ctx, timeout=TIMEOUT)
@@ -1160,18 +1162,10 @@ class MiscTestCase(TestCase):
support.check__all__(self, ftplib, not_exported=not_exported)
def test_main():
tests = [TestFTPClass, TestTimeouts,
TestIPv6Environment,
TestTLS_FTPClassMixin, TestTLS_FTPClass,
MiscTestCase]
def setUpModule():
thread_info = threading_helper.threading_setup()
try:
support.run_unittest(*tests)
finally:
threading_helper.threading_cleanup(*thread_info)
unittest.addModuleCleanup(threading_helper.threading_cleanup, *thread_info)
if __name__ == '__main__':
test_main()
unittest.main()