michael@0: # This Source Code Form is subject to the terms of the Mozilla Public
michael@0: # License, v. 2.0. If a copy of the MPL was not distributed with this
michael@0: # file, You can obtain one at http://mozilla.org/MPL/2.0/.
michael@0: 
michael@0: from __future__ import unicode_literals
michael@0: 
michael@0: import os
michael@0: 
michael@0: import mozunit
michael@0: 
michael@0: from mozpack.copier import (
michael@0:     FileCopier,
michael@0:     FileRegistry,
michael@0: )
michael@0: from mozpack.manifests import (
michael@0:     InstallManifest,
michael@0: )
michael@0: from mozpack.test.test_files import TestWithTmpDir
michael@0: 
michael@0: 
michael@0: class TestInstallManifest(TestWithTmpDir):
michael@0:     def test_construct(self):
michael@0:         m = InstallManifest()
michael@0:         self.assertEqual(len(m), 0)
michael@0: 
michael@0:     def test_adds(self):
michael@0:         m = InstallManifest()
michael@0:         m.add_symlink('s_source', 's_dest')
michael@0:         m.add_copy('c_source', 'c_dest')
michael@0:         m.add_required_exists('e_dest')
michael@0:         m.add_optional_exists('o_dest')
michael@0:         m.add_pattern_symlink('ps_base', 'ps/*', 'ps_dest')
michael@0:         m.add_pattern_copy('pc_base', 'pc/**', 'pc_dest')
michael@0:         m.add_preprocess('p_source', 'p_dest', 'p_source.pp')
michael@0: 
michael@0:         self.assertEqual(len(m), 7)
michael@0:         self.assertIn('s_dest', m)
michael@0:         self.assertIn('c_dest', m)
michael@0:         self.assertIn('p_dest', m)
michael@0:         self.assertIn('e_dest', m)
michael@0:         self.assertIn('o_dest', m)
michael@0: 
michael@0:         with self.assertRaises(ValueError):
michael@0:             m.add_symlink('s_other', 's_dest')
michael@0: 
michael@0:         with self.assertRaises(ValueError):
michael@0:             m.add_copy('c_other', 'c_dest')
michael@0: 
michael@0:         with self.assertRaises(ValueError):
michael@0:             m.add_preprocess('p_other', 'p_dest', 'p_other.pp')
michael@0: 
michael@0:         with self.assertRaises(ValueError):
michael@0:             m.add_required_exists('e_dest')
michael@0: 
michael@0:         with self.assertRaises(ValueError):
michael@0:             m.add_optional_exists('o_dest')
michael@0: 
michael@0:         with self.assertRaises(ValueError):
michael@0:             m.add_pattern_symlink('ps_base', 'ps/*', 'ps_dest')
michael@0: 
michael@0:         with self.assertRaises(ValueError):
michael@0:             m.add_pattern_copy('pc_base', 'pc/**', 'pc_dest')
michael@0: 
michael@0:     def _get_test_manifest(self):
michael@0:         m = InstallManifest()
michael@0:         m.add_symlink(self.tmppath('s_source'), 's_dest')
michael@0:         m.add_copy(self.tmppath('c_source'), 'c_dest')
michael@0:         m.add_preprocess(self.tmppath('p_source'), 'p_dest', self.tmppath('p_source.pp'), '#', {'FOO':'BAR', 'BAZ':'QUX'})
michael@0:         m.add_required_exists('e_dest')
michael@0:         m.add_optional_exists('o_dest')
michael@0:         m.add_pattern_symlink('ps_base', '*', 'ps_dest')
michael@0:         m.add_pattern_copy('pc_base', '**', 'pc_dest')
michael@0: 
michael@0:         return m
michael@0: 
michael@0:     def test_serialization(self):
michael@0:         m = self._get_test_manifest()
michael@0: 
michael@0:         p = self.tmppath('m')
michael@0:         m.write(path=p)
michael@0:         self.assertTrue(os.path.isfile(p))
michael@0: 
michael@0:         with open(p, 'rb') as fh:
michael@0:             c = fh.read()
michael@0: 
michael@0:         self.assertEqual(c.count('\n'), 8)
michael@0: 
michael@0:         lines = c.splitlines()
michael@0:         self.assertEqual(len(lines), 8)
michael@0: 
michael@0:         self.assertEqual(lines[0], '4')
michael@0: 
michael@0:         m2 = InstallManifest(path=p)
michael@0:         self.assertEqual(m, m2)
michael@0:         p2 = self.tmppath('m2')
michael@0:         m2.write(path=p2)
michael@0: 
michael@0:         with open(p2, 'rb') as fh:
michael@0:             c2 = fh.read()
michael@0: 
michael@0:         self.assertEqual(c, c2)
michael@0: 
michael@0:     def test_populate_registry(self):
michael@0:         m = self._get_test_manifest()
michael@0:         r = FileRegistry()
michael@0:         m.populate_registry(r)
michael@0: 
michael@0:         self.assertEqual(len(r), 5)
michael@0:         self.assertEqual(r.paths(), ['c_dest', 'e_dest', 'o_dest', 'p_dest', 's_dest'])
michael@0: 
michael@0:     def test_pattern_expansion(self):
michael@0:         source = self.tmppath('source')
michael@0:         os.mkdir(source)
michael@0:         os.mkdir('%s/base' % source)
michael@0:         os.mkdir('%s/base/foo' % source)
michael@0: 
michael@0:         with open('%s/base/foo/file1' % source, 'a'):
michael@0:             pass
michael@0: 
michael@0:         with open('%s/base/foo/file2' % source, 'a'):
michael@0:             pass
michael@0: 
michael@0:         m = InstallManifest()
michael@0:         m.add_pattern_symlink('%s/base' % source, '**', 'dest')
michael@0: 
michael@0:         c = FileCopier()
michael@0:         m.populate_registry(c)
michael@0:         self.assertEqual(c.paths(), ['dest/foo/file1', 'dest/foo/file2'])
michael@0: 
michael@0:     def test_or(self):
michael@0:         m1 = self._get_test_manifest()
michael@0:         orig_length = len(m1)
michael@0:         m2 = InstallManifest()
michael@0:         m2.add_symlink('s_source2', 's_dest2')
michael@0:         m2.add_copy('c_source2', 'c_dest2')
michael@0: 
michael@0:         m1 |= m2
michael@0: 
michael@0:         self.assertEqual(len(m2), 2)
michael@0:         self.assertEqual(len(m1), orig_length + 2)
michael@0: 
michael@0:         self.assertIn('s_dest2', m1)
michael@0:         self.assertIn('c_dest2', m1)
michael@0: 
michael@0:     def test_copier_application(self):
michael@0:         dest = self.tmppath('dest')
michael@0:         os.mkdir(dest)
michael@0: 
michael@0:         to_delete = self.tmppath('dest/to_delete')
michael@0:         with open(to_delete, 'a'):
michael@0:             pass
michael@0: 
michael@0:         with open(self.tmppath('s_source'), 'wt') as fh:
michael@0:             fh.write('symlink!')
michael@0: 
michael@0:         with open(self.tmppath('c_source'), 'wt') as fh:
michael@0:             fh.write('copy!')
michael@0: 
michael@0:         with open(self.tmppath('p_source'), 'wt') as fh:
michael@0:             fh.write('#define FOO 1\npreprocess!')
michael@0: 
michael@0:         with open(self.tmppath('dest/e_dest'), 'a'):
michael@0:             pass
michael@0: 
michael@0:         with open(self.tmppath('dest/o_dest'), 'a'):
michael@0:             pass
michael@0: 
michael@0:         m = self._get_test_manifest()
michael@0:         c = FileCopier()
michael@0:         m.populate_registry(c)
michael@0:         result = c.copy(dest)
michael@0: 
michael@0:         self.assertTrue(os.path.exists(self.tmppath('dest/s_dest')))
michael@0:         self.assertTrue(os.path.exists(self.tmppath('dest/c_dest')))
michael@0:         self.assertTrue(os.path.exists(self.tmppath('dest/p_dest')))
michael@0:         self.assertTrue(os.path.exists(self.tmppath('dest/e_dest')))
michael@0:         self.assertTrue(os.path.exists(self.tmppath('dest/o_dest')))
michael@0:         self.assertFalse(os.path.exists(to_delete))
michael@0: 
michael@0:         with open(self.tmppath('dest/s_dest'), 'rt') as fh:
michael@0:             self.assertEqual(fh.read(), 'symlink!')
michael@0: 
michael@0:         with open(self.tmppath('dest/c_dest'), 'rt') as fh:
michael@0:             self.assertEqual(fh.read(), 'copy!')
michael@0: 
michael@0:         with open(self.tmppath('dest/p_dest'), 'rt') as fh:
michael@0:             self.assertEqual(fh.read(), 'preprocess!')
michael@0: 
michael@0:         self.assertEqual(result.updated_files, set(self.tmppath(p) for p in (
michael@0:             'dest/s_dest', 'dest/c_dest', 'dest/p_dest')))
michael@0:         self.assertEqual(result.existing_files,
michael@0:             set([self.tmppath('dest/e_dest'), self.tmppath('dest/o_dest')]))
michael@0:         self.assertEqual(result.removed_files, {to_delete})
michael@0:         self.assertEqual(result.removed_directories, set())
michael@0: 
michael@0:     def test_preprocessor(self):
michael@0:         manifest = self.tmppath('m')
michael@0:         deps = self.tmppath('m.pp')
michael@0:         dest = self.tmppath('dest')
michael@0:         include = self.tmppath('p_incl')
michael@0: 
michael@0:         with open(include, 'wt') as fh:
michael@0:             fh.write('#define INCL\n')
michael@0:         time = os.path.getmtime(include) - 3
michael@0:         os.utime(include, (time, time))
michael@0: 
michael@0:         with open(self.tmppath('p_source'), 'wt') as fh:
michael@0:             fh.write('#ifdef FOO\n#if BAZ == QUX\nPASS1\n#endif\n#endif\n')
michael@0:             fh.write('#ifdef DEPTEST\nPASS2\n#endif\n')
michael@0:             fh.write('#include p_incl\n#ifdef INCLTEST\nPASS3\n#endif\n')
michael@0:         time = os.path.getmtime(self.tmppath('p_source')) - 3
michael@0:         os.utime(self.tmppath('p_source'), (time, time))
michael@0: 
michael@0:         # Create and write a manifest with the preprocessed file, then apply it.
michael@0:         # This should write out our preprocessed file.
michael@0:         m = InstallManifest()
michael@0:         m.add_preprocess(self.tmppath('p_source'), 'p_dest', deps, '#', {'FOO':'BAR', 'BAZ':'QUX'})
michael@0:         m.write(path=manifest)
michael@0: 
michael@0:         m = InstallManifest(path=manifest)
michael@0:         c = FileCopier()
michael@0:         m.populate_registry(c)
michael@0:         c.copy(dest)
michael@0: 
michael@0:         self.assertTrue(os.path.exists(self.tmppath('dest/p_dest')))
michael@0: 
michael@0:         with open(self.tmppath('dest/p_dest'), 'rt') as fh:
michael@0:             self.assertEqual(fh.read(), 'PASS1\n')
michael@0: 
michael@0:         # Create a second manifest with the preprocessed file, then apply it.
michael@0:         # Since this manifest does not exist on the disk, there should not be a
michael@0:         # dependency on it, and the preprocessed file should not be modified.
michael@0:         m2 = InstallManifest()
michael@0:         m2.add_preprocess(self.tmppath('p_source'), 'p_dest', deps, '#', {'DEPTEST':True})
michael@0:         c = FileCopier()
michael@0:         m2.populate_registry(c)
michael@0:         result = c.copy(dest)
michael@0: 
michael@0:         self.assertFalse(self.tmppath('dest/p_dest') in result.updated_files)
michael@0:         self.assertTrue(self.tmppath('dest/p_dest') in result.existing_files)
michael@0: 
michael@0:         # Write out the second manifest, then load it back in from the disk.
michael@0:         # This should add the dependency on the manifest file, so our
michael@0:         # preprocessed file should be regenerated with the new defines.
michael@0:         # We also set the mtime on the destination file back, so it will be
michael@0:         # older than the manifest file.
michael@0:         m2.write(path=manifest)
michael@0:         time = os.path.getmtime(manifest) - 1
michael@0:         os.utime(self.tmppath('dest/p_dest'), (time, time))
michael@0:         m2 = InstallManifest(path=manifest)
michael@0:         c = FileCopier()
michael@0:         m2.populate_registry(c)
michael@0:         self.assertTrue(c.copy(dest))
michael@0: 
michael@0:         with open(self.tmppath('dest/p_dest'), 'rt') as fh:
michael@0:             self.assertEqual(fh.read(), 'PASS2\n')
michael@0: 
michael@0:         # Set the time on the manifest back, so it won't be picked up as
michael@0:         # modified in the next test
michael@0:         time = os.path.getmtime(manifest) - 1
michael@0:         os.utime(manifest, (time, time))
michael@0: 
michael@0:         # Update the contents of a file included by the source file. This should
michael@0:         # cause the destination to be regenerated.
michael@0:         with open(include, 'wt') as fh:
michael@0:             fh.write('#define INCLTEST\n')
michael@0: 
michael@0:         time = os.path.getmtime(include) - 1
michael@0:         os.utime(self.tmppath('dest/p_dest'), (time, time))
michael@0:         c = FileCopier()
michael@0:         m2.populate_registry(c)
michael@0:         self.assertTrue(c.copy(dest))
michael@0: 
michael@0:         with open(self.tmppath('dest/p_dest'), 'rt') as fh:
michael@0:             self.assertEqual(fh.read(), 'PASS2\nPASS3\n')
michael@0: 
michael@0:     def test_preprocessor_dependencies(self):
michael@0:         manifest = self.tmppath('m')
michael@0:         deps = self.tmppath('m.pp')
michael@0:         dest = self.tmppath('dest')
michael@0:         source = self.tmppath('p_source')
michael@0:         destfile = self.tmppath('dest/p_dest')
michael@0:         include = self.tmppath('p_incl')
michael@0:         os.mkdir(dest)
michael@0: 
michael@0:         with open(source, 'wt') as fh:
michael@0:             fh.write('#define SRC\nSOURCE\n')
michael@0:         time = os.path.getmtime(source) - 3
michael@0:         os.utime(source, (time, time))
michael@0: 
michael@0:         with open(include, 'wt') as fh:
michael@0:             fh.write('INCLUDE\n')
michael@0:         time = os.path.getmtime(source) - 3
michael@0:         os.utime(include, (time, time))
michael@0: 
michael@0:         # Create and write a manifest with the preprocessed file.
michael@0:         m = InstallManifest()
michael@0:         m.add_preprocess(source, 'p_dest', deps, '#', {'FOO':'BAR', 'BAZ':'QUX'})
michael@0:         m.write(path=manifest)
michael@0: 
michael@0:         time = os.path.getmtime(source) - 5
michael@0:         os.utime(manifest, (time, time))
michael@0: 
michael@0:         # Now read the manifest back in, and apply it. This should write out
michael@0:         # our preprocessed file.
michael@0:         m = InstallManifest(path=manifest)
michael@0:         c = FileCopier()
michael@0:         m.populate_registry(c)
michael@0:         self.assertTrue(c.copy(dest))
michael@0: 
michael@0:         with open(destfile, 'rt') as fh:
michael@0:             self.assertEqual(fh.read(), 'SOURCE\n')
michael@0: 
michael@0:         # Next, modify the source to #INCLUDE another file.
michael@0:         with open(source, 'wt') as fh:
michael@0:             fh.write('SOURCE\n#include p_incl\n')
michael@0:         time = os.path.getmtime(source) - 1
michael@0:         os.utime(destfile, (time, time))
michael@0: 
michael@0:         # Apply the manifest, and confirm that it also reads the newly included
michael@0:         # file.
michael@0:         m = InstallManifest(path=manifest)
michael@0:         c = FileCopier()
michael@0:         m.populate_registry(c)
michael@0:         c.copy(dest)
michael@0: 
michael@0:         with open(destfile, 'rt') as fh:
michael@0:             self.assertEqual(fh.read(), 'SOURCE\nINCLUDE\n')
michael@0: 
michael@0:         # Set the time on the source file back, so it won't be picked up as
michael@0:         # modified in the next test.
michael@0:         time = os.path.getmtime(source) - 1
michael@0:         os.utime(source, (time, time))
michael@0: 
michael@0:         # Now, modify the include file (but not the original source).
michael@0:         with open(include, 'wt') as fh:
michael@0:             fh.write('INCLUDE MODIFIED\n')
michael@0:         time = os.path.getmtime(include) - 1
michael@0:         os.utime(destfile, (time, time))
michael@0: 
michael@0:         # Apply the manifest, and confirm that the change to the include file
michael@0:         # is detected. That should cause the preprocessor to run again.
michael@0:         m = InstallManifest(path=manifest)
michael@0:         c = FileCopier()
michael@0:         m.populate_registry(c)
michael@0:         c.copy(dest)
michael@0: 
michael@0:         with open(destfile, 'rt') as fh:
michael@0:             self.assertEqual(fh.read(), 'SOURCE\nINCLUDE MODIFIED\n')
michael@0: 
michael@0: if __name__ == '__main__':
michael@0:     mozunit.main()