• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2021 The Pigweed Authors
2#
3# Licensed under the Apache License, Version 2.0 (the "License"); you may not
4# use this file except in compliance with the License. You may obtain a copy of
5# the License at
6#
7#     https://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12# License for the specific language governing permissions and limitations under
13# the License.
14"""Unit tests for pw_software_update/update_bundle.py."""
15
16from pathlib import Path
17import tempfile
18import unittest
19
20from pw_software_update import update_bundle
21from pw_software_update.tuf_pb2 import SignedRootMetadata, TargetsMetadata
22
23
24class TargetsFromDirectoryTest(unittest.TestCase):
25    """Test turning a directory into TUF targets."""
26    def test_excludes(self):
27        """Checks that excludes are excluded."""
28        with tempfile.TemporaryDirectory() as tempdir_name:
29            temp_root = Path(tempdir_name)
30            foo_path = temp_root / 'foo.bin'
31            bar_path = temp_root / 'bar.bin'
32            baz_path = temp_root / 'baz.bin'
33            qux_path = temp_root / 'qux.exe'
34            for path in (foo_path, bar_path, baz_path, qux_path):
35                path.touch()
36
37            targets = update_bundle.targets_from_directory(
38                temp_root, exclude=(Path('foo.bin'), Path('baz.bin')))
39
40            self.assertNotIn('foo.bin', targets)
41            self.assertEqual(bar_path, targets['bar.bin'])
42            self.assertNotIn('baz.bin', targets)
43            self.assertEqual(qux_path, targets['qux.exe'])
44
45    def test_excludes_and_remapping(self):
46        """Checks that remapping works, even in combination with excludes."""
47        with tempfile.TemporaryDirectory() as tempdir_name:
48            temp_root = Path(tempdir_name)
49            foo_path = temp_root / 'foo.bin'
50            bar_path = temp_root / 'bar.bin'
51            baz_path = temp_root / 'baz.bin'
52            qux_path = temp_root / 'qux.exe'
53            remap_paths = {
54                Path('foo.bin'): 'main',
55                Path('bar.bin'): 'backup',
56                Path('baz.bin'): 'tertiary',
57            }
58            for path in (foo_path, bar_path, baz_path, qux_path):
59                path.touch()
60
61            targets = update_bundle.targets_from_directory(
62                temp_root,
63                exclude=(Path('qux.exe'), ),
64                remap_paths=remap_paths)
65
66            self.assertEqual(foo_path, targets['main'])
67            self.assertEqual(bar_path, targets['backup'])
68            self.assertEqual(baz_path, targets['tertiary'])
69            self.assertNotIn('qux.exe', targets)
70
71    def test_incomplete_remapping_logs(self):
72        """Checks that incomplete remappings log warnings."""
73        with tempfile.TemporaryDirectory() as tempdir_name:
74            temp_root = Path(tempdir_name)
75            foo_path = temp_root / 'foo.bin'
76            bar_path = temp_root / 'bar.bin'
77            foo_path.touch()
78            bar_path.touch()
79            remap_paths = {Path('foo.bin'): 'main'}
80
81            with self.assertLogs(level='WARNING') as log:
82                update_bundle.targets_from_directory(
83                    temp_root,
84                    exclude=(Path('qux.exe'), ),
85                    remap_paths=remap_paths)
86
87                self.assertIn('Some remaps defined, but not "bar.bin"',
88                              log.output[0])
89
90    def test_remap_of_missing_file(self):
91        """Checks that remapping a missing file raises an error."""
92        with tempfile.TemporaryDirectory() as tempdir_name:
93            temp_root = Path(tempdir_name)
94            foo_path = temp_root / 'foo.bin'
95            foo_path.touch()
96            remap_paths = {
97                Path('foo.bin'): 'main',
98                Path('bar.bin'): 'backup',
99            }
100
101            with self.assertRaises(FileNotFoundError):
102                update_bundle.targets_from_directory(temp_root,
103                                                     remap_paths=remap_paths)
104
105
106class GenUnsignedUpdateBundleTest(unittest.TestCase):
107    """Test the generation of unsigned update bundles."""
108    def test_bundle_generation(self):
109        """Tests basic creation of an UpdateBundle."""
110        with tempfile.TemporaryDirectory() as tempdir_name:
111            temp_root = Path(tempdir_name)
112            foo_path = temp_root / 'foo.bin'
113            bar_path = temp_root / 'bar.bin'
114            baz_path = temp_root / 'baz.bin'
115            qux_path = temp_root / 'subdir' / 'qux.exe'
116            foo_bytes = b'\xf0\x0b\xa4'
117            bar_bytes = b'\x0b\xa4\x99'
118            baz_bytes = b'\xba\x59\x06'
119            qux_bytes = b'\x8a\xf3\x12'
120            foo_path.write_bytes(foo_bytes)
121            bar_path.write_bytes(bar_bytes)
122            baz_path.write_bytes(baz_bytes)
123            (temp_root / 'subdir').mkdir()
124            qux_path.write_bytes(qux_bytes)
125            targets = {
126                foo_path: 'foo',
127                bar_path: 'bar',
128                baz_path: 'baz',
129                qux_path: 'qux',
130            }
131            serialized_root_metadata_bytes = b'\x12\x34\x56\x78'
132
133            bundle = update_bundle.gen_unsigned_update_bundle(
134                targets,
135                targets_metadata_version=42,
136                root_metadata=SignedRootMetadata(
137                    serialized_root_metadata=serialized_root_metadata_bytes))
138
139            self.assertEqual(foo_bytes, bundle.target_payloads['foo'])
140            self.assertEqual(bar_bytes, bundle.target_payloads['bar'])
141            self.assertEqual(baz_bytes, bundle.target_payloads['baz'])
142            self.assertEqual(qux_bytes, bundle.target_payloads['qux'])
143            targets_metadata = TargetsMetadata.FromString(
144                bundle.targets_metadata['targets'].serialized_targets_metadata)
145            self.assertEqual(targets_metadata.common_metadata.version, 42)
146            self.assertEqual(serialized_root_metadata_bytes,
147                             bundle.root_metadata.serialized_root_metadata)
148
149    def test_persist_to_disk(self):
150        """Tests persisting the TUF repo to disk for debugging"""
151        with tempfile.TemporaryDirectory() as tempdir_name:
152            temp_root = Path(tempdir_name)
153            foo_path = temp_root / 'foo.bin'
154            bar_path = temp_root / 'bar.bin'
155            baz_path = temp_root / 'baz.bin'
156            qux_path = temp_root / 'subdir' / 'qux.exe'
157            foo_bytes = b'\xf0\x0b\xa4'
158            bar_bytes = b'\x0b\xa4\x99'
159            baz_bytes = b'\xba\x59\x06'
160            qux_bytes = b'\x8a\xf3\x12'
161            foo_path.write_bytes(foo_bytes)
162            bar_path.write_bytes(bar_bytes)
163            baz_path.write_bytes(baz_bytes)
164            (temp_root / 'subdir').mkdir()
165            qux_path.write_bytes(qux_bytes)
166            targets = {
167                foo_path: 'foo',
168                bar_path: 'bar',
169                baz_path: 'baz',
170                qux_path: 'subdir/qux',
171            }
172            persist_path = temp_root / 'persisted'
173
174            update_bundle.gen_unsigned_update_bundle(targets,
175                                                     persist=persist_path)
176
177            self.assertEqual(foo_bytes, (persist_path / 'foo').read_bytes())
178            self.assertEqual(bar_bytes, (persist_path / 'bar').read_bytes())
179            self.assertEqual(baz_bytes, (persist_path / 'baz').read_bytes())
180            self.assertEqual(qux_bytes,
181                             (persist_path / 'subdir' / 'qux').read_bytes())
182
183
184class ParseTargetArgTest(unittest.TestCase):
185    """Test the parsing of target argument strings."""
186    def test_valid_arg(self):
187        """Checks that valid remap strings are parsed correctly."""
188        file_path, target_name = update_bundle.parse_target_arg(
189            'foo.bin > main')
190
191        self.assertEqual(Path('foo.bin'), file_path)
192        self.assertEqual('main', target_name)
193
194    def test_invalid_arg_raises(self):
195        """Checks that invalid remap string raise an error."""
196        with self.assertRaises(ValueError):
197            update_bundle.parse_target_arg('foo.bin main')
198
199
200if __name__ == '__main__':
201    unittest.main()
202