| #!/usr/bin/python |
| # |
| # Copyright 2019 The Cobalt Authors. All Rights Reserved. |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| |
| import json |
| import os |
| import shutil |
| import stat |
| import subprocess |
| import unittest |
| |
| import _env # pylint: disable=relative-import,unused-import |
| from cobalt.build import cobalt_archive |
| from starboard.build import filelist |
| from starboard.build import filelist_test |
| from starboard.build import port_symlink |
| from starboard.tools import util |
| |
| |
| class CobaltArchiveTest(unittest.TestCase): |
| |
| def testFoldIdenticalFiles(self): |
| tf_root = filelist_test.TempFileSystem('bundler_fold') |
| tf_root.Clear() |
| tf1 = filelist_test.TempFileSystem(os.path.join('bundler_fold', '1')) |
| tf2 = filelist_test.TempFileSystem(os.path.join('bundler_fold', '2')) |
| tf1.Make() |
| tf2.Make() |
| flist = filelist.FileList() |
| subdirs = [tf1.root_in_tmp, tf2.root_in_tmp] |
| flist.AddAllFilesInPaths(tf_root.root_tmp, subdirs) |
| flist.Print() |
| identical_files = [tf1.test_txt, tf2.test_txt] |
| physical_files, copy_files = cobalt_archive._FoldIdenticalFiles( |
| identical_files) |
| self.assertEqual(tf1.test_txt, physical_files[0]) |
| self.assertIn(tf1.test_txt, copy_files[0][0]) |
| self.assertIn(tf2.test_txt, copy_files[0][1]) |
| |
| def testMakesDeployInfo(self): |
| flist = filelist.FileList() |
| tf = filelist_test.TempFileSystem() |
| tf.Clear() |
| tf.Make() |
| bundle_zip = os.path.join(tf.root_tmp, 'bundle.zip') |
| car = cobalt_archive.CobaltArchive(bundle_zip) |
| car.MakeArchive(platform_name='fake', |
| platform_sdk_version='fake_sdk', |
| config='devel', |
| file_list=flist) |
| out_dir = os.path.join(tf.root_tmp, 'out') |
| car.ExtractTo(out_dir) |
| out_metadata_file = os.path.join(out_dir, cobalt_archive._OUT_METADATA_PATH) |
| self.assertEqual(filelist.GetFileType(out_metadata_file), |
| filelist.TYPE_FILE) |
| with open(out_metadata_file) as fd: |
| text = fd.read() |
| js = json.loads(text) |
| self.assertTrue(js) |
| self.assertEqual(js['sdk_version'], 'fake_sdk') |
| self.assertEqual(js['platform'], 'fake') |
| self.assertEqual(js['config'], 'devel') |
| |
| def testExtractTo(self): |
| flist = filelist.FileList() |
| tf = filelist_test.TempFileSystem() |
| tf.Clear() |
| tf.Make() |
| flist.AddFile(tf.root_in_tmp, tf.test_txt) |
| flist.AddSymLink(tf.root_in_tmp, tf.sym_dir) |
| bundle_zip = os.path.join(tf.root_tmp, 'bundle.zip') |
| car = cobalt_archive.CobaltArchive(bundle_zip) |
| car.MakeArchive(platform_name='fake', |
| platform_sdk_version='fake_sdk', |
| config='devel', |
| file_list=flist) |
| out_dir = os.path.join(tf.root_tmp, 'out') |
| car.ExtractTo(out_dir) |
| out_from_dir = os.path.join(out_dir, 'from_dir') |
| out_from_dir_lnk = os.path.join(out_dir, 'from_dir_lnk') |
| self.assertEqual(filelist.GetFileType(out_from_dir), |
| filelist.TYPE_DIRECTORY) |
| self.assertEqual(filelist.GetFileType(out_from_dir_lnk), |
| filelist.TYPE_SYMLINK_DIR) |
| resolved_from_link_path = os.path.join( |
| out_dir, port_symlink.ReadSymLink(out_from_dir_lnk)) |
| self.assertEqual(os.path.abspath(out_from_dir), |
| os.path.abspath(resolved_from_link_path)) |
| |
| def testExtractFileWithLongFileName(self): |
| """Tests that a long file name can be archived and extracted.""" |
| flist = filelist.FileList() |
| tf = filelist_test.TempFileSystem() |
| tf.Clear() |
| tf.Make() |
| self.assertTrue(os.path.exists(tf.root_in_tmp)) |
| suffix_path = os.path.join( |
| 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', |
| 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', |
| 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', |
| 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', |
| 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', |
| 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', |
| 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', |
| 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', |
| 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', |
| 'test.txt' |
| ) |
| input_dst = os.path.join(tf.root_in_tmp, suffix_path) |
| out_dir = os.path.join(tf.root_tmp, 'out') |
| output_dst = os.path.join(out_dir, suffix_path) |
| _MoveFileWithLongPath(tf.test_txt, input_dst) |
| self.assertTrue(_LongPathExists(input_dst)) |
| flist.AddFile(tf.root_in_tmp, input_dst) |
| |
| bundle_zip = os.path.join(tf.root_tmp, 'bundle.zip') |
| car = cobalt_archive.CobaltArchive(bundle_zip) |
| car.MakeArchive(platform_name='fake', |
| platform_sdk_version='fake_sdk', |
| config='devel', |
| file_list=flist) |
| car.ExtractTo(out_dir) |
| self.assertTrue(_LongPathExists(output_dst)) |
| |
| @unittest.skipIf(port_symlink.IsWindows(), 'Any platform but windows.') |
| def testExecutionAttribute(self): |
| flist = filelist.FileList() |
| tf = filelist_test.TempFileSystem() |
| tf.Make() |
| # Execution bit seems to turn off the read bit, so we just set all |
| # read/write/execute bit for the user. |
| write_flags = stat.S_IXUSR | stat.S_IWUSR | stat.S_IRUSR |
| os.chmod(tf.test_txt, write_flags) |
| self.assertNotEqual( |
| 0, write_flags & cobalt_archive._GetFilePermissions(tf.test_txt)) |
| flist.AddFile(tf.root_tmp, tf.test_txt) |
| bundle_zip = os.path.join(tf.root_tmp, 'bundle.zip') |
| car = cobalt_archive.CobaltArchive(bundle_zip) |
| car.MakeArchive(platform_name='fake', |
| platform_sdk_version='fake_sdk', |
| config='devel', |
| file_list=flist) |
| # Now grab the json file and check that the file appears in the |
| # executable_file list. |
| json_str = car.ReadFile( |
| '__cobalt_archive/finalize_decompression/decompress.json') |
| decompress_dict = json.loads(json_str) |
| executable_files = decompress_dict.get('executable_files') |
| # Expect that the executable file appears in the executable_files. |
| self.assertTrue(executable_files) |
| archive_path = os.path.relpath(tf.test_txt, tf.root_tmp) |
| self.assertIn(archive_path, executable_files) |
| out_dir = os.path.join(tf.root_tmp, 'out') |
| car.ExtractTo(output_dir=out_dir) |
| out_file = os.path.join(out_dir, tf.test_txt) |
| self.assertTrue(_LongPathExists(out_file)) |
| perms = cobalt_archive._GetFilePermissions(out_file) |
| self.assertTrue(perms & stat.S_IXUSR) |
| |
| |
| def _SilentCall(cmd_str): |
| proc = subprocess.Popen(cmd_str, |
| shell=True, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.STDOUT) |
| (_, _) = proc.communicate() |
| return proc.returncode |
| |
| |
| def _LongPathExists(p): |
| if port_symlink.IsWindows(): |
| rc = _SilentCall('dir /s /b "%s"' % p) |
| return rc == 0 |
| else: |
| return os.path.isfile(p) |
| |
| |
| def _MoveFileWithLongPath(src_file, dst_file): |
| dst_dir = os.path.dirname(dst_file) |
| if port_symlink.IsWindows(): |
| # Work around for file-length path limitations on Windows. |
| src_dir = os.path.dirname(src_file) |
| file_name = os.path.basename(src_file) |
| shell_cmd = 'robocopy "%s" "%s" "%s" /MOV' % (src_dir, dst_dir, file_name) |
| rc = _SilentCall(shell_cmd) |
| if 1 != rc: # Robocopy returns 1 if a file was copied. |
| raise OSError('File %s was not copied' % src_file) |
| expected_out_file = os.path.join(dst_dir, file_name) |
| if not _LongPathExists(expected_out_file): |
| raise OSError('File did not end up in %s' % dst_dir) |
| return |
| else: |
| if not os.path.isdir(dst_dir): |
| os.makedirs(dst_dir) |
| shutil.move(src_file, dst_file) |
| if not os.path.isfile(dst_file): |
| raise OSError('File did not end up in %s' % dst_dir) |
| |
| |
| if __name__ == '__main__': |
| util.SetupDefaultLoggingConfig() |
| unittest.main(verbosity=2) |
