# Copyright 2021 The Pigweed Authors # # 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 # # https://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. """Unit tests for pw_software_update/update_bundle.py.""" from pathlib import Path import tempfile import unittest from pw_software_update import update_bundle from pw_software_update.tuf_pb2 import SignedRootMetadata, TargetsMetadata class TargetsFromDirectoryTest(unittest.TestCase): """Test turning a directory into TUF targets.""" def test_excludes(self): """Checks that excludes are excluded.""" with tempfile.TemporaryDirectory() as tempdir_name: temp_root = Path(tempdir_name) foo_path = temp_root / 'foo.bin' bar_path = temp_root / 'bar.bin' baz_path = temp_root / 'baz.bin' qux_path = temp_root / 'qux.exe' for path in (foo_path, bar_path, baz_path, qux_path): path.touch() targets = update_bundle.targets_from_directory( temp_root, exclude=(Path('foo.bin'), Path('baz.bin'))) self.assertNotIn('foo.bin', targets) self.assertEqual(bar_path, targets['bar.bin']) self.assertNotIn('baz.bin', targets) self.assertEqual(qux_path, targets['qux.exe']) def test_excludes_and_remapping(self): """Checks that remapping works, even in combination with excludes.""" with tempfile.TemporaryDirectory() as tempdir_name: temp_root = Path(tempdir_name) foo_path = temp_root / 'foo.bin' bar_path = temp_root / 'bar.bin' baz_path = temp_root / 'baz.bin' qux_path = temp_root / 'qux.exe' remap_paths = { Path('foo.bin'): 'main', Path('bar.bin'): 'backup', Path('baz.bin'): 'tertiary', } for path in (foo_path, bar_path, baz_path, qux_path): path.touch() targets = update_bundle.targets_from_directory( temp_root, exclude=(Path('qux.exe'), ), remap_paths=remap_paths) self.assertEqual(foo_path, targets['main']) self.assertEqual(bar_path, targets['backup']) self.assertEqual(baz_path, targets['tertiary']) self.assertNotIn('qux.exe', targets) def test_incomplete_remapping_logs(self): """Checks that incomplete remappings log warnings.""" with tempfile.TemporaryDirectory() as tempdir_name: temp_root = Path(tempdir_name) foo_path = temp_root / 'foo.bin' bar_path = temp_root / 'bar.bin' foo_path.touch() bar_path.touch() remap_paths = {Path('foo.bin'): 'main'} with self.assertLogs(level='WARNING') as log: update_bundle.targets_from_directory( temp_root, exclude=(Path('qux.exe'), ), remap_paths=remap_paths) self.assertIn('Some remaps defined, but not "bar.bin"', log.output[0]) def test_remap_of_missing_file(self): """Checks that remapping a missing file raises an error.""" with tempfile.TemporaryDirectory() as tempdir_name: temp_root = Path(tempdir_name) foo_path = temp_root / 'foo.bin' foo_path.touch() remap_paths = { Path('foo.bin'): 'main', Path('bar.bin'): 'backup', } with self.assertRaises(FileNotFoundError): update_bundle.targets_from_directory(temp_root, remap_paths=remap_paths) class GenUnsignedUpdateBundleTest(unittest.TestCase): """Test the generation of unsigned update bundles.""" def test_bundle_generation(self): """Tests basic creation of an UpdateBundle.""" with tempfile.TemporaryDirectory() as tempdir_name: temp_root = Path(tempdir_name) foo_path = temp_root / 'foo.bin' bar_path = temp_root / 'bar.bin' baz_path = temp_root / 'baz.bin' qux_path = temp_root / 'subdir' / 'qux.exe' foo_bytes = b'\xf0\x0b\xa4' bar_bytes = b'\x0b\xa4\x99' baz_bytes = b'\xba\x59\x06' qux_bytes = b'\x8a\xf3\x12' foo_path.write_bytes(foo_bytes) bar_path.write_bytes(bar_bytes) baz_path.write_bytes(baz_bytes) (temp_root / 'subdir').mkdir() qux_path.write_bytes(qux_bytes) targets = { foo_path: 'foo', bar_path: 'bar', baz_path: 'baz', qux_path: 'qux', } serialized_root_metadata_bytes = b'\x12\x34\x56\x78' bundle = update_bundle.gen_unsigned_update_bundle( targets, targets_metadata_version=42, root_metadata=SignedRootMetadata( serialized_root_metadata=serialized_root_metadata_bytes)) self.assertEqual(foo_bytes, bundle.target_payloads['foo']) self.assertEqual(bar_bytes, bundle.target_payloads['bar']) self.assertEqual(baz_bytes, bundle.target_payloads['baz']) self.assertEqual(qux_bytes, bundle.target_payloads['qux']) targets_metadata = TargetsMetadata.FromString( bundle.targets_metadata['targets'].serialized_targets_metadata) self.assertEqual(targets_metadata.common_metadata.version, 42) self.assertEqual(serialized_root_metadata_bytes, bundle.root_metadata.serialized_root_metadata) def test_persist_to_disk(self): """Tests persisting the TUF repo to disk for debugging""" with tempfile.TemporaryDirectory() as tempdir_name: temp_root = Path(tempdir_name) foo_path = temp_root / 'foo.bin' bar_path = temp_root / 'bar.bin' baz_path = temp_root / 'baz.bin' qux_path = temp_root / 'subdir' / 'qux.exe' foo_bytes = b'\xf0\x0b\xa4' bar_bytes = b'\x0b\xa4\x99' baz_bytes = b'\xba\x59\x06' qux_bytes = b'\x8a\xf3\x12' foo_path.write_bytes(foo_bytes) bar_path.write_bytes(bar_bytes) baz_path.write_bytes(baz_bytes) (temp_root / 'subdir').mkdir() qux_path.write_bytes(qux_bytes) targets = { foo_path: 'foo', bar_path: 'bar', baz_path: 'baz', qux_path: 'subdir/qux', } persist_path = temp_root / 'persisted' update_bundle.gen_unsigned_update_bundle(targets, persist=persist_path) self.assertEqual(foo_bytes, (persist_path / 'foo').read_bytes()) self.assertEqual(bar_bytes, (persist_path / 'bar').read_bytes()) self.assertEqual(baz_bytes, (persist_path / 'baz').read_bytes()) self.assertEqual(qux_bytes, (persist_path / 'subdir' / 'qux').read_bytes()) class ParseTargetArgTest(unittest.TestCase): """Test the parsing of target argument strings.""" def test_valid_arg(self): """Checks that valid remap strings are parsed correctly.""" file_path, target_name = update_bundle.parse_target_arg( 'foo.bin > main') self.assertEqual(Path('foo.bin'), file_path) self.assertEqual('main', target_name) def test_invalid_arg_raises(self): """Checks that invalid remap string raise an error.""" with self.assertRaises(ValueError): update_bundle.parse_target_arg('foo.bin main') if __name__ == '__main__': unittest.main()