8000 Update ``Lib/ftplib.py`` and ``Lib/test/test_ftplib`` to 3.12 by Eclips4 · Pull Request #5196 · RustPython/RustPython · GitHub
[go: up one dir, main page]
Skip to content

Update Lib/ftplib.py and Lib/test/test_ftplib to 3.12 #5196

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Mar 12, 2024
Merged

Update Lib/ftplib.py and Lib/test/test_ftplib to 3.12 #5196

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
8000
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 7 additions & 29 deletions Lib/ftplib.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -434,10 +434,7 @@ def retrbinary(self, cmd, callback, blocksize=8192, rest=None):
"""
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):
Expand Down Expand Up @@ -496,10 +493,7 @@ def storbinary(self, cmd, fp, blocksize=8192, callback=None, rest=None):
"""
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)
Expand Down Expand Up @@ -561,7 +555,7 @@ def dir(self, *args):
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:
Expand Down Expand Up @@ -713,28 +707,12 @@ class FTP_TLS(FTP):
'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,
Expand All @@ -749,7 +727,7 @@ def auth(self):
'''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')
Expand Down
22 changes: 8 additions & 14 deletions Lib/test/test_ftplib.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
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
Expand Down Expand Up @@ -992,11 +994,11 @@ def test_context(self):
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)
Expand Down Expand Up @@ -1160,18 +1162,10 @@ def test__all__(self):
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()
0