mirror of
https://github.com/RustPython/RustPython.git
synced 2026-10-29 03:04:38 +08:00
Update shutil.py & zipfile to 3.14.5 (#7894)
* Update `shutil.py` to 3.14.5 * Update `zipfile` to 3.14.5
This commit is contained in:
Shahar Naveh authored and GitHub committed
1 parent
8c988711dd
commit
5a5230a400
6 files changed
+236
-72
No files matched your search
Vendored
+3
-21
@@ -1314,27 +1314,9 @@ def _unpack_zipfile(filename, extract_dir):
|
||||
if not zipfile.is_zipfile(filename):
|
||||
raise ReadError("%s is not a zip file" % filename)
|
||||
|
||||
zip = zipfile.ZipFile(filename)
|
||||
try:
|
||||
for info in zip.infolist():
|
||||
name = info.filename
|
||||
|
||||
# don't extract absolute paths or ones with .. in them
|
||||
if name.startswith('/') or '..' in name:
|
||||
continue
|
||||
|
||||
targetpath = os.path.join(extract_dir, *name.split('/'))
|
||||
if not targetpath:
|
||||
continue
|
||||
|
||||
_ensure_directory(targetpath)
|
||||
if not name.endswith('/'):
|
||||
# file
|
||||
with zip.open(name, 'r') as source, \
|
||||
open(targetpath, 'wb') as target:
|
||||
copyfileobj(source, target)
|
||||
finally:
|
||||
zip.close()
|
||||
with zipfile.ZipFile(filename) as zip:
|
||||
zip._ignore_invalid_names = True
|
||||
zip.extractall(extract_dir)
|
||||
|
||||
def _unpack_tarfile(filename, extract_dir, *, filter=None):
|
||||
"""Unpack tar/tar.gz/tar.bz2/tar.xz/tar.zst `filename` to `extract_dir`
|
||||
|
||||
Vendored
+65
-2
@@ -2110,8 +2110,6 @@ class TestArchives(BaseTest, unittest.TestCase):
|
||||
def check_unpack_archive(self, format, **kwargs):
|
||||
self.check_unpack_archive_with_converter(
|
||||
format, lambda path: path, **kwargs)
|
||||
self.check_unpack_archive_with_converter(
|
||||
format, FakePath, **kwargs)
|
||||
self.check_unpack_archive_with_converter(format, FakePath, **kwargs)
|
||||
|
||||
def check_unpack_archive_with_converter(self, format, converter, **kwargs):
|
||||
@@ -2168,6 +2166,71 @@ class TestArchives(BaseTest, unittest.TestCase):
|
||||
with self.assertRaises(TypeError):
|
||||
self.check_unpack_archive('zip', filter='data')
|
||||
|
||||
def test_unpack_archive_zip_badpaths(self):
|
||||
srcdir = self.mkdtemp()
|
||||
zipname = os.path.join(srcdir, 'test.zip')
|
||||
abspath = os.path.join(srcdir, 'abspath')
|
||||
with zipfile.ZipFile(zipname, 'w') as zf:
|
||||
zf.writestr(abspath, 'badfile')
|
||||
zf.writestr(os.sep + abspath, 'badfile')
|
||||
zf.writestr('/abspath', 'badfile')
|
||||
zf.writestr('C:/abspath', 'badfile')
|
||||
zf.writestr('D:\\abspath', 'badfile')
|
||||
zf.writestr('E:abspath', 'badfile')
|
||||
zf.writestr('F:/G:/abspath', 'badfile')
|
||||
zf.writestr('//server/share/abspath', 'badfile')
|
||||
zf.writestr('\\\\server2\\share\\abspath', 'badfile')
|
||||
zf.writestr('../relpath', 'badfile')
|
||||
zf.writestr(os.pardir + os.sep + 'relpath2', 'badfile')
|
||||
zf.writestr('good/file', 'goodfile')
|
||||
zf.writestr('good..file', 'goodfile')
|
||||
|
||||
dstdir = os.path.join(self.mkdtemp(), 'dst')
|
||||
unpack_archive(zipname, dstdir)
|
||||
self.assertTrue(os.path.isfile(os.path.join(dstdir, 'good', 'file')))
|
||||
self.assertTrue(os.path.isfile(os.path.join(dstdir, 'good..file')))
|
||||
self.assertFalse(os.path.exists(abspath))
|
||||
self.assertFalse(os.path.exists(os.path.join(dstdir, 'abspath')))
|
||||
self.assertFalse(os.path.exists(os.path.join(dstdir, 'G_')))
|
||||
self.assertFalse(os.path.exists(os.path.join(dstdir, 'server')))
|
||||
if os.name != 'nt':
|
||||
self.assertTrue(os.path.isfile(os.path.join(dstdir, 'C:', 'abspath')))
|
||||
self.assertTrue(os.path.isfile(os.path.join(dstdir, 'D:\\abspath')))
|
||||
self.assertTrue(os.path.isfile(os.path.join(dstdir, 'E:abspath')))
|
||||
self.assertTrue(os.path.isfile(os.path.join(dstdir, 'F:', 'G:', 'abspath')))
|
||||
self.assertTrue(os.path.isfile(os.path.join(dstdir, '\\\\server2\\share\\abspath')))
|
||||
if os.pardir == '..':
|
||||
self.assertFalse(os.path.exists(os.path.join(dstdir, '..', 'relpath')))
|
||||
self.assertFalse(os.path.exists(os.path.join(dstdir, 'relpath')))
|
||||
else:
|
||||
self.assertTrue(os.path.isfile(os.path.join(dstdir, '..', 'relpath')))
|
||||
self.assertFalse(os.path.exists(os.path.join(dstdir, os.pardir, 'relpath2')))
|
||||
self.assertFalse(os.path.exists(os.path.join(dstdir, 'relpath2')))
|
||||
|
||||
dstdir2 = os.path.join(self.mkdtemp(), 'dst')
|
||||
os.mkdir(dstdir2)
|
||||
with os_helper.change_cwd(dstdir2):
|
||||
unpack_archive(zipname, '')
|
||||
self.assertTrue(os.path.isfile(os.path.join('good', 'file')))
|
||||
self.assertTrue(os.path.isfile('good..file'))
|
||||
self.assertFalse(os.path.exists(abspath))
|
||||
self.assertFalse(os.path.exists('abspath'))
|
||||
self.assertFalse(os.path.exists('C_'))
|
||||
self.assertFalse(os.path.exists('server'))
|
||||
if os.name != 'nt':
|
||||
self.assertTrue(os.path.isfile(os.path.join('C:', 'abspath')))
|
||||
self.assertTrue(os.path.isfile('D:\\abspath'))
|
||||
self.assertTrue(os.path.isfile('E:abspath'))
|
||||
self.assertTrue(os.path.isfile(os.path.join('F:', 'G:', 'abspath')))
|
||||
self.assertTrue(os.path.isfile('\\\\server2\\share\\abspath'))
|
||||
if os.pardir == '..':
|
||||
self.assertFalse(os.path.exists(os.path.join('..', 'relpath')))
|
||||
self.assertFalse(os.path.exists('relpath'))
|
||||
else:
|
||||
self.assertTrue(os.path.isfile(os.path.join('..', 'relpath')))
|
||||
self.assertFalse(os.path.exists(os.path.join(os.pardir, 'relpath2')))
|
||||
self.assertFalse(os.path.exists('relpath2'))
|
||||
|
||||
def test_unpack_registry(self):
|
||||
|
||||
formats = get_unpack_formats()
|
||||
|
||||
+1
@@ -0,0 +1 @@
|
||||
|
||||
+1
-19
@@ -5,7 +5,6 @@ import pathlib
|
||||
import pickle
|
||||
import stat
|
||||
import sys
|
||||
import time
|
||||
import unittest
|
||||
import zipfile
|
||||
import zipfile._path
|
||||
@@ -649,7 +648,7 @@ class TestPath(unittest.TestCase):
|
||||
"""
|
||||
data = io.BytesIO()
|
||||
zf = zipfile.ZipFile(data, "w")
|
||||
zf.writestr(DirtyZipInfo.for_name("foo\\bar", zf), b"content")
|
||||
zf.writestr(DirtyZipInfo("foo\\bar")._for_archive(zf), b"content")
|
||||
zf.filename = ''
|
||||
root = zipfile.Path(zf)
|
||||
(first,) = root.iterdir()
|
||||
@@ -672,20 +671,3 @@ class DirtyZipInfo(zipfile.ZipInfo):
|
||||
def __init__(self, filename, *args, **kwargs):
|
||||
super().__init__(filename, *args, **kwargs)
|
||||
self.filename = filename
|
||||
|
||||
@classmethod
|
||||
def for_name(cls, name, archive):
|
||||
"""
|
||||
Construct the same way that ZipFile.writestr does.
|
||||
|
||||
TODO: extract this functionality and re-use
|
||||
"""
|
||||
self = cls(filename=name, date_time=time.localtime(time.time())[:6])
|
||||
self.compress_type = archive.compression
|
||||
self.compress_level = archive.compresslevel
|
||||
if self.filename.endswith('/'): # pragma: no cover
|
||||
self.external_attr = 0o40775 << 16 # drwxrwxr-x
|
||||
self.external_attr |= 0x10 # MS-DOS directory flag
|
||||
else:
|
||||
self.external_attr = 0o600 << 16 # ?rw-------
|
||||
return self
|
||||
Vendored
+150
-25
@@ -6,6 +6,7 @@ import io
|
||||
import itertools
|
||||
import os
|
||||
import posixpath
|
||||
import stat
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
@@ -19,14 +20,16 @@ from tempfile import TemporaryFile
|
||||
from random import randint, random, randbytes
|
||||
|
||||
from test import archiver_tests
|
||||
from test.support import script_helper
|
||||
from test.support import script_helper, os_helper
|
||||
from test.support import (
|
||||
findfile, requires_zlib, requires_bz2, requires_lzma,
|
||||
captured_stdout, captured_stderr, requires_subprocess
|
||||
requires_zstd, captured_stdout, captured_stderr, requires_subprocess,
|
||||
cpython_only
|
||||
)
|
||||
from test.support.os_helper import (
|
||||
TESTFN, unlink, rmtree, temp_dir, temp_cwd, fd_count, FakePath
|
||||
)
|
||||
from test.support.import_helper import ensure_lazy_imports
|
||||
|
||||
|
||||
TESTFN2 = TESTFN + "2"
|
||||
@@ -48,6 +51,13 @@ def get_files(test):
|
||||
yield f
|
||||
test.assertFalse(f.closed)
|
||||
|
||||
|
||||
class LazyImportTest(unittest.TestCase):
|
||||
@cpython_only
|
||||
def test_lazy_import(self):
|
||||
ensure_lazy_imports("zipfile", {"typing"})
|
||||
|
||||
|
||||
class AbstractTestsWithSourceFile:
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
@@ -692,6 +702,10 @@ class LzmaTestsWithSourceFile(AbstractTestsWithSourceFile,
|
||||
unittest.TestCase):
|
||||
compression = zipfile.ZIP_LZMA
|
||||
|
||||
@requires_zstd()
|
||||
class ZstdTestsWithSourceFile(AbstractTestsWithSourceFile,
|
||||
unittest.TestCase):
|
||||
compression = zipfile.ZIP_ZSTANDARD
|
||||
|
||||
class AbstractTestZip64InSmallFiles:
|
||||
# These tests test the ZIP64 functionality without using large files,
|
||||
@@ -1343,6 +1357,10 @@ class LzmaTestZip64InSmallFiles(AbstractTestZip64InSmallFiles,
|
||||
unittest.TestCase):
|
||||
compression = zipfile.ZIP_LZMA
|
||||
|
||||
@requires_zstd()
|
||||
class ZstdTestZip64InSmallFiles(AbstractTestZip64InSmallFiles,
|
||||
unittest.TestCase):
|
||||
compression = zipfile.ZIP_ZSTANDARD
|
||||
|
||||
class AbstractWriterTests:
|
||||
|
||||
@@ -1412,6 +1430,9 @@ class Bzip2WriterTests(AbstractWriterTests, unittest.TestCase):
|
||||
class LzmaWriterTests(AbstractWriterTests, unittest.TestCase):
|
||||
compression = zipfile.ZIP_LZMA
|
||||
|
||||
@requires_zstd()
|
||||
class ZstdWriterTests(AbstractWriterTests, unittest.TestCase):
|
||||
compression = zipfile.ZIP_ZSTANDARD
|
||||
|
||||
class PyZipFileTests(unittest.TestCase):
|
||||
def assertCompiledIn(self, name, namelist):
|
||||
@@ -1431,7 +1452,7 @@ class PyZipFileTests(unittest.TestCase):
|
||||
self.skipTest('requires write access to the installed location')
|
||||
unlink(filename)
|
||||
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON
|
||||
def test_write_pyfile(self):
|
||||
self.requiresWriteAccess(os.path.dirname(__file__))
|
||||
with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp:
|
||||
@@ -1462,7 +1483,7 @@ class PyZipFileTests(unittest.TestCase):
|
||||
self.assertNotIn(bn, zipfp.namelist())
|
||||
self.assertCompiledIn(bn, zipfp.namelist())
|
||||
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON
|
||||
def test_write_python_package(self):
|
||||
import email
|
||||
packagedir = os.path.dirname(email.__file__)
|
||||
@@ -1477,7 +1498,7 @@ class PyZipFileTests(unittest.TestCase):
|
||||
self.assertCompiledIn('email/__init__.py', names)
|
||||
self.assertCompiledIn('email/mime/text.py', names)
|
||||
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON; - AttributeError: module 'os' has no attribute 'supports_effective_ids'
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON; - AttributeError: module 'os' has no attribute 'supports_effective_ids'
|
||||
def test_write_filtered_python_package(self):
|
||||
import test
|
||||
packagedir = os.path.dirname(test.__file__)
|
||||
@@ -1508,7 +1529,7 @@ class PyZipFileTests(unittest.TestCase):
|
||||
print(reportStr)
|
||||
self.assertTrue('SyntaxError' not in reportStr)
|
||||
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON
|
||||
def test_write_with_optimization(self):
|
||||
import email
|
||||
packagedir = os.path.dirname(email.__file__)
|
||||
@@ -1859,6 +1880,35 @@ class OtherTests(unittest.TestCase):
|
||||
zinfo.flag_bits |= zipfile._MASK_USE_DATA_DESCRIPTOR # Include an extended local header.
|
||||
orig_zip.writestr(zinfo, data)
|
||||
|
||||
def test_write_with_source_date_epoch(self):
|
||||
with os_helper.EnvironmentVarGuard() as env:
|
||||
# Set the SOURCE_DATE_EPOCH environment variable to a specific timestamp
|
||||
env['SOURCE_DATE_EPOCH'] = "1735715999"
|
||||
|
||||
with zipfile.ZipFile(TESTFN, "w") as zf:
|
||||
zf.writestr("test_source_date_epoch.txt", "Testing SOURCE_DATE_EPOCH")
|
||||
|
||||
with zipfile.ZipFile(TESTFN, "r") as zf:
|
||||
zip_info = zf.getinfo("test_source_date_epoch.txt")
|
||||
get_time = time.localtime(int(os.environ['SOURCE_DATE_EPOCH']))[:6]
|
||||
# Compare each element of the date_time tuple
|
||||
# Allow for a 1-second difference
|
||||
for z_time, g_time in zip(zip_info.date_time, get_time):
|
||||
self.assertAlmostEqual(z_time, g_time, delta=1)
|
||||
|
||||
def test_write_without_source_date_epoch(self):
|
||||
with os_helper.EnvironmentVarGuard() as env:
|
||||
del env['SOURCE_DATE_EPOCH']
|
||||
|
||||
with zipfile.ZipFile(TESTFN, "w") as zf:
|
||||
zf.writestr("test_no_source_date_epoch.txt", "Testing without SOURCE_DATE_EPOCH")
|
||||
|
||||
with zipfile.ZipFile(TESTFN, "r") as zf:
|
||||
zip_info = zf.getinfo("test_no_source_date_epoch.txt")
|
||||
current_time = time.localtime()[:6]
|
||||
for z_time, c_time in zip(zip_info.date_time, current_time):
|
||||
self.assertAlmostEqual(z_time, c_time, delta=2)
|
||||
|
||||
def test_close(self):
|
||||
"""Check that the zipfile is closed after the 'with' block."""
|
||||
with zipfile.ZipFile(TESTFN2, "w") as zipfp:
|
||||
@@ -2019,6 +2069,25 @@ class OtherTests(unittest.TestCase):
|
||||
self.assertFalse(zipfile.is_zipfile(fp))
|
||||
fp.seek(0, 0)
|
||||
self.assertFalse(zipfile.is_zipfile(fp))
|
||||
# - passing non-zipfile with ZIP header elements
|
||||
# data created using pyPNG like so:
|
||||
# d = [(ord('P'), ord('K'), 5, 6), (ord('P'), ord('K'), 6, 6)]
|
||||
# w = png.Writer(1,2,alpha=True,compression=0)
|
||||
# f = open('onepix.png', 'wb')
|
||||
# w.write(f, d)
|
||||
# w.close()
|
||||
data = (b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00"
|
||||
b"\x00\x02\x08\x06\x00\x00\x00\x99\x81\xb6'\x00\x00\x00\x15I"
|
||||
b"DATx\x01\x01\n\x00\xf5\xff\x00PK\x05\x06\x00PK\x06\x06\x07"
|
||||
b"\xac\x01N\xc6|a\r\x00\x00\x00\x00IEND\xaeB`\x82")
|
||||
# - passing a filename
|
||||
with open(TESTFN, "wb") as fp:
|
||||
fp.write(data)
|
||||
self.assertFalse(zipfile.is_zipfile(TESTFN))
|
||||
# - passing a file-like object
|
||||
fp = io.BytesIO()
|
||||
fp.write(data)
|
||||
self.assertFalse(zipfile.is_zipfile(fp))
|
||||
|
||||
def test_damaged_zipfile(self):
|
||||
"""Check that zipfiles with missing bytes at the end raise BadZipFile."""
|
||||
@@ -2048,10 +2117,16 @@ class OtherTests(unittest.TestCase):
|
||||
zip_contents = fp.read()
|
||||
# - passing a file-like object
|
||||
fp = io.BytesIO()
|
||||
fp.write(zip_contents)
|
||||
end = fp.write(zip_contents)
|
||||
self.assertEqual(fp.tell(), end)
|
||||
mid = end // 2
|
||||
fp.seek(mid, 0)
|
||||
self.assertTrue(zipfile.is_zipfile(fp))
|
||||
fp.seek(0, 0)
|
||||
# check that the position is left unchanged after the call
|
||||
# see: https://github.com/python/cpython/issues/122356
|
||||
self.assertEqual(fp.tell(), mid)
|
||||
self.assertTrue(zipfile.is_zipfile(fp))
|
||||
self.assertEqual(fp.tell(), mid)
|
||||
|
||||
def test_non_existent_file_raises_OSError(self):
|
||||
# make sure we don't raise an AttributeError when a partially-constructed
|
||||
@@ -2286,6 +2361,34 @@ class OtherTests(unittest.TestCase):
|
||||
zi = zipfile.ZipInfo(filename="empty")
|
||||
self.assertEqual(repr(zi), "<ZipInfo filename='empty' file_size=0>")
|
||||
|
||||
def test_for_archive(self):
|
||||
base_filename = TESTFN2.rstrip('/')
|
||||
|
||||
with zipfile.ZipFile(TESTFN, mode="w", compresslevel=1,
|
||||
compression=zipfile.ZIP_STORED) as zf:
|
||||
# no trailing forward slash
|
||||
zi = zipfile.ZipInfo(base_filename)._for_archive(zf)
|
||||
self.assertEqual(zi.compress_level, 1)
|
||||
self.assertEqual(zi.compress_type, zipfile.ZIP_STORED)
|
||||
# ?rw- --- ---
|
||||
filemode = stat.S_IRUSR | stat.S_IWUSR
|
||||
# filemode is stored as the highest 16 bits of external_attr
|
||||
self.assertEqual(zi.external_attr >> 16, filemode)
|
||||
self.assertEqual(zi.external_attr & 0xFF, 0) # no MS-DOS flag
|
||||
|
||||
with zipfile.ZipFile(TESTFN, mode="w", compresslevel=1,
|
||||
compression=zipfile.ZIP_STORED) as zf:
|
||||
# with a trailing slash
|
||||
zi = zipfile.ZipInfo(f'{base_filename}/')._for_archive(zf)
|
||||
self.assertEqual(zi.compress_level, 1)
|
||||
self.assertEqual(zi.compress_type, zipfile.ZIP_STORED)
|
||||
# d rwx rwx r-x
|
||||
filemode = stat.S_IFDIR
|
||||
filemode |= stat.S_IRWXU | stat.S_IRWXG
|
||||
filemode |= stat.S_IROTH | stat.S_IXOTH
|
||||
self.assertEqual(zi.external_attr >> 16, filemode)
|
||||
self.assertEqual(zi.external_attr & 0xFF, 0x10) # MS-DOS flag
|
||||
|
||||
def test_create_empty_zipinfo_default_attributes(self):
|
||||
"""Ensure all required attributes are set."""
|
||||
zi = zipfile.ZipInfo()
|
||||
@@ -2432,6 +2535,10 @@ class OtherTests(unittest.TestCase):
|
||||
|
||||
@requires_zlib()
|
||||
def test_full_overlap_different_names(self):
|
||||
# The ZIP file contains two central directory entries with
|
||||
# different names which refer to the same local header.
|
||||
# The name of the local header matches the name of the first
|
||||
# central directory entry.
|
||||
data = (
|
||||
b'PK\x03\x04\x14\x00\x00\x00\x08\x00\xa0lH\x05\xe2\x1e'
|
||||
b'8\xbb\x10\x00\x00\x00\t\x04\x00\x00\x01\x00\x00\x00b\xed'
|
||||
@@ -2461,6 +2568,10 @@ class OtherTests(unittest.TestCase):
|
||||
|
||||
@requires_zlib()
|
||||
def test_full_overlap_different_names2(self):
|
||||
# The ZIP file contains two central directory entries with
|
||||
# different names which refer to the same local header.
|
||||
# The name of the local header matches the name of the second
|
||||
# central directory entry.
|
||||
data = (
|
||||
b'PK\x03\x04\x14\x00\x00\x00\x08\x00\xa0lH\x05\xe2\x1e'
|
||||
b'8\xbb\x10\x00\x00\x00\t\x04\x00\x00\x01\x00\x00\x00a\xed'
|
||||
@@ -2492,6 +2603,8 @@ class OtherTests(unittest.TestCase):
|
||||
|
||||
@requires_zlib()
|
||||
def test_full_overlap_same_name(self):
|
||||
# The ZIP file contains two central directory entries with
|
||||
# the same name which refer to the same local header.
|
||||
data = (
|
||||
b'PK\x03\x04\x14\x00\x00\x00\x08\x00\xa0lH\x05\xe2\x1e'
|
||||
b'8\xbb\x10\x00\x00\x00\t\x04\x00\x00\x01\x00\x00\x00a\xed'
|
||||
@@ -2524,6 +2637,8 @@ class OtherTests(unittest.TestCase):
|
||||
|
||||
@requires_zlib()
|
||||
def test_quoted_overlap(self):
|
||||
# The ZIP file contains two files. The second local header
|
||||
# is contained in the range of the first file.
|
||||
data = (
|
||||
b'PK\x03\x04\x14\x00\x00\x00\x08\x00\xa0lH\x05Y\xfc'
|
||||
b'8\x044\x00\x00\x00(\x04\x00\x00\x01\x00\x00\x00a\x00'
|
||||
@@ -2555,6 +2670,7 @@ class OtherTests(unittest.TestCase):
|
||||
|
||||
@requires_zlib()
|
||||
def test_overlap_with_central_dir(self):
|
||||
# The local header offset is equal to the central directory offset.
|
||||
data = (
|
||||
b'PK\x01\x02\x14\x03\x14\x00\x00\x00\x08\x00G_|Z'
|
||||
b'\xe2\x1e8\xbb\x0b\x00\x00\x00\t\x04\x00\x00\x01\x00\x00\x00'
|
||||
@@ -2569,11 +2685,15 @@ class OtherTests(unittest.TestCase):
|
||||
self.assertEqual(zi.header_offset, 0)
|
||||
self.assertEqual(zi.compress_size, 11)
|
||||
self.assertEqual(zi.file_size, 1033)
|
||||
# Found central directory signature PK\x01\x02 instead of
|
||||
# local header signature PK\x03\x04.
|
||||
with self.assertRaisesRegex(zipfile.BadZipFile, 'Bad magic number'):
|
||||
zipf.read('a')
|
||||
|
||||
@requires_zlib()
|
||||
def test_overlap_with_archive_comment(self):
|
||||
# The local header is written after the central directory,
|
||||
# in the archive comment.
|
||||
data = (
|
||||
b'PK\x01\x02\x14\x03\x14\x00\x00\x00\x08\x00G_|Z'
|
||||
b'\xe2\x1e8\xbb\x0b\x00\x00\x00\t\x04\x00\x00\x01\x00\x00\x00'
|
||||
@@ -2685,6 +2805,17 @@ class LzmaBadCrcTests(AbstractBadCrcTests, unittest.TestCase):
|
||||
b'ePK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x003\x00\x00'
|
||||
b'\x00>\x00\x00\x00\x00\x00')
|
||||
|
||||
@requires_zstd()
|
||||
class ZstdBadCrcTests(AbstractBadCrcTests, unittest.TestCase):
|
||||
compression = zipfile.ZIP_ZSTANDARD
|
||||
zip_with_bad_crc = (
|
||||
b'PK\x03\x04?\x00\x00\x00]\x00\x00\x00!\x00V\xb1\x17J\x14\x00'
|
||||
b'\x00\x00\x0b\x00\x00\x00\x05\x00\x00\x00afile(\xb5/\xfd\x00'
|
||||
b'XY\x00\x00Hello WorldPK\x01\x02?\x03?\x00\x00\x00]\x00\x00\x00'
|
||||
b'!\x00V\xb0\x17J\x14\x00\x00\x00\x0b\x00\x00\x00\x05\x00\x00\x00'
|
||||
b'\x00\x00\x00\x00\x00\x00\x00\x00\x80\x01\x00\x00\x00\x00afilePK'
|
||||
b'\x05\x06\x00\x00\x00\x00\x01\x00\x01\x003\x00\x00\x007\x00\x00\x00'
|
||||
b'\x00\x00')
|
||||
|
||||
class DecryptionTests(unittest.TestCase):
|
||||
"""Check that ZIP decryption works. Since the library does not
|
||||
@@ -2912,6 +3043,10 @@ class LzmaTestsWithRandomBinaryFiles(AbstractTestsWithRandomBinaryFiles,
|
||||
unittest.TestCase):
|
||||
compression = zipfile.ZIP_LZMA
|
||||
|
||||
@requires_zstd()
|
||||
class ZstdTestsWithRandomBinaryFiles(AbstractTestsWithRandomBinaryFiles,
|
||||
unittest.TestCase):
|
||||
compression = zipfile.ZIP_ZSTANDARD
|
||||
|
||||
# Provide the tell() method but not seek()
|
||||
class Tellable:
|
||||
@@ -3160,7 +3295,7 @@ class TestWithDirectory(unittest.TestCase):
|
||||
with zipfile.ZipFile(TESTFN, "w") as zipf:
|
||||
zipf.write(dirpath)
|
||||
zinfo = zipf.filelist[0]
|
||||
self.assertTrue(zinfo.filename.endswith("/x/"))
|
||||
self.assertEndsWith(zinfo.filename, "/x/")
|
||||
self.assertEqual(zinfo.external_attr, (mode << 16) | 0x10)
|
||||
zipf.write(dirpath, "y")
|
||||
zinfo = zipf.filelist[1]
|
||||
@@ -3168,7 +3303,7 @@ class TestWithDirectory(unittest.TestCase):
|
||||
self.assertEqual(zinfo.external_attr, (mode << 16) | 0x10)
|
||||
with zipfile.ZipFile(TESTFN, "r") as zipf:
|
||||
zinfo = zipf.filelist[0]
|
||||
self.assertTrue(zinfo.filename.endswith("/x/"))
|
||||
self.assertEndsWith(zinfo.filename, "/x/")
|
||||
self.assertEqual(zinfo.external_attr, (mode << 16) | 0x10)
|
||||
zinfo = zipf.filelist[1]
|
||||
self.assertTrue(zinfo.filename, "y/")
|
||||
@@ -3188,7 +3323,7 @@ class TestWithDirectory(unittest.TestCase):
|
||||
self.assertEqual(zinfo.external_attr, (0o40775 << 16) | 0x10)
|
||||
with zipfile.ZipFile(TESTFN, "r") as zipf:
|
||||
zinfo = zipf.filelist[0]
|
||||
self.assertTrue(zinfo.filename.endswith("x/"))
|
||||
self.assertEndsWith(zinfo.filename, "x/")
|
||||
self.assertEqual(zinfo.external_attr, (0o40775 << 16) | 0x10)
|
||||
target = os.path.join(TESTFN2, "target")
|
||||
os.mkdir(target)
|
||||
@@ -3415,7 +3550,7 @@ class TestExecutablePrependedZip(unittest.TestCase):
|
||||
def test_read_zip64_with_exe_prepended(self):
|
||||
self._test_zip_works(self.exe_zip64)
|
||||
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON
|
||||
@unittest.skipUnless(sys.executable, 'sys.executable required.')
|
||||
@unittest.skipUnless(os.access('/bin/bash', os.X_OK),
|
||||
'Test relies on #!/bin/bash working.')
|
||||
@@ -3424,7 +3559,7 @@ class TestExecutablePrependedZip(unittest.TestCase):
|
||||
output = subprocess.check_output([self.exe_zip, sys.executable])
|
||||
self.assertIn(b'number in executable: 5', output)
|
||||
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON
|
||||
@unittest.skipUnless(sys.executable, 'sys.executable required.')
|
||||
@unittest.skipUnless(os.access('/bin/bash', os.X_OK),
|
||||
'Test relies on #!/bin/bash working.')
|
||||
@@ -3433,17 +3568,15 @@ class TestExecutablePrependedZip(unittest.TestCase):
|
||||
output = subprocess.check_output([self.exe_zip64, sys.executable])
|
||||
self.assertIn(b'number in executable: 5', output)
|
||||
|
||||
|
||||
@unittest.skip("TODO: RUSTPYTHON shift_jis encoding unsupported")
|
||||
@unittest.skip("TODO: RUSTPYTHON; LookupError: unknown encoding: shift_jis")
|
||||
class EncodedMetadataTests(unittest.TestCase):
|
||||
file_names = ['\u4e00', '\u4e8c', '\u4e09'] # Han 'one', 'two', 'three'
|
||||
file_content = [
|
||||
"This is pure ASCII.\n".encode('ascii'),
|
||||
# This is modern Japanese. (UTF-8)
|
||||
"\u3053\u308c\u306f\u73fe\u4ee3\u7684\u65e5\u672c\u8a9e\u3067\u3059\u3002\n".encode('utf-8'),
|
||||
# TODO RUSTPYTHON
|
||||
# Uncomment when Shift JIS is supported
|
||||
# This is obsolete Japanese. (Shift JIS)
|
||||
# TODO: RUSTPYTHON; LookupError: unknown encoding: shift_jis
|
||||
# "\u3053\u308c\u306f\u53e4\u3044\u65e5\u672c\u8a9e\u3067\u3059\u3002\n".encode('shift_jis'),
|
||||
]
|
||||
|
||||
@@ -3484,13 +3617,11 @@ class EncodedMetadataTests(unittest.TestCase):
|
||||
self.assertEqual(info.file_size, len(content))
|
||||
self.assertEqual(zipfp.read(name), content)
|
||||
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON;
|
||||
def test_read_with_metadata_encoding(self):
|
||||
# Read the ZIP archive with correct metadata_encoding
|
||||
with zipfile.ZipFile(TESTFN, "r", metadata_encoding='shift_jis') as zipfp:
|
||||
self._test_read(zipfp, self.file_names, self.file_content)
|
||||
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON;
|
||||
def test_read_without_metadata_encoding(self):
|
||||
# Read the ZIP archive without metadata_encoding
|
||||
expected_names = [name.encode('shift_jis').decode('cp437')
|
||||
@@ -3498,7 +3629,6 @@ class EncodedMetadataTests(unittest.TestCase):
|
||||
with zipfile.ZipFile(TESTFN, "r") as zipfp:
|
||||
self._test_read(zipfp, expected_names, self.file_content)
|
||||
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON;
|
||||
def test_read_with_incorrect_metadata_encoding(self):
|
||||
# Read the ZIP archive with incorrect metadata_encoding
|
||||
expected_names = [name.encode('shift_jis').decode('koi8-u')
|
||||
@@ -3506,7 +3636,6 @@ class EncodedMetadataTests(unittest.TestCase):
|
||||
with zipfile.ZipFile(TESTFN, "r", metadata_encoding='koi8-u') as zipfp:
|
||||
self._test_read(zipfp, expected_names, self.file_content)
|
||||
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON;
|
||||
def test_read_with_unsuitable_metadata_encoding(self):
|
||||
# Read the ZIP archive with metadata_encoding unsuitable for
|
||||
# decoding metadata
|
||||
@@ -3515,7 +3644,6 @@ class EncodedMetadataTests(unittest.TestCase):
|
||||
with self.assertRaises(UnicodeDecodeError):
|
||||
zipfile.ZipFile(TESTFN, "r", metadata_encoding='utf-8')
|
||||
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON;
|
||||
def test_read_after_append(self):
|
||||
newname = '\u56db' # Han 'four'
|
||||
expected_names = [name.encode('shift_jis').decode('cp437')
|
||||
@@ -3542,7 +3670,6 @@ class EncodedMetadataTests(unittest.TestCase):
|
||||
else:
|
||||
self.assertEqual(zipfp.read(name), content)
|
||||
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON;
|
||||
def test_write_with_metadata_encoding(self):
|
||||
ZF = zipfile.ZipFile
|
||||
for mode in ("w", "x", "a"):
|
||||
@@ -3550,7 +3677,6 @@ class EncodedMetadataTests(unittest.TestCase):
|
||||
"^metadata_encoding is only"):
|
||||
ZF("nonesuch.zip", mode, metadata_encoding="shift_jis")
|
||||
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON;
|
||||
def test_cli_with_metadata_encoding(self):
|
||||
errmsg = "Non-conforming encodings not supported with -c."
|
||||
args = ["--metadata-encoding=shift_jis", "-c", "nonesuch", "nonesuch"]
|
||||
@@ -3570,7 +3696,6 @@ class EncodedMetadataTests(unittest.TestCase):
|
||||
for name in self.file_names:
|
||||
self.assertIn(name, listing)
|
||||
|
||||
@unittest.expectedFailure # TODO: RUSTPYTHON;
|
||||
def test_cli_with_metadata_encoding_extract(self):
|
||||
os.mkdir(TESTFN2)
|
||||
self.addCleanup(rmtree, TESTFN2)
|
||||
|
||||
Vendored
+16
-5
@@ -1410,6 +1410,7 @@ class ZipFile:
|
||||
|
||||
fp = None # Set here since __del__ checks it
|
||||
_windows_illegal_name_trans_table = None
|
||||
_ignore_invalid_names = False
|
||||
|
||||
def __init__(self, file, mode="r", compression=ZIP_STORED, allowZip64=True,
|
||||
compresslevel=None, *, strict_timestamps=True, metadata_encoding=None):
|
||||
@@ -1890,21 +1891,31 @@ class ZipFile:
|
||||
|
||||
# build the destination pathname, replacing
|
||||
# forward slashes to platform specific separators.
|
||||
arcname = member.filename.replace('/', os.path.sep)
|
||||
|
||||
if os.path.altsep:
|
||||
arcname = member.filename
|
||||
if os.path.sep != '/':
|
||||
arcname = arcname.replace('/', os.path.sep)
|
||||
if os.path.altsep and os.path.altsep != '/':
|
||||
arcname = arcname.replace(os.path.altsep, os.path.sep)
|
||||
# interpret absolute pathname as relative, remove drive letter or
|
||||
# UNC path, redundant separators, "." and ".." components.
|
||||
arcname = os.path.splitdrive(arcname)[1]
|
||||
drive, root, arcname = os.path.splitroot(arcname)
|
||||
if self._ignore_invalid_names and (drive or root):
|
||||
return None
|
||||
if self._ignore_invalid_names and os.path.pardir in arcname.split(os.path.sep):
|
||||
return None
|
||||
invalid_path_parts = ('', os.path.curdir, os.path.pardir)
|
||||
arcname = os.path.sep.join(x for x in arcname.split(os.path.sep)
|
||||
if x not in invalid_path_parts)
|
||||
if os.path.sep == '\\':
|
||||
# filter illegal characters on Windows
|
||||
arcname = self._sanitize_windows_name(arcname, os.path.sep)
|
||||
arcname2 = self._sanitize_windows_name(arcname, os.path.sep)
|
||||
if self._ignore_invalid_names and arcname2 != arcname:
|
||||
return None
|
||||
arcname = arcname2
|
||||
|
||||
if not arcname and not member.is_dir():
|
||||
if self._ignore_invalid_names:
|
||||
return None
|
||||
raise ValueError("Empty filename.")
|
||||
|
||||
targetpath = os.path.join(targetpath, arcname)
|
||||
|
||||
Reference in new issue
Block a user