mirror of
https://github.com/RustPython/RustPython.git
synced 2026-06-02 19:39:49 +09:00
5e1fc93f50
* Add Lib/test/archiver_tests.py @ 3.13.5 Needed for updated zipfile tests. * Update zipfile to 3.13.5 Notes: - I have to skip some brand new tests due to shift_jis encoding not being supported - `test_write_filtered_python_package` marked as `expectedFailure` with "AttributeError: module 'os' has no attribute 'supports_effective_ids'" - I didn't want to do a partial or full update to os module in this PR --------- Co-authored-by: Jack O'Connor <jack@jackoconnor.dev>
178 lines
7.1 KiB
Python
Vendored
178 lines
7.1 KiB
Python
Vendored
"""Tests common to tarfile and zipfile."""
|
|
|
|
import os
|
|
import sys
|
|
|
|
from test.support import swap_attr
|
|
from test.support import os_helper
|
|
|
|
class OverwriteTests:
|
|
|
|
def setUp(self):
|
|
os.makedirs(self.testdir)
|
|
self.addCleanup(os_helper.rmtree, self.testdir)
|
|
|
|
def create_file(self, path, content=b''):
|
|
with open(path, 'wb') as f:
|
|
f.write(content)
|
|
|
|
def open(self, path):
|
|
raise NotImplementedError
|
|
|
|
def extractall(self, ar):
|
|
raise NotImplementedError
|
|
|
|
|
|
def test_overwrite_file_as_file(self):
|
|
target = os.path.join(self.testdir, 'test')
|
|
self.create_file(target, b'content')
|
|
with self.open(self.ar_with_file) as ar:
|
|
self.extractall(ar)
|
|
self.assertTrue(os.path.isfile(target))
|
|
with open(target, 'rb') as f:
|
|
self.assertEqual(f.read(), b'newcontent')
|
|
|
|
def test_overwrite_dir_as_dir(self):
|
|
target = os.path.join(self.testdir, 'test')
|
|
os.mkdir(target)
|
|
with self.open(self.ar_with_dir) as ar:
|
|
self.extractall(ar)
|
|
self.assertTrue(os.path.isdir(target))
|
|
|
|
def test_overwrite_dir_as_implicit_dir(self):
|
|
target = os.path.join(self.testdir, 'test')
|
|
os.mkdir(target)
|
|
with self.open(self.ar_with_implicit_dir) as ar:
|
|
self.extractall(ar)
|
|
self.assertTrue(os.path.isdir(target))
|
|
self.assertTrue(os.path.isfile(os.path.join(target, 'file')))
|
|
with open(os.path.join(target, 'file'), 'rb') as f:
|
|
self.assertEqual(f.read(), b'newcontent')
|
|
|
|
def test_overwrite_dir_as_file(self):
|
|
target = os.path.join(self.testdir, 'test')
|
|
os.mkdir(target)
|
|
with self.open(self.ar_with_file) as ar:
|
|
with self.assertRaises(PermissionError if sys.platform == 'win32'
|
|
else IsADirectoryError):
|
|
self.extractall(ar)
|
|
self.assertTrue(os.path.isdir(target))
|
|
|
|
def test_overwrite_file_as_dir(self):
|
|
target = os.path.join(self.testdir, 'test')
|
|
self.create_file(target, b'content')
|
|
with self.open(self.ar_with_dir) as ar:
|
|
with self.assertRaises(FileExistsError):
|
|
self.extractall(ar)
|
|
self.assertTrue(os.path.isfile(target))
|
|
with open(target, 'rb') as f:
|
|
self.assertEqual(f.read(), b'content')
|
|
|
|
def test_overwrite_file_as_implicit_dir(self):
|
|
target = os.path.join(self.testdir, 'test')
|
|
self.create_file(target, b'content')
|
|
with self.open(self.ar_with_implicit_dir) as ar:
|
|
with self.assertRaises(FileNotFoundError if sys.platform == 'win32'
|
|
else NotADirectoryError):
|
|
self.extractall(ar)
|
|
self.assertTrue(os.path.isfile(target))
|
|
with open(target, 'rb') as f:
|
|
self.assertEqual(f.read(), b'content')
|
|
|
|
@os_helper.skip_unless_symlink
|
|
def test_overwrite_file_symlink_as_file(self):
|
|
# XXX: It is potential security vulnerability.
|
|
target = os.path.join(self.testdir, 'test')
|
|
target2 = os.path.join(self.testdir, 'test2')
|
|
self.create_file(target2, b'content')
|
|
os.symlink('test2', target)
|
|
with self.open(self.ar_with_file) as ar:
|
|
self.extractall(ar)
|
|
self.assertTrue(os.path.islink(target))
|
|
self.assertTrue(os.path.isfile(target2))
|
|
with open(target2, 'rb') as f:
|
|
self.assertEqual(f.read(), b'newcontent')
|
|
|
|
@os_helper.skip_unless_symlink
|
|
def test_overwrite_broken_file_symlink_as_file(self):
|
|
# XXX: It is potential security vulnerability.
|
|
target = os.path.join(self.testdir, 'test')
|
|
target2 = os.path.join(self.testdir, 'test2')
|
|
os.symlink('test2', target)
|
|
with self.open(self.ar_with_file) as ar:
|
|
self.extractall(ar)
|
|
self.assertTrue(os.path.islink(target))
|
|
self.assertTrue(os.path.isfile(target2))
|
|
with open(target2, 'rb') as f:
|
|
self.assertEqual(f.read(), b'newcontent')
|
|
|
|
@os_helper.skip_unless_symlink
|
|
def test_overwrite_dir_symlink_as_dir(self):
|
|
# XXX: It is potential security vulnerability.
|
|
target = os.path.join(self.testdir, 'test')
|
|
target2 = os.path.join(self.testdir, 'test2')
|
|
os.mkdir(target2)
|
|
os.symlink('test2', target, target_is_directory=True)
|
|
with self.open(self.ar_with_dir) as ar:
|
|
self.extractall(ar)
|
|
self.assertTrue(os.path.islink(target))
|
|
self.assertTrue(os.path.isdir(target2))
|
|
|
|
@os_helper.skip_unless_symlink
|
|
def test_overwrite_dir_symlink_as_implicit_dir(self):
|
|
# XXX: It is potential security vulnerability.
|
|
target = os.path.join(self.testdir, 'test')
|
|
target2 = os.path.join(self.testdir, 'test2')
|
|
os.mkdir(target2)
|
|
os.symlink('test2', target, target_is_directory=True)
|
|
with self.open(self.ar_with_implicit_dir) as ar:
|
|
self.extractall(ar)
|
|
self.assertTrue(os.path.islink(target))
|
|
self.assertTrue(os.path.isdir(target2))
|
|
self.assertTrue(os.path.isfile(os.path.join(target2, 'file')))
|
|
with open(os.path.join(target2, 'file'), 'rb') as f:
|
|
self.assertEqual(f.read(), b'newcontent')
|
|
|
|
@os_helper.skip_unless_symlink
|
|
def test_overwrite_broken_dir_symlink_as_dir(self):
|
|
target = os.path.join(self.testdir, 'test')
|
|
target2 = os.path.join(self.testdir, 'test2')
|
|
os.symlink('test2', target, target_is_directory=True)
|
|
with self.open(self.ar_with_dir) as ar:
|
|
with self.assertRaises(FileExistsError):
|
|
self.extractall(ar)
|
|
self.assertTrue(os.path.islink(target))
|
|
self.assertFalse(os.path.exists(target2))
|
|
|
|
@os_helper.skip_unless_symlink
|
|
def test_overwrite_broken_dir_symlink_as_implicit_dir(self):
|
|
target = os.path.join(self.testdir, 'test')
|
|
target2 = os.path.join(self.testdir, 'test2')
|
|
os.symlink('test2', target, target_is_directory=True)
|
|
with self.open(self.ar_with_implicit_dir) as ar:
|
|
with self.assertRaises(FileExistsError):
|
|
self.extractall(ar)
|
|
self.assertTrue(os.path.islink(target))
|
|
self.assertFalse(os.path.exists(target2))
|
|
|
|
def test_concurrent_extract_dir(self):
|
|
target = os.path.join(self.testdir, 'test')
|
|
def concurrent_mkdir(*args, **kwargs):
|
|
orig_mkdir(*args, **kwargs)
|
|
orig_mkdir(*args, **kwargs)
|
|
with swap_attr(os, 'mkdir', concurrent_mkdir) as orig_mkdir:
|
|
with self.open(self.ar_with_dir) as ar:
|
|
self.extractall(ar)
|
|
self.assertTrue(os.path.isdir(target))
|
|
|
|
def test_concurrent_extract_implicit_dir(self):
|
|
target = os.path.join(self.testdir, 'test')
|
|
def concurrent_mkdir(*args, **kwargs):
|
|
orig_mkdir(*args, **kwargs)
|
|
orig_mkdir(*args, **kwargs)
|
|
with swap_attr(os, 'mkdir', concurrent_mkdir) as orig_mkdir:
|
|
with self.open(self.ar_with_implicit_dir) as ar:
|
|
self.extractall(ar)
|
|
self.assertTrue(os.path.isdir(target))
|
|
self.assertTrue(os.path.isfile(os.path.join(target, 'file')))
|