The Tor Browser: python/mozbuild/mozpack/test/test_copier.py@6474c204b198 (annotated)

python/mozbuild/mozpack/test/test_copier.py@6474c204b198 (annotated)

python/mozbuild/mozpack/test/test_copier.py

Wed, 31 Dec 2014 06:09:35 +0100

author: Michael Schloh von Bennewitz <michael@schloh.com>
date: Wed, 31 Dec 2014 06:09:35 +0100
changeset 0: 6474c204b198
permissions: -rw-r--r--

Cloned upstream origin tor-browser at tor-browser-31.3.0esr-4.5-1-build1
revision ID fc1c9ff7c1b2defdbc039f12214767608f46423f for hacking purpose.

 # This Source Code Form is subject to the terms of the Mozilla Public
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 from mozpack.copier import (
     FileCopier,
     FilePurger,
     FileRegistry,
     Jarrer,
 )
 from mozpack.files import (
     GeneratedFile,
     ExistingFile,
 )
 from mozpack.mozjar import JarReader
 import mozpack.path
 import unittest
 import mozunit
 import os
 import stat
 from mozpack.errors import ErrorMessage
 from mozpack.test.test_files import (
     MockDest,
     MatchTestTemplate,
     TestWithTmpDir,
 )
 class TestFileRegistry(MatchTestTemplate, unittest.TestCase):
     def add(self, path):
         self.registry.add(path, GeneratedFile(path))
     def do_check(self, pattern, result):
         self.checked = True
         if result:
             self.assertTrue(self.registry.contains(pattern))
         else:
             self.assertFalse(self.registry.contains(pattern))
         self.assertEqual(self.registry.match(pattern), result)
     def test_file_registry(self):
         self.registry = FileRegistry()
         self.registry.add('foo', GeneratedFile('foo'))
         bar = GeneratedFile('bar')
         self.registry.add('bar', bar)
         self.assertEqual(self.registry.paths(), ['foo', 'bar'])
         self.assertEqual(self.registry['bar'], bar)
         self.assertRaises(ErrorMessage, self.registry.add, 'foo',
                           GeneratedFile('foo2'))
         self.assertRaises(ErrorMessage, self.registry.remove, 'qux')
         self.assertRaises(ErrorMessage, self.registry.add, 'foo/bar',
                           GeneratedFile('foobar'))
         self.assertRaises(ErrorMessage, self.registry.add, 'foo/bar/baz',
                           GeneratedFile('foobar'))
         self.assertEqual(self.registry.paths(), ['foo', 'bar'])
         self.registry.remove('foo')
         self.assertEqual(self.registry.paths(), ['bar'])
         self.registry.remove('bar')
         self.assertEqual(self.registry.paths(), [])
         self.prepare_match_test()
         self.do_match_test()
         self.assertTrue(self.checked)
         self.assertEqual(self.registry.paths(), [
             'bar',
             'foo/bar',
             'foo/baz',
             'foo/qux/1',
             'foo/qux/bar',
             'foo/qux/2/test',
             'foo/qux/2/test2',
         ])
         self.registry.remove('foo/qux')
         self.assertEqual(self.registry.paths(), ['bar', 'foo/bar', 'foo/baz'])
         self.registry.add('foo/qux', GeneratedFile('fooqux'))
         self.assertEqual(self.registry.paths(), ['bar', 'foo/bar', 'foo/baz',
                                                  'foo/qux'])
         self.registry.remove('foo/b*')
         self.assertEqual(self.registry.paths(), ['bar', 'foo/qux'])
         self.assertEqual([f for f, c in self.registry], ['bar', 'foo/qux'])
         self.assertEqual(len(self.registry), 2)
         self.add('foo/.foo')
         self.assertTrue(self.registry.contains('foo/.foo'))
     def test_registry_paths(self):
         self.registry = FileRegistry()
         # Can't add a file if it requires a directory in place of a
         # file we also require.
         self.registry.add('foo', GeneratedFile('foo'))
         self.assertRaises(ErrorMessage, self.registry.add, 'foo/bar',
                           GeneratedFile('foobar'))
         # Can't add a file if we already have a directory there.
         self.registry.add('bar/baz', GeneratedFile('barbaz'))
         self.assertRaises(ErrorMessage, self.registry.add, 'bar',
                           GeneratedFile('bar'))
         # Bump the count of things that require bar/ to 2.
         self.registry.add('bar/zot', GeneratedFile('barzot'))
         self.assertRaises(ErrorMessage, self.registry.add, 'bar',
                           GeneratedFile('bar'))
         # Drop the count of things that require bar/ to 1.
         self.registry.remove('bar/baz')
         self.assertRaises(ErrorMessage, self.registry.add, 'bar',
                           GeneratedFile('bar'))
         # Drop the count of things that require bar/ to 0.
         self.registry.remove('bar/zot')
         self.registry.add('bar/zot', GeneratedFile('barzot'))
     def test_required_directories(self):
         self.registry = FileRegistry()
         self.registry.add('foo', GeneratedFile('foo'))
         self.assertEqual(self.registry.required_directories(), set())
         self.registry.add('bar/baz', GeneratedFile('barbaz'))
         self.assertEqual(self.registry.required_directories(), {'bar'})
         self.registry.add('bar/zot', GeneratedFile('barzot'))
         self.assertEqual(self.registry.required_directories(), {'bar'})
         self.registry.add('bar/zap/zot', GeneratedFile('barzapzot'))
         self.assertEqual(self.registry.required_directories(), {'bar', 'bar/zap'})
         self.registry.remove('bar/zap/zot')
         self.assertEqual(self.registry.required_directories(), {'bar'})
         self.registry.remove('bar/baz')
         self.assertEqual(self.registry.required_directories(), {'bar'})
         self.registry.remove('bar/zot')
         self.assertEqual(self.registry.required_directories(), set())
         self.registry.add('x/y/z', GeneratedFile('xyz'))
         self.assertEqual(self.registry.required_directories(), {'x', 'x/y'})
 class TestFileCopier(TestWithTmpDir):
     def all_dirs(self, base):
         all_dirs = set()
         for root, dirs, files in os.walk(base):
             if not dirs:
                 all_dirs.add(mozpack.path.relpath(root, base))
         return all_dirs
     def all_files(self, base):
         all_files = set()
         for root, dirs, files in os.walk(base):
             for f in files:
                 all_files.add(
                     mozpack.path.join(mozpack.path.relpath(root, base), f))
         return all_files
     def test_file_copier(self):
         copier = FileCopier()
         copier.add('foo/bar', GeneratedFile('foobar'))
         copier.add('foo/qux', GeneratedFile('fooqux'))
         copier.add('foo/deep/nested/directory/file', GeneratedFile('fooz'))
         copier.add('bar', GeneratedFile('bar'))
         copier.add('qux/foo', GeneratedFile('quxfoo'))
         copier.add('qux/bar', GeneratedFile(''))
         result = copier.copy(self.tmpdir)
         self.assertEqual(self.all_files(self.tmpdir), set(copier.paths()))
         self.assertEqual(self.all_dirs(self.tmpdir),
                          set(['foo/deep/nested/directory', 'qux']))
         self.assertEqual(result.updated_files, set(self.tmppath(p) for p in
             self.all_files(self.tmpdir)))
         self.assertEqual(result.existing_files, set())
         self.assertEqual(result.removed_files, set())
         self.assertEqual(result.removed_directories, set())
         copier.remove('foo')
         copier.add('test', GeneratedFile('test'))
         result = copier.copy(self.tmpdir)
         self.assertEqual(self.all_files(self.tmpdir), set(copier.paths()))
         self.assertEqual(self.all_dirs(self.tmpdir), set(['qux']))
         self.assertEqual(result.removed_files, set(self.tmppath(p) for p in
             ('foo/bar', 'foo/qux', 'foo/deep/nested/directory/file')))
     def test_symlink_directory_replaced(self):
         """Directory symlinks in destination are replaced if they need to be
         real directories."""
         if not self.symlink_supported:
             return
         dest = self.tmppath('dest')
         copier = FileCopier()
         copier.add('foo/bar/baz', GeneratedFile('foobarbaz'))
         os.makedirs(self.tmppath('dest/foo'))
         dummy = self.tmppath('dummy')
         os.mkdir(dummy)
         link = self.tmppath('dest/foo/bar')
         os.symlink(dummy, link)
         result = copier.copy(dest)
         st = os.lstat(link)
         self.assertFalse(stat.S_ISLNK(st.st_mode))
         self.assertTrue(stat.S_ISDIR(st.st_mode))
         self.assertEqual(self.all_files(dest), set(copier.paths()))
         self.assertEqual(result.removed_directories, set())
         self.assertEqual(len(result.updated_files), 1)
     def test_remove_unaccounted_directory_symlinks(self):
         """Directory symlinks in destination that are not in the way are
         deleted according to remove_unaccounted and
         remove_all_directory_symlinks.
         """
         if not self.symlink_supported:
             return
         dest = self.tmppath('dest')
         copier = FileCopier()
         copier.add('foo/bar/baz', GeneratedFile('foobarbaz'))
         os.makedirs(self.tmppath('dest/foo'))
         dummy = self.tmppath('dummy')
         os.mkdir(dummy)
         os.mkdir(self.tmppath('dest/zot'))
         link = self.tmppath('dest/zot/zap')
         os.symlink(dummy, link)
         # If not remove_unaccounted but remove_empty_directories, then
         # the symlinked directory remains (as does its containing
         # directory).
         result = copier.copy(dest, remove_unaccounted=False,
             remove_empty_directories=True,
             remove_all_directory_symlinks=False)
         st = os.lstat(link)
         self.assertTrue(stat.S_ISLNK(st.st_mode))
         self.assertFalse(stat.S_ISDIR(st.st_mode))
         self.assertEqual(self.all_files(dest), set(copier.paths()))
         self.assertEqual(self.all_dirs(dest), set(['foo/bar']))
         self.assertEqual(result.removed_directories, set())
         self.assertEqual(len(result.updated_files), 1)
         # If remove_unaccounted but not remove_empty_directories, then
         # only the symlinked directory is removed.
         result = copier.copy(dest, remove_unaccounted=True,
             remove_empty_directories=False,
             remove_all_directory_symlinks=False)
         st = os.lstat(self.tmppath('dest/zot'))
         self.assertFalse(stat.S_ISLNK(st.st_mode))
         self.assertTrue(stat.S_ISDIR(st.st_mode))
         self.assertEqual(result.removed_files, set([link]))
         self.assertEqual(result.removed_directories, set())
         self.assertEqual(self.all_files(dest), set(copier.paths()))
         self.assertEqual(self.all_dirs(dest), set(['foo/bar', 'zot']))
         # If remove_unaccounted and remove_empty_directories, then
         # both the symlink and its containing directory are removed.
         link = self.tmppath('dest/zot/zap')
         os.symlink(dummy, link)
         result = copier.copy(dest, remove_unaccounted=True,
             remove_empty_directories=True,
             remove_all_directory_symlinks=False)
         self.assertEqual(result.removed_files, set([link]))
         self.assertEqual(result.removed_directories, set([self.tmppath('dest/zot')]))
         self.assertEqual(self.all_files(dest), set(copier.paths()))
         self.assertEqual(self.all_dirs(dest), set(['foo/bar']))
     def test_permissions(self):
         """Ensure files without write permission can be deleted."""
         with open(self.tmppath('dummy'), 'a'):
             pass
         p = self.tmppath('no_perms')
         with open(p, 'a'):
             pass
         # Make file and directory unwritable. Reminder: making a directory
         # unwritable prevents modifications (including deletes) from the list
         # of files in that directory.
         os.chmod(p, 0400)
         os.chmod(self.tmpdir, 0400)
         copier = FileCopier()
         copier.add('dummy', GeneratedFile('content'))
         result = copier.copy(self.tmpdir)
         self.assertEqual(result.removed_files_count, 1)
         self.assertFalse(os.path.exists(p))
     def test_no_remove(self):
         copier = FileCopier()
         copier.add('foo', GeneratedFile('foo'))
         with open(self.tmppath('bar'), 'a'):
             pass
         os.mkdir(self.tmppath('emptydir'))
         d = self.tmppath('populateddir')
         os.mkdir(d)
         with open(self.tmppath('populateddir/foo'), 'a'):
             pass
         result = copier.copy(self.tmpdir, remove_unaccounted=False)
         self.assertEqual(self.all_files(self.tmpdir), set(['foo', 'bar',
             'populateddir/foo']))
         self.assertEqual(self.all_dirs(self.tmpdir), set(['populateddir']))
         self.assertEqual(result.removed_files, set())
         self.assertEqual(result.removed_directories,
             set([self.tmppath('emptydir')]))
     def test_no_remove_empty_directories(self):
         copier = FileCopier()
         copier.add('foo', GeneratedFile('foo'))
         with open(self.tmppath('bar'), 'a'):
             pass
         os.mkdir(self.tmppath('emptydir'))
         d = self.tmppath('populateddir')
         os.mkdir(d)
         with open(self.tmppath('populateddir/foo'), 'a'):
             pass
         result = copier.copy(self.tmpdir, remove_unaccounted=False,
             remove_empty_directories=False)
         self.assertEqual(self.all_files(self.tmpdir), set(['foo', 'bar',
             'populateddir/foo']))
         self.assertEqual(self.all_dirs(self.tmpdir), set(['emptydir',
             'populateddir']))
         self.assertEqual(result.removed_files, set())
         self.assertEqual(result.removed_directories, set())
     def test_optional_exists_creates_unneeded_directory(self):
         """Demonstrate that a directory not strictly required, but specified
         as the path to an optional file, will be unnecessarily created.
         This behaviour is wrong; fixing it is tracked by Bug 972432;
         and this test exists to guard against unexpected changes in
         behaviour.
         """
         dest = self.tmppath('dest')
         copier = FileCopier()
         copier.add('foo/bar', ExistingFile(required=False))
         result = copier.copy(dest)
         st = os.lstat(self.tmppath('dest/foo'))
         self.assertFalse(stat.S_ISLNK(st.st_mode))
         self.assertTrue(stat.S_ISDIR(st.st_mode))
         # What's worse, we have no record that dest was created.
         self.assertEquals(len(result.updated_files), 0)
         # But we do have an erroneous record of an optional file
         # existing when it does not.
         self.assertIn(self.tmppath('dest/foo/bar'), result.existing_files)
 class TestFilePurger(TestWithTmpDir):
     def test_file_purger(self):
         existing = os.path.join(self.tmpdir, 'existing')
         extra = os.path.join(self.tmpdir, 'extra')
         empty_dir = os.path.join(self.tmpdir, 'dir')
         with open(existing, 'a'):
             pass
         with open(extra, 'a'):
             pass
         os.mkdir(empty_dir)
         with open(os.path.join(empty_dir, 'foo'), 'a'):
             pass
         self.assertTrue(os.path.exists(existing))
         self.assertTrue(os.path.exists(extra))
         purger = FilePurger()
         purger.add('existing')
         result = purger.purge(self.tmpdir)
         self.assertEqual(result.removed_files, set(self.tmppath(p) for p in
             ('extra', 'dir/foo')))
         self.assertEqual(result.removed_files_count, 2)
         self.assertEqual(result.removed_directories_count, 1)
         self.assertTrue(os.path.exists(existing))
         self.assertFalse(os.path.exists(extra))
         self.assertFalse(os.path.exists(empty_dir))
 class TestJarrer(unittest.TestCase):
     def check_jar(self, dest, copier):
         jar = JarReader(fileobj=dest)
         self.assertEqual([f.filename for f in jar], copier.paths())
         for f in jar:
             self.assertEqual(f.uncompressed_data.read(),
                              copier[f.filename].content)
     def test_jarrer(self):
         copier = Jarrer()
         copier.add('foo/bar', GeneratedFile('foobar'))
         copier.add('foo/qux', GeneratedFile('fooqux'))
         copier.add('foo/deep/nested/directory/file', GeneratedFile('fooz'))
         copier.add('bar', GeneratedFile('bar'))
         copier.add('qux/foo', GeneratedFile('quxfoo'))
         copier.add('qux/bar', GeneratedFile(''))
         dest = MockDest()
         copier.copy(dest)
         self.check_jar(dest, copier)
         copier.remove('foo')
         copier.add('test', GeneratedFile('test'))
         copier.copy(dest)
         self.check_jar(dest, copier)
         copier.remove('test')
         copier.add('test', GeneratedFile('replaced-content'))
         copier.copy(dest)
         self.check_jar(dest, copier)
         copier.copy(dest)
         self.check_jar(dest, copier)
         preloaded = ['qux/bar', 'bar']
         copier.preload(preloaded)
         copier.copy(dest)
         dest.seek(0)
         jar = JarReader(fileobj=dest)
         self.assertEqual([f.filename for f in jar], preloaded +
                          [p for p in copier.paths() if not p in preloaded])
         self.assertEqual(jar.last_preloaded, preloaded[-1])
 if __name__ == '__main__':
     mozunit.main()

The Tor Browser / annotate

python/mozbuild/mozpack/test/test_copier.py@6474c204b198 (annotated)

python/mozbuild/mozpack/test/test_copier.py