From 8578114a5e2202a3bdb8b006cc9c81f8591ab7be Mon Sep 17 00:00:00 2001 From: CPython Developers <> Date: Sat, 14 Feb 2026 13:59:36 +0000 Subject: [PATCH] Update shutil from v3.14.3 --- Lib/shutil.py | 224 +++++++++++++++++++++++++++------------- Lib/test/test_shutil.py | 110 ++++++++++++-------- 2 files changed, 223 insertions(+), 111 deletions(-) diff --git a/Lib/shutil.py b/Lib/shutil.py index 7df972012..8d8fe1455 100644 --- a/Lib/shutil.py +++ b/Lib/shutil.py @@ -32,6 +32,13 @@ try: except ImportError: _LZMA_SUPPORTED = False +try: + from compression import zstd + del zstd + _ZSTD_SUPPORTED = True +except ImportError: + _ZSTD_SUPPORTED = False + _WINDOWS = os.name == 'nt' posix = nt = None if os.name == 'posix': @@ -44,11 +51,12 @@ if sys.platform == 'win32': else: _winapi = None -COPY_BUFSIZE = 1024 * 1024 if _WINDOWS else 64 * 1024 +COPY_BUFSIZE = 1024 * 1024 if _WINDOWS else 256 * 1024 # This should never be removed, see rationale in: # https://bugs.python.org/issue43743#msg393429 _USE_CP_SENDFILE = (hasattr(os, "sendfile") - and sys.platform.startswith(("linux", "android"))) + and sys.platform.startswith(("linux", "android", "sunos"))) +_USE_CP_COPY_FILE_RANGE = hasattr(os, "copy_file_range") _HAS_FCOPYFILE = posix and hasattr(posix, "_fcopyfile") # macOS # CMD defaults in Windows 10 @@ -56,7 +64,7 @@ _WIN_DEFAULT_PATHEXT = ".COM;.EXE;.BAT;.CMD;.VBS;.JS;.WS;.MSC" __all__ = ["copyfileobj", "copyfile", "copymode", "copystat", "copy", "copy2", "copytree", "move", "rmtree", "Error", "SpecialFileError", - "ExecError", "make_archive", "get_archive_formats", + "make_archive", "get_archive_formats", "register_archive_format", "unregister_archive_format", "get_unpack_formats", "register_unpack_format", "unregister_unpack_format", "unpack_archive", @@ -74,8 +82,6 @@ class SpecialFileError(OSError): """Raised when trying to do a kind of operation (e.g. copying) which is not supported on a special file (e.g. a named pipe)""" -class ExecError(OSError): - """Raised when a command could not be executed""" class ReadError(OSError): """Raised when an archive cannot be read""" @@ -109,10 +115,70 @@ def _fastcopy_fcopyfile(fsrc, fdst, flags): else: raise err from None +def _determine_linux_fastcopy_blocksize(infd): + """Determine blocksize for fastcopying on Linux. + + Hopefully the whole file will be copied in a single call. + The copying itself should be performed in a loop 'till EOF is + reached (0 return) so a blocksize smaller or bigger than the actual + file size should not make any difference, also in case the file + content changes while being copied. + """ + try: + blocksize = max(os.fstat(infd).st_size, 2 ** 23) # min 8 MiB + except OSError: + blocksize = 2 ** 27 # 128 MiB + # On 32-bit architectures truncate to 1 GiB to avoid OverflowError, + # see gh-82500. + if sys.maxsize < 2 ** 32: + blocksize = min(blocksize, 2 ** 30) + return blocksize + +def _fastcopy_copy_file_range(fsrc, fdst): + """Copy data from one regular mmap-like fd to another by using + a high-performance copy_file_range(2) syscall that gives filesystems + an opportunity to implement the use of reflinks or server-side copy. + + This should work on Linux >= 4.5 only. + """ + try: + infd = fsrc.fileno() + outfd = fdst.fileno() + except Exception as err: + raise _GiveupOnFastCopy(err) # not a regular file + + blocksize = _determine_linux_fastcopy_blocksize(infd) + offset = 0 + while True: + try: + n_copied = os.copy_file_range(infd, outfd, blocksize, offset_dst=offset) + except OSError as err: + # ...in oder to have a more informative exception. + err.filename = fsrc.name + err.filename2 = fdst.name + + if err.errno == errno.ENOSPC: # filesystem is full + raise err from None + + # Give up on first call and if no data was copied. + if offset == 0 and os.lseek(outfd, 0, os.SEEK_CUR) == 0: + raise _GiveupOnFastCopy(err) + + raise err + else: + if n_copied == 0: + # If no bytes have been copied yet, copy_file_range + # might silently fail. + # https://lore.kernel.org/linux-fsdevel/20210126233840.GG4626@dread.disaster.area/T/#m05753578c7f7882f6e9ffe01f981bc223edef2b0 + if offset == 0: + raise _GiveupOnFastCopy() + break + offset += n_copied + def _fastcopy_sendfile(fsrc, fdst): """Copy data from one regular mmap-like fd to another by using high-performance sendfile(2) syscall. - This should work on Linux >= 2.6.33 only. + This should work on Linux >= 2.6.33, Android and Solaris. """ # Note: copyfileobj() is left alone in order to not introduce any # unexpected breakage. Possible risks by using zero-copy calls @@ -130,26 +196,13 @@ def _fastcopy_sendfile(fsrc, fdst): except Exception as err: raise _GiveupOnFastCopy(err) # not a regular file - # Hopefully the whole file will be copied in a single call. - # sendfile() is called in a loop 'till EOF is reached (0 return) - # so a bufsize smaller or bigger than the actual file size - # should not make any difference, also in case the file content - # changes while being copied. - try: - blocksize = max(os.fstat(infd).st_size, 2 ** 23) # min 8MiB - except OSError: - blocksize = 2 ** 27 # 128MiB - # On 32-bit architectures truncate to 1GiB to avoid OverflowError, - # see bpo-38319. - if sys.maxsize < 2 ** 32: - blocksize = min(blocksize, 2 ** 30) - + blocksize = _determine_linux_fastcopy_blocksize(infd) offset = 0 while True: try: sent = os.sendfile(outfd, infd, offset, blocksize) except OSError as err: - # ...in oder to have a more informative exception. + # ...in order to have a more informative exception. err.filename = fsrc.name err.filename2 = fdst.name @@ -267,13 +320,21 @@ def copyfile(src, dst, *, follow_symlinks=True): return dst except _GiveupOnFastCopy: pass - # Linux - elif _USE_CP_SENDFILE: - try: - _fastcopy_sendfile(fsrc, fdst) - return dst - except _GiveupOnFastCopy: - pass + # Linux / Android / Solaris + elif _USE_CP_SENDFILE or _USE_CP_COPY_FILE_RANGE: + # reflink may be implicit in copy_file_range. + if _USE_CP_COPY_FILE_RANGE: + try: + _fastcopy_copy_file_range(fsrc, fdst) + return dst + except _GiveupOnFastCopy: + pass + if _USE_CP_SENDFILE: + try: + _fastcopy_sendfile(fsrc, fdst) + return dst + except _GiveupOnFastCopy: + pass # Windows, see: # https://github.com/python/cpython/pull/7160#discussion_r195405230 elif _WINDOWS and file_size > 0: @@ -605,7 +666,22 @@ else: return stat.S_ISLNK(st.st_mode) # version vulnerable to race conditions -def _rmtree_unsafe(path, onexc): +def _rmtree_unsafe(path, dir_fd, onexc): + if dir_fd is not None: + raise NotImplementedError("dir_fd unavailable on this platform") + try: + st = os.lstat(path) + except OSError as err: + onexc(os.lstat, path, err) + return + try: + if _rmtree_islink(st): + # symlinks to directories are forbidden, see bug #1669 + raise OSError("Cannot call rmtree on a symbolic link") + except OSError as err: + onexc(os.path.islink, path, err) + # can't continue even if onexc hook returns + return def onerror(err): if not isinstance(err, FileNotFoundError): onexc(os.scandir, err.filename, err) @@ -635,7 +711,26 @@ def _rmtree_unsafe(path, onexc): onexc(os.rmdir, path, err) # Version using fd-based APIs to protect against races -def _rmtree_safe_fd(stack, onexc): +def _rmtree_safe_fd(path, dir_fd, onexc): + # While the unsafe rmtree works fine on bytes, the fd based does not. + if isinstance(path, bytes): + path = os.fsdecode(path) + stack = [(os.lstat, dir_fd, path, None)] + try: + while stack: + _rmtree_safe_fd_step(stack, onexc) + finally: + # Close any file descriptors still on the stack. + while stack: + func, fd, path, entry = stack.pop() + if func is not os.close: + continue + try: + os.close(fd) + except OSError as err: + onexc(os.close, path, err) + +def _rmtree_safe_fd_step(stack, onexc): # Each stack item has four elements: # * func: The first operation to perform: os.lstat, os.close or os.rmdir. # Walking a directory starts with an os.lstat() to detect symlinks; in @@ -710,6 +805,7 @@ _use_fd_functions = ({os.open, os.stat, os.unlink, os.rmdir} <= os.supports_dir_fd and os.scandir in os.supports_fd and os.stat in os.supports_follow_symlinks) +_rmtree_impl = _rmtree_safe_fd if _use_fd_functions else _rmtree_unsafe def rmtree(path, ignore_errors=False, onerror=None, *, onexc=None, dir_fd=None): """Recursively delete a directory tree. @@ -753,41 +849,7 @@ def rmtree(path, ignore_errors=False, onerror=None, *, onexc=None, dir_fd=None): exc_info = type(exc), exc, exc.__traceback__ return onerror(func, path, exc_info) - if _use_fd_functions: - # While the unsafe rmtree works fine on bytes, the fd based does not. - if isinstance(path, bytes): - path = os.fsdecode(path) - stack = [(os.lstat, dir_fd, path, None)] - try: - while stack: - _rmtree_safe_fd(stack, onexc) - finally: - # Close any file descriptors still on the stack. - while stack: - func, fd, path, entry = stack.pop() - if func is not os.close: - continue - try: - os.close(fd) - except OSError as err: - onexc(os.close, path, err) - else: - if dir_fd is not None: - raise NotImplementedError("dir_fd unavailable on this platform") - try: - st = os.lstat(path) - except OSError as err: - onexc(os.lstat, path, err) - return - try: - if _rmtree_islink(st): - # symlinks to directories are forbidden, see bug #1669 - raise OSError("Cannot call rmtree on a symbolic link") - except OSError as err: - onexc(os.path.islink, path, err) - # can't continue even if onexc hook returns - return - return _rmtree_unsafe(path, onexc) + _rmtree_impl(path, dir_fd, onexc) # Allow introspection of whether or not the hardening against symlink # attacks is supported on the current platform @@ -932,14 +994,14 @@ def _make_tarball(base_name, base_dir, compress="gzip", verbose=0, dry_run=0, """Create a (possibly compressed) tar file from all the files under 'base_dir'. - 'compress' must be "gzip" (the default), "bzip2", "xz", or None. + 'compress' must be "gzip" (the default), "bzip2", "xz", "zst", or None. 'owner' and 'group' can be used to define an owner and a group for the archive that is being built. If not provided, the current owner and group will be used. The output tar file will be named 'base_name' + ".tar", possibly plus - the appropriate compression extension (".gz", ".bz2", or ".xz"). + the appropriate compression extension (".gz", ".bz2", ".xz", or ".zst"). Returns the output filename. """ @@ -951,6 +1013,8 @@ def _make_tarball(base_name, base_dir, compress="gzip", verbose=0, dry_run=0, tar_compression = 'bz2' elif _LZMA_SUPPORTED and compress == 'xz': tar_compression = 'xz' + elif _ZSTD_SUPPORTED and compress == 'zst': + tar_compression = 'zst' else: raise ValueError("bad value for 'compress', or compression format not " "supported : {0}".format(compress)) @@ -1079,6 +1143,10 @@ if _LZMA_SUPPORTED: _ARCHIVE_FORMATS['xztar'] = (_make_tarball, [('compress', 'xz')], "xz'ed tar-file") +if _ZSTD_SUPPORTED: + _ARCHIVE_FORMATS['zstdtar'] = (_make_tarball, [('compress', 'zst')], + "zstd'ed tar-file") + def get_archive_formats(): """Returns a list of supported formats for archiving and unarchiving. @@ -1119,7 +1187,7 @@ def make_archive(base_name, format, root_dir=None, base_dir=None, verbose=0, 'base_name' is the name of the file to create, minus any format-specific extension; 'format' is the archive format: one of "zip", "tar", "gztar", - "bztar", or "xztar". Or any other registered format. + "bztar", "xztar", or "zstdtar". Or any other registered format. 'root_dir' is a directory that will be the root directory of the archive; ie. we typically chdir into 'root_dir' before creating the @@ -1269,7 +1337,7 @@ def _unpack_zipfile(filename, extract_dir): zip.close() def _unpack_tarfile(filename, extract_dir, *, filter=None): - """Unpack tar/tar.gz/tar.bz2/tar.xz `filename` to `extract_dir` + """Unpack tar/tar.gz/tar.bz2/tar.xz/tar.zst `filename` to `extract_dir` """ import tarfile # late import for breaking circular dependency try: @@ -1304,6 +1372,10 @@ if _LZMA_SUPPORTED: _UNPACK_FORMATS['xztar'] = (['.tar.xz', '.txz'], _unpack_tarfile, [], "xz'ed tar-file") +if _ZSTD_SUPPORTED: + _UNPACK_FORMATS['zstdtar'] = (['.tar.zst', '.tzst'], _unpack_tarfile, [], + "zstd'ed tar-file") + def _find_unpack_format(filename): for name, info in _UNPACK_FORMATS.items(): for extension in info[0]: @@ -1320,7 +1392,7 @@ def unpack_archive(filename, extract_dir=None, format=None, *, filter=None): is unpacked. If not provided, the current working directory is used. `format` is the archive format: one of "zip", "tar", "gztar", "bztar", - or "xztar". Or any other registered format. If not provided, + "xztar", or "zstdtar". Or any other registered format. If not provided, unpack_archive will use the filename extension and see if an unpacker was registered for that extension. @@ -1581,3 +1653,15 @@ def which(cmd, mode=os.F_OK | os.X_OK, path=None): if _access_check(name, mode): return name return None + +def __getattr__(name): + if name == "ExecError": + import warnings + warnings._deprecated( + "shutil.ExecError", + f"{warnings._DEPRECATED_MSG}; it " + "isn't raised by any shutil function.", + remove=(3, 16) + ) + return RuntimeError + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/Lib/test/test_shutil.py b/Lib/test/test_shutil.py index 0e9732556..62c80aab4 100644 --- a/Lib/test/test_shutil.py +++ b/Lib/test/test_shutil.py @@ -10,7 +10,6 @@ import os import os.path import errno import functools -import pathlib import subprocess import random import string @@ -32,7 +31,6 @@ except ImportError: from test import support from test.support import os_helper from test.support.os_helper import TESTFN, FakePath -from test.support import warnings_helper TESTFN2 = TESTFN + "2" TESTFN_SRC = TESTFN + "_SRC" @@ -429,12 +427,12 @@ class TestRmTree(BaseTest, unittest.TestCase): else: self.assertIs(func, os.listdir) self.assertIn(arg, [TESTFN, self.child_dir_path]) - self.assertTrue(issubclass(exc[0], OSError)) + self.assertIsSubclass(exc[0], OSError) self.errorState += 1 else: self.assertEqual(func, os.rmdir) self.assertEqual(arg, TESTFN) - self.assertTrue(issubclass(exc[0], OSError)) + self.assertIsSubclass(exc[0], OSError) self.errorState = 3 @unittest.skipIf(sys.platform[:6] == 'cygwin', @@ -557,25 +555,23 @@ class TestRmTree(BaseTest, unittest.TestCase): os.listdir in os.supports_fd and os.stat in os.supports_follow_symlinks) if _use_fd_functions: - self.assertTrue(shutil._use_fd_functions) self.assertTrue(shutil.rmtree.avoids_symlink_attacks) tmp_dir = self.mkdtemp() d = os.path.join(tmp_dir, 'a') os.mkdir(d) try: - real_rmtree = shutil._rmtree_safe_fd + real_open = os.open class Called(Exception): pass def _raiser(*args, **kwargs): raise Called - shutil._rmtree_safe_fd = _raiser + os.open = _raiser self.assertRaises(Called, shutil.rmtree, d) finally: - shutil._rmtree_safe_fd = real_rmtree + os.open = real_open else: - self.assertFalse(shutil._use_fd_functions) self.assertFalse(shutil.rmtree.avoids_symlink_attacks) - @unittest.skipUnless(shutil._use_fd_functions, "requires safe rmtree") + @unittest.skipUnless(shutil.rmtree.avoids_symlink_attacks, "requires safe rmtree") def test_rmtree_fails_on_close(self): # Test that the error handler is called for failed os.close() and that # os.close() is only called once for a file descriptor. @@ -610,7 +606,7 @@ class TestRmTree(BaseTest, unittest.TestCase): self.assertEqual(errors[1][1], dir1) self.assertEqual(close_count, 2) - @unittest.skipUnless(shutil._use_fd_functions, "dir_fd is not supported") + @unittest.skipUnless(shutil.rmtree.avoids_symlink_attacks, "dir_fd is not supported") def test_rmtree_with_dir_fd(self): tmp_dir = self.mkdtemp() victim = 'killme' @@ -624,7 +620,7 @@ class TestRmTree(BaseTest, unittest.TestCase): shutil.rmtree(victim, dir_fd=dir_fd) self.assertFalse(os.path.exists(fullname)) - @unittest.skipIf(shutil._use_fd_functions, "dir_fd is supported") + @unittest.skipIf(shutil.rmtree.avoids_symlink_attacks, "dir_fd is supported") def test_rmtree_with_dir_fd_unsupported(self): tmp_dir = self.mkdtemp() with self.assertRaises(NotImplementedError): @@ -906,10 +902,10 @@ class TestCopyTree(BaseTest, unittest.TestCase): os.mkdir(os.path.join(src_dir, 'test_dir', 'subdir')) create_file((src_dir, 'test_dir', 'subdir', 'test.txt'), '456') - invokations = [] + invocations = [] def _ignore(src, names): - invokations.append(src) + invocations.append(src) self.assertIsInstance(src, str) self.assertIsInstance(names, list) self.assertEqual(len(names), len(set(names))) @@ -934,7 +930,7 @@ class TestCopyTree(BaseTest, unittest.TestCase): self.assertTrue(exists(join(dst_dir, 'test_dir', 'subdir', 'test.txt'))) - self.assertEqual(len(invokations), 9) + self.assertEqual(len(invocations), 9) def test_copytree_retains_permissions(self): tmp_dir = self.mkdtemp() @@ -2145,9 +2141,6 @@ class TestArchives(BaseTest, unittest.TestCase): def check_unpack_tarball(self, format): self.check_unpack_archive(format, filter='fully_trusted') self.check_unpack_archive(format, filter='data') - with warnings_helper.check_warnings( - ('Python 3.14', DeprecationWarning)): - self.check_unpack_archive(format) def test_unpack_archive_tar(self): self.check_unpack_tarball('tar') @@ -2160,6 +2153,10 @@ class TestArchives(BaseTest, unittest.TestCase): def test_unpack_archive_bztar(self): self.check_unpack_tarball('bztar') + @support.requires_zstd() + def test_unpack_archive_zstdtar(self): + self.check_unpack_tarball('zstdtar') + @support.requires_lzma() @unittest.skipIf(AIX and not _maxdataOK(), "AIX MAXDATA must be 0x20000000 or larger") def test_unpack_archive_xztar(self): @@ -3244,12 +3241,8 @@ class _ZeroCopyFileTest(object): self.assertRaises(OSError, self.zerocopy_fun, src, dst) -@unittest.skipIf(not SUPPORTS_SENDFILE, 'os.sendfile() not supported') -class TestZeroCopySendfile(_ZeroCopyFileTest, unittest.TestCase): - PATCHPOINT = "os.sendfile" - - def zerocopy_fun(self, fsrc, fdst): - return shutil._fastcopy_sendfile(fsrc, fdst) +class _ZeroCopyFileLinuxTest(_ZeroCopyFileTest): + BLOCKSIZE_INDEX = None def test_non_regular_file_src(self): with io.BytesIO(self.FILEDATA) as src: @@ -3270,65 +3263,65 @@ class TestZeroCopySendfile(_ZeroCopyFileTest, unittest.TestCase): self.assertEqual(dst.read(), self.FILEDATA) def test_exception_on_second_call(self): - def sendfile(*args, **kwargs): + def syscall(*args, **kwargs): if not flag: flag.append(None) - return orig_sendfile(*args, **kwargs) + return orig_syscall(*args, **kwargs) else: raise OSError(errno.EBADF, "yo") flag = [] - orig_sendfile = os.sendfile - with unittest.mock.patch('os.sendfile', create=True, - side_effect=sendfile): + orig_syscall = eval(self.PATCHPOINT) + with unittest.mock.patch(self.PATCHPOINT, create=True, + side_effect=syscall): with self.get_files() as (src, dst): with self.assertRaises(OSError) as cm: - shutil._fastcopy_sendfile(src, dst) + self.zerocopy_fun(src, dst) assert flag self.assertEqual(cm.exception.errno, errno.EBADF) def test_cant_get_size(self): # Emulate a case where src file size cannot be determined. # Internally bufsize will be set to a small value and - # sendfile() will be called repeatedly. + # a system call will be called repeatedly. with unittest.mock.patch('os.fstat', side_effect=OSError) as m: with self.get_files() as (src, dst): - shutil._fastcopy_sendfile(src, dst) + self.zerocopy_fun(src, dst) assert m.called self.assertEqual(read_file(TESTFN2, binary=True), self.FILEDATA) def test_small_chunks(self): # Force internal file size detection to be smaller than the - # actual file size. We want to force sendfile() to be called + # actual file size. We want to force a system call to be called # multiple times, also in order to emulate a src fd which gets # bigger while it is being copied. mock = unittest.mock.Mock() mock.st_size = 65536 + 1 with unittest.mock.patch('os.fstat', return_value=mock) as m: with self.get_files() as (src, dst): - shutil._fastcopy_sendfile(src, dst) + self.zerocopy_fun(src, dst) assert m.called self.assertEqual(read_file(TESTFN2, binary=True), self.FILEDATA) def test_big_chunk(self): # Force internal file size detection to be +100MB bigger than - # the actual file size. Make sure sendfile() does not rely on + # the actual file size. Make sure a system call does not rely on # file size value except for (maybe) a better throughput / # performance. mock = unittest.mock.Mock() mock.st_size = self.FILESIZE + (100 * 1024 * 1024) with unittest.mock.patch('os.fstat', return_value=mock) as m: with self.get_files() as (src, dst): - shutil._fastcopy_sendfile(src, dst) + self.zerocopy_fun(src, dst) assert m.called self.assertEqual(read_file(TESTFN2, binary=True), self.FILEDATA) def test_blocksize_arg(self): - with unittest.mock.patch('os.sendfile', + with unittest.mock.patch(self.PATCHPOINT, side_effect=ZeroDivisionError) as m: self.assertRaises(ZeroDivisionError, shutil.copyfile, TESTFN, TESTFN2) - blocksize = m.call_args[0][3] + blocksize = m.call_args[0][self.BLOCKSIZE_INDEX] # Make sure file size and the block size arg passed to # sendfile() are the same. self.assertEqual(blocksize, os.path.getsize(TESTFN)) @@ -3338,9 +3331,19 @@ class TestZeroCopySendfile(_ZeroCopyFileTest, unittest.TestCase): self.addCleanup(os_helper.unlink, TESTFN2 + '3') self.assertRaises(ZeroDivisionError, shutil.copyfile, TESTFN2, TESTFN2 + '3') - blocksize = m.call_args[0][3] + blocksize = m.call_args[0][self.BLOCKSIZE_INDEX] self.assertEqual(blocksize, 2 ** 23) + +@unittest.skipIf(not SUPPORTS_SENDFILE, 'os.sendfile() not supported') +@unittest.mock.patch.object(shutil, "_USE_CP_COPY_FILE_RANGE", False) +class TestZeroCopySendfile(_ZeroCopyFileLinuxTest, unittest.TestCase): + PATCHPOINT = "os.sendfile" + BLOCKSIZE_INDEX = 3 + + def zerocopy_fun(self, fsrc, fdst): + return shutil._fastcopy_sendfile(fsrc, fdst) + def test_file2file_not_supported(self): # Emulate a case where sendfile() only support file->socket # fds. In such a case copyfile() is supposed to skip the @@ -3363,6 +3366,29 @@ class TestZeroCopySendfile(_ZeroCopyFileTest, unittest.TestCase): shutil._USE_CP_SENDFILE = True +@unittest.skipUnless(shutil._USE_CP_COPY_FILE_RANGE, "os.copy_file_range() not supported") +class TestZeroCopyCopyFileRange(_ZeroCopyFileLinuxTest, unittest.TestCase): + PATCHPOINT = "os.copy_file_range" + BLOCKSIZE_INDEX = 2 + + def zerocopy_fun(self, fsrc, fdst): + return shutil._fastcopy_copy_file_range(fsrc, fdst) + + def test_empty_file(self): + srcname = f"{TESTFN}src" + dstname = f"{TESTFN}dst" + self.addCleanup(lambda: os_helper.unlink(srcname)) + self.addCleanup(lambda: os_helper.unlink(dstname)) + with open(srcname, "wb"): + pass + + with open(srcname, "rb") as src, open(dstname, "wb") as dst: + # _fastcopy_copy_file_range gives up copying empty files due + # to a bug in older Linux. + with self.assertRaises(shutil._GiveupOnFastCopy): + self.zerocopy_fun(src, dst) + + @unittest.skipIf(not MACOS, 'macOS only') class TestZeroCopyMACOS(_ZeroCopyFileTest, unittest.TestCase): PATCHPOINT = "posix._fcopyfile" @@ -3453,10 +3479,10 @@ class PublicAPITests(unittest.TestCase): """Ensures that the correct values are exposed in the public API.""" def test_module_all_attribute(self): - self.assertTrue(hasattr(shutil, '__all__')) + self.assertHasAttr(shutil, '__all__') target_api = ['copyfileobj', 'copyfile', 'copymode', 'copystat', 'copy', 'copy2', 'copytree', 'move', 'rmtree', 'Error', - 'SpecialFileError', 'ExecError', 'make_archive', + 'SpecialFileError', 'make_archive', 'get_archive_formats', 'register_archive_format', 'unregister_archive_format', 'get_unpack_formats', 'register_unpack_format', 'unregister_unpack_format', @@ -3465,6 +3491,8 @@ class PublicAPITests(unittest.TestCase): if hasattr(os, 'statvfs') or os.name == 'nt': target_api.append('disk_usage') self.assertEqual(set(shutil.__all__), set(target_api)) + with self.assertWarns(DeprecationWarning): + from shutil import ExecError if __name__ == '__main__':