395 lines
14 KiB
Python
395 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Tests for distutils.archive_util."""
|
|
import unittest
|
|
import os
|
|
import sys
|
|
import tarfile
|
|
from os.path import splitdrive
|
|
import warnings
|
|
|
|
from distutils import archive_util
|
|
from distutils.archive_util import (check_archive_formats, make_tarball,
|
|
make_zipfile, make_archive,
|
|
ARCHIVE_FORMATS)
|
|
from distutils.spawn import find_executable, spawn
|
|
from distutils.tests import support
|
|
from test.support import check_warnings, run_unittest, patch, change_cwd
|
|
|
|
try:
|
|
import grp
|
|
import pwd
|
|
UID_GID_SUPPORT = True
|
|
except ImportError:
|
|
UID_GID_SUPPORT = False
|
|
|
|
try:
|
|
import zipfile
|
|
ZIP_SUPPORT = True
|
|
except ImportError:
|
|
ZIP_SUPPORT = find_executable('zip')
|
|
|
|
try:
|
|
import zlib
|
|
ZLIB_SUPPORT = True
|
|
except ImportError:
|
|
ZLIB_SUPPORT = False
|
|
|
|
try:
|
|
import bz2
|
|
except ImportError:
|
|
bz2 = None
|
|
|
|
try:
|
|
import lzma
|
|
except ImportError:
|
|
lzma = None
|
|
|
|
def can_fs_encode(filename):
|
|
"""
|
|
Return True if the filename can be saved in the file system.
|
|
"""
|
|
if os.path.supports_unicode_filenames:
|
|
return True
|
|
try:
|
|
filename.encode(sys.getfilesystemencoding())
|
|
except UnicodeEncodeError:
|
|
return False
|
|
return True
|
|
|
|
|
|
class ArchiveUtilTestCase(support.TempdirManager,
|
|
support.LoggingSilencer,
|
|
unittest.TestCase):
|
|
|
|
@unittest.skipUnless(ZLIB_SUPPORT, 'Need zlib support to run')
|
|
def test_make_tarball(self, name='archive'):
|
|
# creating something to tar
|
|
tmpdir = self._create_files()
|
|
self._make_tarball(tmpdir, name, '.tar.gz')
|
|
# trying an uncompressed one
|
|
self._make_tarball(tmpdir, name, '.tar', compress=None)
|
|
|
|
@unittest.skipUnless(ZLIB_SUPPORT, 'Need zlib support to run')
|
|
def test_make_tarball_gzip(self):
|
|
tmpdir = self._create_files()
|
|
self._make_tarball(tmpdir, 'archive', '.tar.gz', compress='gzip')
|
|
|
|
@unittest.skipUnless(bz2, 'Need bz2 support to run')
|
|
def test_make_tarball_bzip2(self):
|
|
tmpdir = self._create_files()
|
|
self._make_tarball(tmpdir, 'archive', '.tar.bz2', compress='bzip2')
|
|
|
|
@unittest.skipUnless(lzma, 'Need lzma support to run')
|
|
def test_make_tarball_xz(self):
|
|
tmpdir = self._create_files()
|
|
self._make_tarball(tmpdir, 'archive', '.tar.xz', compress='xz')
|
|
|
|
@unittest.skipUnless(can_fs_encode('årchiv'),
|
|
'File system cannot handle this filename')
|
|
def test_make_tarball_latin1(self):
|
|
"""
|
|
Mirror test_make_tarball, except filename contains latin characters.
|
|
"""
|
|
self.test_make_tarball('årchiv') # note this isn't a real word
|
|
|
|
@unittest.skipUnless(can_fs_encode('のアーカイブ'),
|
|
'File system cannot handle this filename')
|
|
def test_make_tarball_extended(self):
|
|
"""
|
|
Mirror test_make_tarball, except filename contains extended
|
|
characters outside the latin charset.
|
|
"""
|
|
self.test_make_tarball('のアーカイブ') # japanese for archive
|
|
|
|
def _make_tarball(self, tmpdir, target_name, suffix, **kwargs):
|
|
tmpdir2 = self.mkdtemp()
|
|
unittest.skipUnless(splitdrive(tmpdir)[0] == splitdrive(tmpdir2)[0],
|
|
"source and target should be on same drive")
|
|
|
|
base_name = os.path.join(tmpdir2, target_name)
|
|
|
|
# working with relative paths to avoid tar warnings
|
|
with change_cwd(tmpdir):
|
|
make_tarball(splitdrive(base_name)[1], 'dist', **kwargs)
|
|
|
|
# check if the compressed tarball was created
|
|
tarball = base_name + suffix
|
|
self.assertTrue(os.path.exists(tarball))
|
|
self.assertEqual(self._tarinfo(tarball), self._created_files)
|
|
|
|
def _tarinfo(self, path):
|
|
tar = tarfile.open(path)
|
|
try:
|
|
names = tar.getnames()
|
|
names.sort()
|
|
return names
|
|
finally:
|
|
tar.close()
|
|
|
|
_zip_created_files = ['dist/', 'dist/file1', 'dist/file2',
|
|
'dist/sub/', 'dist/sub/file3', 'dist/sub2/']
|
|
_created_files = [p.rstrip('/') for p in _zip_created_files]
|
|
|
|
def _create_files(self):
|
|
# creating something to tar
|
|
tmpdir = self.mkdtemp()
|
|
dist = os.path.join(tmpdir, 'dist')
|
|
os.mkdir(dist)
|
|
self.write_file([dist, 'file1'], 'xxx')
|
|
self.write_file([dist, 'file2'], 'xxx')
|
|
os.mkdir(os.path.join(dist, 'sub'))
|
|
self.write_file([dist, 'sub', 'file3'], 'xxx')
|
|
os.mkdir(os.path.join(dist, 'sub2'))
|
|
return tmpdir
|
|
|
|
@unittest.skipUnless(find_executable('tar') and find_executable('gzip')
|
|
and ZLIB_SUPPORT,
|
|
'Need the tar, gzip and zlib command to run')
|
|
def test_tarfile_vs_tar(self):
|
|
tmpdir = self._create_files()
|
|
tmpdir2 = self.mkdtemp()
|
|
base_name = os.path.join(tmpdir2, 'archive')
|
|
old_dir = os.getcwd()
|
|
os.chdir(tmpdir)
|
|
try:
|
|
make_tarball(base_name, 'dist')
|
|
finally:
|
|
os.chdir(old_dir)
|
|
|
|
# check if the compressed tarball was created
|
|
tarball = base_name + '.tar.gz'
|
|
self.assertTrue(os.path.exists(tarball))
|
|
|
|
# now create another tarball using `tar`
|
|
tarball2 = os.path.join(tmpdir, 'archive2.tar.gz')
|
|
tar_cmd = ['tar', '-cf', 'archive2.tar', 'dist']
|
|
gzip_cmd = ['gzip', '-f', '-9', 'archive2.tar']
|
|
old_dir = os.getcwd()
|
|
os.chdir(tmpdir)
|
|
try:
|
|
spawn(tar_cmd)
|
|
spawn(gzip_cmd)
|
|
finally:
|
|
os.chdir(old_dir)
|
|
|
|
self.assertTrue(os.path.exists(tarball2))
|
|
# let's compare both tarballs
|
|
self.assertEqual(self._tarinfo(tarball), self._created_files)
|
|
self.assertEqual(self._tarinfo(tarball2), self._created_files)
|
|
|
|
# trying an uncompressed one
|
|
base_name = os.path.join(tmpdir2, 'archive')
|
|
old_dir = os.getcwd()
|
|
os.chdir(tmpdir)
|
|
try:
|
|
make_tarball(base_name, 'dist', compress=None)
|
|
finally:
|
|
os.chdir(old_dir)
|
|
tarball = base_name + '.tar'
|
|
self.assertTrue(os.path.exists(tarball))
|
|
|
|
# now for a dry_run
|
|
base_name = os.path.join(tmpdir2, 'archive')
|
|
old_dir = os.getcwd()
|
|
os.chdir(tmpdir)
|
|
try:
|
|
make_tarball(base_name, 'dist', compress=None, dry_run=True)
|
|
finally:
|
|
os.chdir(old_dir)
|
|
tarball = base_name + '.tar'
|
|
self.assertTrue(os.path.exists(tarball))
|
|
|
|
@unittest.skipUnless(find_executable('compress'),
|
|
'The compress program is required')
|
|
def test_compress_deprecated(self):
|
|
tmpdir = self._create_files()
|
|
base_name = os.path.join(self.mkdtemp(), 'archive')
|
|
|
|
# using compress and testing the PendingDeprecationWarning
|
|
old_dir = os.getcwd()
|
|
os.chdir(tmpdir)
|
|
try:
|
|
with check_warnings() as w:
|
|
warnings.simplefilter("always")
|
|
make_tarball(base_name, 'dist', compress='compress')
|
|
finally:
|
|
os.chdir(old_dir)
|
|
tarball = base_name + '.tar.Z'
|
|
self.assertTrue(os.path.exists(tarball))
|
|
self.assertEqual(len(w.warnings), 1)
|
|
|
|
# same test with dry_run
|
|
os.remove(tarball)
|
|
old_dir = os.getcwd()
|
|
os.chdir(tmpdir)
|
|
try:
|
|
with check_warnings() as w:
|
|
warnings.simplefilter("always")
|
|
make_tarball(base_name, 'dist', compress='compress',
|
|
dry_run=True)
|
|
finally:
|
|
os.chdir(old_dir)
|
|
self.assertFalse(os.path.exists(tarball))
|
|
self.assertEqual(len(w.warnings), 1)
|
|
|
|
@unittest.skipUnless(ZIP_SUPPORT and ZLIB_SUPPORT,
|
|
'Need zip and zlib support to run')
|
|
def test_make_zipfile(self):
|
|
# creating something to tar
|
|
tmpdir = self._create_files()
|
|
base_name = os.path.join(self.mkdtemp(), 'archive')
|
|
with change_cwd(tmpdir):
|
|
make_zipfile(base_name, 'dist')
|
|
|
|
# check if the compressed tarball was created
|
|
tarball = base_name + '.zip'
|
|
self.assertTrue(os.path.exists(tarball))
|
|
with zipfile.ZipFile(tarball) as zf:
|
|
self.assertEqual(sorted(zf.namelist()), self._zip_created_files)
|
|
|
|
@unittest.skipUnless(ZIP_SUPPORT, 'Need zip support to run')
|
|
def test_make_zipfile_no_zlib(self):
|
|
patch(self, archive_util.zipfile, 'zlib', None) # force zlib ImportError
|
|
|
|
called = []
|
|
zipfile_class = zipfile.ZipFile
|
|
def fake_zipfile(*a, **kw):
|
|
if kw.get('compression', None) == zipfile.ZIP_STORED:
|
|
called.append((a, kw))
|
|
return zipfile_class(*a, **kw)
|
|
|
|
patch(self, archive_util.zipfile, 'ZipFile', fake_zipfile)
|
|
|
|
# create something to tar and compress
|
|
tmpdir = self._create_files()
|
|
base_name = os.path.join(self.mkdtemp(), 'archive')
|
|
with change_cwd(tmpdir):
|
|
make_zipfile(base_name, 'dist')
|
|
|
|
tarball = base_name + '.zip'
|
|
self.assertEqual(called,
|
|
[((tarball, "w"), {'compression': zipfile.ZIP_STORED})])
|
|
self.assertTrue(os.path.exists(tarball))
|
|
with zipfile.ZipFile(tarball) as zf:
|
|
self.assertEqual(sorted(zf.namelist()), self._zip_created_files)
|
|
|
|
def test_check_archive_formats(self):
|
|
self.assertEqual(check_archive_formats(['gztar', 'xxx', 'zip']),
|
|
'xxx')
|
|
self.assertIsNone(check_archive_formats(['gztar', 'bztar', 'xztar',
|
|
'ztar', 'tar', 'zip']))
|
|
|
|
def test_make_archive(self):
|
|
tmpdir = self.mkdtemp()
|
|
base_name = os.path.join(tmpdir, 'archive')
|
|
self.assertRaises(ValueError, make_archive, base_name, 'xxx')
|
|
|
|
def test_make_archive_cwd(self):
|
|
current_dir = os.getcwd()
|
|
def _breaks(*args, **kw):
|
|
raise RuntimeError()
|
|
ARCHIVE_FORMATS['xxx'] = (_breaks, [], 'xxx file')
|
|
try:
|
|
try:
|
|
make_archive('xxx', 'xxx', root_dir=self.mkdtemp())
|
|
except:
|
|
pass
|
|
self.assertEqual(os.getcwd(), current_dir)
|
|
finally:
|
|
del ARCHIVE_FORMATS['xxx']
|
|
|
|
def test_make_archive_tar(self):
|
|
base_dir = self._create_files()
|
|
base_name = os.path.join(self.mkdtemp() , 'archive')
|
|
res = make_archive(base_name, 'tar', base_dir, 'dist')
|
|
self.assertTrue(os.path.exists(res))
|
|
self.assertEqual(os.path.basename(res), 'archive.tar')
|
|
self.assertEqual(self._tarinfo(res), self._created_files)
|
|
|
|
@unittest.skipUnless(ZLIB_SUPPORT, 'Need zlib support to run')
|
|
def test_make_archive_gztar(self):
|
|
base_dir = self._create_files()
|
|
base_name = os.path.join(self.mkdtemp() , 'archive')
|
|
res = make_archive(base_name, 'gztar', base_dir, 'dist')
|
|
self.assertTrue(os.path.exists(res))
|
|
self.assertEqual(os.path.basename(res), 'archive.tar.gz')
|
|
self.assertEqual(self._tarinfo(res), self._created_files)
|
|
|
|
@unittest.skipUnless(bz2, 'Need bz2 support to run')
|
|
def test_make_archive_bztar(self):
|
|
base_dir = self._create_files()
|
|
base_name = os.path.join(self.mkdtemp() , 'archive')
|
|
res = make_archive(base_name, 'bztar', base_dir, 'dist')
|
|
self.assertTrue(os.path.exists(res))
|
|
self.assertEqual(os.path.basename(res), 'archive.tar.bz2')
|
|
self.assertEqual(self._tarinfo(res), self._created_files)
|
|
|
|
@unittest.skipUnless(lzma, 'Need xz support to run')
|
|
def test_make_archive_xztar(self):
|
|
base_dir = self._create_files()
|
|
base_name = os.path.join(self.mkdtemp() , 'archive')
|
|
res = make_archive(base_name, 'xztar', base_dir, 'dist')
|
|
self.assertTrue(os.path.exists(res))
|
|
self.assertEqual(os.path.basename(res), 'archive.tar.xz')
|
|
self.assertEqual(self._tarinfo(res), self._created_files)
|
|
|
|
def test_make_archive_owner_group(self):
|
|
# testing make_archive with owner and group, with various combinations
|
|
# this works even if there's not gid/uid support
|
|
if UID_GID_SUPPORT:
|
|
group = grp.getgrgid(0)[0]
|
|
owner = pwd.getpwuid(0)[0]
|
|
else:
|
|
group = owner = 'root'
|
|
|
|
base_dir = self._create_files()
|
|
root_dir = self.mkdtemp()
|
|
base_name = os.path.join(self.mkdtemp() , 'archive')
|
|
res = make_archive(base_name, 'zip', root_dir, base_dir, owner=owner,
|
|
group=group)
|
|
self.assertTrue(os.path.exists(res))
|
|
|
|
res = make_archive(base_name, 'zip', root_dir, base_dir)
|
|
self.assertTrue(os.path.exists(res))
|
|
|
|
res = make_archive(base_name, 'tar', root_dir, base_dir,
|
|
owner=owner, group=group)
|
|
self.assertTrue(os.path.exists(res))
|
|
|
|
res = make_archive(base_name, 'tar', root_dir, base_dir,
|
|
owner='kjhkjhkjg', group='oihohoh')
|
|
self.assertTrue(os.path.exists(res))
|
|
|
|
@unittest.skipUnless(ZLIB_SUPPORT, "Requires zlib")
|
|
@unittest.skipUnless(UID_GID_SUPPORT, "Requires grp and pwd support")
|
|
def test_tarfile_root_owner(self):
|
|
tmpdir = self._create_files()
|
|
base_name = os.path.join(self.mkdtemp(), 'archive')
|
|
old_dir = os.getcwd()
|
|
os.chdir(tmpdir)
|
|
group = grp.getgrgid(0)[0]
|
|
owner = pwd.getpwuid(0)[0]
|
|
try:
|
|
archive_name = make_tarball(base_name, 'dist', compress=None,
|
|
owner=owner, group=group)
|
|
finally:
|
|
os.chdir(old_dir)
|
|
|
|
# check if the compressed tarball was created
|
|
self.assertTrue(os.path.exists(archive_name))
|
|
|
|
# now checks the rights
|
|
archive = tarfile.open(archive_name)
|
|
try:
|
|
for member in archive.getmembers():
|
|
self.assertEqual(member.uid, 0)
|
|
self.assertEqual(member.gid, 0)
|
|
finally:
|
|
archive.close()
|
|
|
|
def test_suite():
|
|
return unittest.makeSuite(ArchiveUtilTestCase)
|
|
|
|
if __name__ == "__main__":
|
|
run_unittest(test_suite())
|