| # Copyright 2015 The Chromium Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import os |
| import shutil |
| import stat |
| import sys |
| import tempfile |
| import unittest |
| import uuid |
| import zipfile |
| |
| import mock |
| |
| from dependency_manager import dependency_manager_util |
| from dependency_manager import exceptions |
| |
| |
| class DependencyManagerUtilTest(unittest.TestCase): |
| # This class intentionally uses actual file I/O to test real system behavior. |
| |
| def setUp(self): |
| self.tmp_dir = os.path.abspath(tempfile.mkdtemp(prefix='telemetry')) |
| self.sub_dir = os.path.join(self.tmp_dir, 'sub_dir') |
| os.mkdir(self.sub_dir) |
| |
| self.read_only_path = (os.path.join(self.tmp_dir, 'read_only')) |
| with open(self.read_only_path, 'w+') as read_file: |
| read_file.write('Read-only file') |
| os.chmod(self.read_only_path, stat.S_IRUSR) |
| |
| self.writable_path = (os.path.join(self.tmp_dir, 'writable')) |
| with open(self.writable_path, 'w+') as writable_file: |
| writable_file.write('Writable file') |
| os.chmod(self.writable_path, stat.S_IRUSR | stat.S_IWUSR) |
| |
| self.executable_path = (os.path.join(self.tmp_dir, 'executable')) |
| with open(self.executable_path, 'w+') as executable_file: |
| executable_file.write('Executable file') |
| os.chmod(self.executable_path, stat.S_IRWXU) |
| |
| self.sub_read_only_path = (os.path.join(self.sub_dir, 'read_only')) |
| with open(self.sub_read_only_path, 'w+') as read_file: |
| read_file.write('Read-only sub file') |
| os.chmod(self.sub_read_only_path, stat.S_IRUSR) |
| |
| self.sub_writable_path = (os.path.join(self.sub_dir, 'writable')) |
| with open(self.sub_writable_path, 'w+') as writable_file: |
| writable_file.write('Writable sub file') |
| os.chmod(self.sub_writable_path, stat.S_IRUSR | stat.S_IWUSR) |
| |
| self.sub_executable_path = (os.path.join(self.sub_dir, 'executable')) |
| with open(self.sub_executable_path, 'w+') as executable_file: |
| executable_file.write('Executable sub file') |
| os.chmod(self.sub_executable_path, stat.S_IRWXU) |
| |
| self.AssertExpectedDirFiles(self.tmp_dir) |
| self.archive_path = self.CreateZipArchiveFromDir(self.tmp_dir) |
| |
| def tearDown(self): |
| if os.path.isdir(self.tmp_dir): |
| dependency_manager_util.RemoveDir(self.tmp_dir) |
| if os.path.isfile(self.archive_path): |
| os.remove(self.archive_path) |
| |
| def AssertExpectedDirFiles(self, top_dir): |
| sub_dir = os.path.join(top_dir, 'sub_dir') |
| read_only_path = (os.path.join(top_dir, 'read_only')) |
| writable_path = (os.path.join(top_dir, 'writable')) |
| executable_path = (os.path.join(top_dir, 'executable')) |
| sub_read_only_path = (os.path.join(sub_dir, 'read_only')) |
| sub_writable_path = (os.path.join(sub_dir, 'writable')) |
| sub_executable_path = (os.path.join(sub_dir, 'executable')) |
| # assert contents as expected |
| self.assertTrue(os.path.isdir(top_dir)) |
| self.assertTrue(os.path.isdir(sub_dir)) |
| self.assertTrue(os.path.isfile(read_only_path)) |
| self.assertTrue(os.path.isfile(writable_path)) |
| self.assertTrue(os.path.isfile(executable_path)) |
| self.assertTrue(os.path.isfile(sub_read_only_path)) |
| self.assertTrue(os.path.isfile(sub_writable_path)) |
| self.assertTrue(os.path.isfile(sub_executable_path)) |
| |
| # assert permissions as expected |
| self.assertTrue( |
| stat.S_IRUSR & stat.S_IMODE(os.stat(read_only_path).st_mode)) |
| self.assertTrue( |
| stat.S_IRUSR & stat.S_IMODE(os.stat(sub_read_only_path).st_mode)) |
| self.assertTrue( |
| stat.S_IRUSR & stat.S_IMODE(os.stat(writable_path).st_mode)) |
| self.assertTrue( |
| stat.S_IWUSR & stat.S_IMODE(os.stat(writable_path).st_mode)) |
| self.assertTrue( |
| stat.S_IRUSR & stat.S_IMODE(os.stat(sub_writable_path).st_mode)) |
| self.assertTrue( |
| stat.S_IWUSR & stat.S_IMODE(os.stat(sub_writable_path).st_mode)) |
| if not sys.platform.startswith('win'): |
| self.assertEqual( |
| stat.S_IRWXU, |
| stat.S_IRWXU & stat.S_IMODE(os.stat(executable_path).st_mode)) |
| self.assertEqual( |
| stat.S_IRWXU, |
| stat.S_IRWXU & stat.S_IMODE(os.stat(sub_executable_path).st_mode)) |
| |
| def CreateZipArchiveFromDir(self, dir_path): |
| try: |
| base_path = os.path.join(tempfile.gettempdir(), str(uuid.uuid4())) |
| archive_path = shutil.make_archive(base_path, 'zip', dir_path) |
| self.assertTrue(os.path.exists(archive_path)) |
| self.assertTrue(zipfile.is_zipfile(archive_path)) |
| except: |
| if os.path.isfile(archive_path): |
| os.remove(archive_path) |
| raise |
| return archive_path |
| |
| def testRemoveDirWithSubDir(self): |
| dependency_manager_util.RemoveDir(self.tmp_dir) |
| |
| self.assertFalse(os.path.exists(self.tmp_dir)) |
| self.assertFalse(os.path.exists(self.sub_dir)) |
| self.assertFalse(os.path.exists(self.read_only_path)) |
| self.assertFalse(os.path.exists(self.writable_path)) |
| self.assertFalse(os.path.isfile(self.executable_path)) |
| self.assertFalse(os.path.exists(self.sub_read_only_path)) |
| self.assertFalse(os.path.exists(self.sub_writable_path)) |
| self.assertFalse(os.path.isfile(self.sub_executable_path)) |
| |
| def testUnzipFile(self): |
| self.AssertExpectedDirFiles(self.tmp_dir) |
| unzip_path = os.path.join(tempfile.gettempdir(), str(uuid.uuid4())) |
| dependency_manager_util.UnzipArchive(self.archive_path, unzip_path) |
| self.AssertExpectedDirFiles(unzip_path) |
| self.AssertExpectedDirFiles(self.tmp_dir) |
| dependency_manager_util.RemoveDir(unzip_path) |
| |
| def testUnzipFileContainingLongPath(self): |
| try: |
| dir_path = self.tmp_dir |
| if sys.platform.startswith('win'): |
| dir_path = u'\\\\?\\' + dir_path |
| |
| archive_suffix = '' |
| # 260 is the Windows API path length limit. |
| while len(archive_suffix) < 260: |
| archive_suffix = os.path.join(archive_suffix, 'really') |
| contents_dir_path = os.path.join(dir_path, archive_suffix) |
| os.makedirs(contents_dir_path) |
| filename = os.path.join(contents_dir_path, 'longpath.txt') |
| open(filename, 'a').close() |
| |
| base_path = os.path.join(tempfile.gettempdir(), str(uuid.uuid4())) |
| archive_path = shutil.make_archive(base_path, 'zip', dir_path) |
| self.assertTrue(os.path.exists(archive_path)) |
| self.assertTrue(zipfile.is_zipfile(archive_path)) |
| except: |
| if os.path.isfile(archive_path): |
| os.remove(archive_path) |
| raise |
| |
| unzip_path = os.path.join(tempfile.gettempdir(), str(uuid.uuid4())) |
| dependency_manager_util.UnzipArchive(archive_path, unzip_path) |
| dependency_manager_util.RemoveDir(unzip_path) |
| |
| def testUnzipFileFailure(self): |
| # zipfile is not used on MacOS. See crbug.com/700097. |
| if sys.platform.startswith('darwin'): |
| return |
| unzip_path = os.path.join(tempfile.gettempdir(), str(uuid.uuid4())) |
| self.assertFalse(os.path.exists(unzip_path)) |
| with mock.patch( |
| 'dependency_manager.dependency_manager_util.zipfile.ZipFile.extractall' # pylint: disable=line-too-long |
| ) as zipfile_mock: |
| zipfile_mock.side_effect = IOError |
| self.assertRaises( |
| IOError, dependency_manager_util.UnzipArchive, self.archive_path, |
| unzip_path) |
| self.AssertExpectedDirFiles(self.tmp_dir) |
| self.assertFalse(os.path.exists(unzip_path)) |
| |
| def testVerifySafeArchivePasses(self): |
| with zipfile.ZipFile(self.archive_path) as archive: |
| dependency_manager_util.VerifySafeArchive(archive) |
| |
| def testVerifySafeArchiveFailsOnRelativePathWithPardir(self): |
| tmp_file = tempfile.NamedTemporaryFile(delete=False) |
| tmp_file_name = tmp_file.name |
| tmp_file.write('Bad file!') |
| tmp_file.close() |
| with zipfile.ZipFile(self.archive_path, 'w') as archive: |
| archive.write(tmp_file_name, '../../foo') |
| self.assertRaises( |
| exceptions.ArchiveError, dependency_manager_util.VerifySafeArchive, |
| archive) |
| |
