blob: 0965c605676d14ec8dff82046f8b085df0a68266 [file] [log] [blame]
#!/usr/bin/env python3.8
# Copyright 2020 The Fuchsia 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 tempfile
import shutil
import tarfile
import unittest
from unittest import mock
from gather_package_deps import GatherPackageDeps
class GatherPackageDepsTests(unittest.TestCase):
source_dir = None
package_json_path = None
meta_far_path = None
output_dir = None
def setUp(self):
self.source_dir = tempfile.TemporaryDirectory()
self.output_dir = tempfile.TemporaryDirectory()
self.package_json_path = os.path.join(self.source_dir.name, 'pkg.json')
self.meta_far_path = os.path.join(self.source_dir.name, 'meta.far')
# Create placeholder files.
open(self.package_json_path, 'a').close()
open(self.meta_far_path, 'a').close()
def tearDown(self):
self.source_dir.cleanup()
self.output_dir.cleanup()
def test_init(self):
GatherPackageDeps(
self.package_json_path, self.meta_far_path, self.output_dir.name)
with self.assertRaises(ValueError):
GatherPackageDeps('', self.meta_far_path, self.output_dir.name)
with self.assertRaises(ValueError):
GatherPackageDeps(None, self.meta_far_path, self.output_dir.name)
with self.assertRaises(ValueError):
GatherPackageDeps(self.package_json_path, '', self.output_dir.name)
with self.assertRaises(ValueError):
GatherPackageDeps(
self.package_json_path, None, self.output_dir.name)
with self.assertRaises(ValueError):
GatherPackageDeps(self.package_json_path, self.meta_far_path, '')
with self.assertRaises(ValueError):
GatherPackageDeps(self.package_json_path, self.meta_far_path, None)
def test_parse_package_json(self):
gatherer = GatherPackageDeps(
self.package_json_path, self.meta_far_path, self.output_dir.name)
with open(self.package_json_path, 'w') as f:
f.write('noooot JSOOOOON')
with self.assertRaises(ValueError):
gatherer.parse_package_json()
with open(self.package_json_path, 'w') as f:
f.write(r'{ }')
with self.assertRaises(KeyError):
gatherer.parse_package_json()
with open(self.package_json_path, 'w') as f:
f.write(r'{ "blobs": [] }')
manifest_dict = gatherer.parse_package_json()
self.assertDictEqual(manifest_dict, {})
with open(self.package_json_path, 'w') as f:
f.write(
"""{ "blobs":
[ { "source_path": "some/path/A", "path": "path/A" } ]
}
""")
manifest_dict = gatherer.parse_package_json()
self.assertDictEqual(manifest_dict, {'path/A': 'some/path/A'})
with open(self.package_json_path, 'w') as f:
f.write(
"""{ "blobs":
[
{ "source_path": "some/path/A", "path": "path/A" },
{ "source_path": "some/path/B", "path": "path/B" }
]
}
""")
manifest_dict = gatherer.parse_package_json()
self.assertDictEqual(
manifest_dict, {
'path/A': 'some/path/A',
'path/B': 'some/path/B'
})
@unittest.mock.patch.object(shutil, 'copyfile', autospec=True)
def test_copy_meta_far(self, copyfile_mock):
gatherer = GatherPackageDeps(
self.package_json_path, self.meta_far_path, self.output_dir.name)
gatherer.copy_meta_far()
copyfile_mock.assert_called_once_with(
self.meta_far_path, os.path.join(self.output_dir.name, 'meta.far'))
@unittest.mock.patch.object(shutil, 'copyfile', autospec=True)
@unittest.mock.patch.object(os, 'makedirs', autospec=True)
def test_copy_to_output_dir(self, makedirs_mock, copyfile_mock):
gatherer = GatherPackageDeps(
self.package_json_path, self.meta_far_path, self.output_dir.name)
test_dict = {
'path/A': '../../../thing/A',
'path/B': 'thing/B',
'path/to/C': '/abs/path/C'
}
gatherer.copy_to_output_dir(test_dict)
self.assertEqual(3, makedirs_mock.call_count)
self.assertEqual(3, copyfile_mock.call_count)
self.assertDictEqual(
test_dict, {
'path/A': 'thing/A',
'path/B': 'thing/B',
'path/to/C': 'abs/path/C'
})
def test_write_new_manifest(self):
gatherer = GatherPackageDeps(
self.package_json_path, self.meta_far_path, self.output_dir.name)
gatherer.write_new_manifest(
{
'path/A': 'some/path/A',
'path/B': 'some/path/B'
})
with open(os.path.join(self.output_dir.name, 'package.manifest'),
'r') as f:
manifest_data = f.read()
self.assertIn('path/A=some/path/A', manifest_data)
self.assertIn('path/B=some/path/B', manifest_data)
self.assertIn('meta/package=meta.far', manifest_data)
def test_archive_output(self):
tar_path = os.path.join(self.output_dir.name, 'package.tar')
self.assertFalse(os.path.isfile(tar_path))
gatherer = GatherPackageDeps(
self.package_json_path, self.meta_far_path, self.output_dir.name)
gatherer.archive_output()
self.assertTrue(os.path.isfile(tar_path))
file_a = os.path.join(self.output_dir.name, 'fileA')
file_b = os.path.join(self.output_dir.name, 'sub', 'fileB')
file_c = os.path.join(self.output_dir.name, 'another', 'dir', 'fileC')
with open(file_a, 'w') as f:
f.write('A')
os.makedirs(os.path.dirname(file_b), exist_ok=False)
with open(file_b, 'w') as f:
f.write('BB')
os.makedirs(os.path.dirname(file_c), exist_ok=False)
with open(file_c, 'w') as f:
f.write('CCC')
gatherer.archive_output()
self.assertTrue(os.path.isfile(tar_path))
# Main thing we need to check here is that the paths within the archive
# are relative instead of absolute, because absolute paths relative to the
# system that created the archive are meaningless for anyone unarchiving.
size_index = {'./fileA': 1, 'sub/fileB': 2, 'another/dir/fileC': 3}
with tarfile.open(tar_path, 'r') as tar:
for member in tar.getmembers():
self.assertIn(member.name, size_index)
self.assertEqual(member.size, size_index[member.name])
def test_run(self):
backup_cwd = os.getcwd()
os.chdir(self.source_dir.name)
file_meta = 'meta.far'
file_a = 'fileA'
file_b = os.path.join('sub', 'fileB')
file_c = os.path.join('another', 'dir', 'fileC')
open(file_meta, 'a').close()
open(file_a, 'a').close()
os.makedirs(os.path.dirname(file_b), exist_ok=False)
open(file_b, 'a').close()
os.makedirs(os.path.dirname(file_c), exist_ok=False)
open(file_c, 'a').close()
with open(self.package_json_path, 'w') as f:
f.write(
"""{{ "blobs":
[
{{ "source_path": "{}", "path": "path/A" }},
{{ "source_path": "{}", "path": "path/B" }},
{{ "source_path": "{}", "path": "path/C" }}
]
}}
""".format(file_a, file_b, file_c))
gatherer = GatherPackageDeps(
self.package_json_path, self.meta_far_path, self.output_dir.name)
gatherer.run()
expected_files = {
'./package.manifest', './meta.far', './fileA', 'sub/fileB',
'another/dir/fileC'
}
observed_files = set()
expected_manifest_lines = {
'path/A=fileA', 'path/B=sub/fileB', 'path/C=another/dir/fileC',
'meta/package=meta.far'
}
observed_manifest_lines = set()
with tarfile.open(os.path.join(self.output_dir.name, 'package.tar'),
'r') as tar:
for member in tar.getmembers():
observed_files.add(member.name)
if member.name == './package.manifest':
for line in tar.extractfile(member).readlines():
observed_manifest_lines.add(
line.decode("utf-8").strip())
self.assertEqual(observed_files, expected_files)
self.assertEqual(observed_manifest_lines, expected_manifest_lines)
os.chdir(backup_cwd)
if __name__ == '__main__':
unittest.main()