| # Copyright 2015 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Test the image_lib module.""" |
| |
| import collections |
| import gc |
| import glob |
| import os |
| from pathlib import Path |
| import stat |
| from unittest import mock |
| |
| from chromite.lib import chromeos_version |
| from chromite.lib import constants |
| from chromite.lib import cros_build_lib |
| from chromite.lib import cros_test_lib |
| from chromite.lib import git |
| from chromite.lib import image_lib |
| from chromite.lib import install_mask |
| from chromite.lib import osutils |
| from chromite.lib import partial_mock |
| from chromite.lib import portage_util |
| from chromite.lib import retry_util |
| from chromite.utils import c_loop |
| from chromite.utils import os_util |
| |
| |
| # pylint: disable=protected-access |
| |
| |
| class FakeException(Exception): |
| """Fake exception used for testing exception handling.""" |
| |
| |
| FAKE_PATH = "/imaginary/file" |
| LOOP_DEV = "/dev/loop9999" |
| LOOP_PART_COUNT = 12 |
| LOOP_PARTITION_INFO = [ |
| image_lib.PartitionInfo(1, 2928640, 28672, "", "STATE"), |
| image_lib.PartitionInfo(2, 20480, 32768, "", "KERN-A"), |
| image_lib.PartitionInfo(3, 286720, 2641920, "", "ROOT-A"), |
| image_lib.PartitionInfo(4, 53248, 32768, "", "KERN-B"), |
| image_lib.PartitionInfo(5, 282624, 4096, "", "ROOT-B"), |
| image_lib.PartitionInfo(6, 16448, 1, "", "KERN-C"), |
| image_lib.PartitionInfo(7, 16449, 1, "", "ROOT-C"), |
| image_lib.PartitionInfo(8, 86016, 32768, "", "OEM"), |
| image_lib.PartitionInfo(9, 16450, 1, "", "reserved"), |
| image_lib.PartitionInfo(10, 16451, 1, "", "reserved"), |
| image_lib.PartitionInfo(11, 64, 16384, "", "RWFW"), |
| image_lib.PartitionInfo(12, 249856, 32768, "", "EFI-SYSTEM"), |
| ] |
| LOOP_PARTS_DICT = { |
| p.number: "%sp%d" % (LOOP_DEV, p.number) for p in LOOP_PARTITION_INFO |
| } |
| LOOP_PARTS_LIST = LOOP_PARTS_DICT.values() |
| FAKE_DATE_STRING = "2022_07_20_203326" |
| |
| |
| class LoopbackPartitionsMock(image_lib.LoopbackPartitions): |
| """Mocked loopback partition class to use in unit tests.""" |
| |
| def __init__(self, *args, **kwargs) -> None: |
| super().__init__(*args, **kwargs) |
| self.enable_rw_called = set() |
| self.disable_rw_called = set() |
| |
| def _InitGpt(self) -> None: |
| """Initialize the GPT info.""" |
| self._gpt_table = LOOP_PARTITION_INFO |
| |
| def Attach(self) -> None: |
| """Initialize the loopback device.""" |
| self.dev = LOOP_DEV |
| if not self.destination: |
| self.destination = osutils.TempDir() |
| self.parts = { |
| p.number: "%sp%s" % (self.dev, p.number) for p in self._gpt_table |
| } |
| |
| def EnableRwMount(self, part_id, offset=0) -> None: |
| """Stub out enable rw mount.""" |
| self.enable_rw_called.add((part_id, offset)) |
| |
| def DisableRwMount(self, part_id, offset=0) -> None: |
| """Stub out disable rw mount.""" |
| self.disable_rw_called.add((part_id, offset)) |
| |
| def _Mount(self, part, mount_opts): |
| """Stub out mount operations.""" |
| dest_number, _ = self._GetMountPointAndSymlink(part) |
| # Don't actually even try to mount it. |
| self._mounted.add(part) |
| return dest_number |
| |
| def _Unmount(self, part) -> None: |
| """Stub out unmount operations.""" |
| self._mounted.remove(part) |
| |
| def close(self) -> None: |
| pass |
| |
| |
| class LoopbackPartitionsTest(cros_test_lib.MockTempDirTestCase): |
| """Test the loopback partitions class""" |
| |
| def setUp(self) -> None: |
| self.rc_mock = cros_test_lib.RunCommandMock() |
| self.StartPatcher(self.rc_mock) |
| self.rc_mock.SetDefaultCmdResult() |
| self.rc_mock.AddCmdResult(partial_mock.In("--show"), stdout=LOOP_DEV) |
| |
| self.PatchObject( |
| image_lib, |
| "GetImageDiskPartitionInfo", |
| return_value=LOOP_PARTITION_INFO, |
| ) |
| self.PatchObject(glob, "glob", return_value=LOOP_PARTS_LIST) |
| self.mount_mock = self.PatchObject(osutils, "MountDir") |
| self.umount_mock = self.PatchObject(osutils, "UmountDir") |
| self.retry_mock = self.PatchObject(retry_util, "RetryException") |
| self.addpart_mock = self.PatchObject( |
| image_lib.LoopbackPartitions, "_AddPartitions" |
| ) |
| self.delpart_mock = self.PatchObject( |
| image_lib.LoopbackPartitions, "_DeletePartitions" |
| ) |
| self.detach_mock = self.PatchObject(c_loop, "detach") |
| |
| def fake_which(val, *_arg, **_kwargs): |
| return val |
| |
| self.PatchObject(osutils, "Which", side_effect=fake_which) |
| |
| # Patch osutils.IsRootUser() to pretend running as root so we attempt to |
| # do all the setup directly instead of falling back to our sudo helper. |
| self.PatchObject(os_util, "is_root_user", return_value=True) |
| |
| def testContextManager(self) -> None: |
| """Test using the loopback class as a context manager.""" |
| with image_lib.LoopbackPartitions(FAKE_PATH) as lb: |
| self.rc_mock.assertCommandContains( |
| ["losetup", "--show", "-f", FAKE_PATH] |
| ) |
| self.delpart_mock.assert_called_once() |
| self.delpart_mock.reset_mock() |
| self.addpart_mock.assert_called_once() |
| self.detach_mock.assert_not_called() |
| self.assertEqual(lb.parts, LOOP_PARTS_DICT) |
| self.assertEqual(lb._gpt_table, LOOP_PARTITION_INFO) |
| self.delpart_mock.assert_called_once() |
| self.detach_mock.assert_called_once() |
| |
| def testContextManagerWithMounts(self) -> None: |
| """Test using the loopback class as a context manager with mounts.""" |
| syml = self.PatchObject(osutils, "SafeSymlink") |
| part_ids = (1, "ROOT-A") |
| with image_lib.LoopbackPartitions( |
| FAKE_PATH, part_ids=part_ids, mount_opts=("ro",) |
| ) as lb: |
| expected_mounts = set() |
| expected_calls = [] |
| for part_id in part_ids: |
| for part in LOOP_PARTITION_INFO: |
| if part_id in (part.name, part.number): |
| expected_mounts.add(part) |
| expected_calls.append( |
| mock.call( |
| "dir-%d" % part.number, |
| os.path.join( |
| lb.destination, "dir-%s" % part.name |
| ), |
| sudo=True, |
| ) |
| ) |
| break |
| self.rc_mock.assertCommandContains( |
| ["losetup", "--show", "-f", FAKE_PATH] |
| ) |
| self.delpart_mock.assert_called_once() |
| self.delpart_mock.reset_mock() |
| self.addpart_mock.assert_called_once() |
| self.detach_mock.assert_not_called() |
| self.assertEqual(lb.parts, LOOP_PARTS_DICT) |
| self.assertEqual(lb._gpt_table, LOOP_PARTITION_INFO) |
| self.assertEqual(expected_calls, syml.call_args_list) |
| self.assertEqual(expected_mounts, lb._mounted) |
| self.delpart_mock.assert_called_once() |
| self.detach_mock.assert_called_once() |
| |
| def testManual(self) -> None: |
| """Test using the loopback class closed manually.""" |
| lb = image_lib.LoopbackPartitions(FAKE_PATH) |
| lb.Attach() |
| self.rc_mock.assertCommandContains( |
| ["losetup", "--show", "-f", FAKE_PATH] |
| ) |
| self.delpart_mock.assert_called_once() |
| self.delpart_mock.reset_mock() |
| self.addpart_mock.assert_called_once() |
| self.detach_mock.assert_not_called() |
| self.assertEqual(lb.parts, LOOP_PARTS_DICT) |
| self.assertEqual(lb._gpt_table, LOOP_PARTITION_INFO) |
| lb.close() |
| self.delpart_mock.assert_called_once() |
| self.detach_mock.assert_called_once() |
| |
| def gcFunc(self) -> None: |
| """Isolates a local variable so it'll be garbage collected.""" |
| lb = image_lib.LoopbackPartitions(FAKE_PATH) |
| lb.Attach() |
| self.rc_mock.assertCommandContains( |
| ["losetup", "--show", "-f", FAKE_PATH] |
| ) |
| self.delpart_mock.assert_called_once() |
| self.delpart_mock.reset_mock() |
| self.addpart_mock.assert_called_once() |
| self.detach_mock.assert_not_called() |
| self.assertEqual(lb.parts, LOOP_PARTS_DICT) |
| self.assertEqual(lb._gpt_table, LOOP_PARTITION_INFO) |
| |
| def testGarbageCollected(self) -> None: |
| """Test using the loopback class closed by garbage collection.""" |
| self.gcFunc() |
| # Force garbage collection in case python didn't already clean up the |
| # loopback object. |
| gc.collect() |
| self.delpart_mock.assert_called_once() |
| self.detach_mock.assert_called_once() |
| |
| def testMountUnmount(self) -> None: |
| """Test Mount() and Unmount() entry points.""" |
| lb = image_lib.LoopbackPartitions(FAKE_PATH, destination=self.tempdir) |
| lb.Attach() |
| # Mount four partitions. |
| lb.Mount((1, 3, "ROOT-B", "ROOT-C")) |
| for p in (1, 3, 5, 7): |
| self.mount_mock.assert_any_call( |
| "%sp%d" % (LOOP_DEV, p), |
| "%s/dir-%d" % (self.tempdir, p), |
| makedirs=True, |
| skip_mtab=False, |
| sudo=True, |
| mount_opts=("ro",), |
| ) |
| linkname = "%s/dir-%s" % ( |
| self.tempdir, |
| LOOP_PARTITION_INFO[p - 1].name, |
| ) |
| self.assertTrue(stat.S_ISLNK(os.lstat(linkname).st_mode)) |
| self.assertEqual(4, self.mount_mock.call_count) |
| self.umount_mock.assert_not_called() |
| |
| # Check that Mounted provides the right info. |
| mounts = lb.Mounted() |
| self.assertDictEqual( |
| mounts, |
| { |
| "STATE": f"{self.tempdir}/dir-1", |
| "ROOT-A": f"{self.tempdir}/dir-3", |
| "ROOT-B": f"{self.tempdir}/dir-5", |
| "ROOT-C": f"{self.tempdir}/dir-7", |
| }, |
| ) |
| |
| # Unmount half of them, confirm that they were unmounted. |
| lb.Unmount((1, "ROOT-B")) |
| for p in (1, 5): |
| self.umount_mock.assert_any_call( |
| "%s/dir-%d" % (self.tempdir, p), cleanup=False |
| ) |
| self.assertEqual(2, self.umount_mock.call_count) |
| self.umount_mock.reset_mock() |
| |
| # Check that Mounted has been updated |
| mounts = lb.Mounted() |
| self.assertDictEqual( |
| mounts, |
| { |
| "ROOT-A": f"{self.tempdir}/dir-3", |
| "ROOT-C": f"{self.tempdir}/dir-7", |
| }, |
| ) |
| |
| # Close the object, so that we unmount the other half of them. |
| lb.close() |
| for p in (3, 7): |
| self.umount_mock.assert_any_call( |
| "%s/dir-%d" % (self.tempdir, p), cleanup=False |
| ) |
| self.assertEqual(2, self.umount_mock.call_count) |
| |
| # Verify that the directories were cleaned up. |
| for p in (1, 3): |
| self.retry_mock.assert_any_call( |
| cros_build_lib.RunCommandError, |
| 60, |
| osutils.RmDir, |
| "%s/dir-%d" % (self.tempdir, p), |
| sudo=True, |
| sleep=1, |
| ) |
| |
| def testMountingMountedPartReturnsName(self) -> None: |
| """Verify Mount returns the directory name even when already mounted.""" |
| lb = image_lib.LoopbackPartitions(FAKE_PATH, destination=self.tempdir) |
| lb.Attach() |
| dirname = "%s/dir-%d" % (self.tempdir, lb._gpt_table[0].number) |
| # First make sure we get the directory name when we actually mount. |
| self.assertEqual(dirname, lb._Mount(lb._gpt_table[0], ("ro",))) |
| # Then make sure we get it when we call it again. |
| self.assertEqual(dirname, lb._Mount(lb._gpt_table[0], ("ro",))) |
| lb.close() |
| |
| def testRemountCallsMount(self) -> None: |
| """Verify Mount returns the directory name even when already mounted.""" |
| lb = image_lib.LoopbackPartitions(FAKE_PATH, destination=self.tempdir) |
| lb.Attach() |
| devname = "%sp%d" % (LOOP_DEV, lb._gpt_table[0].number) |
| dirname = "%s/dir-%d" % (self.tempdir, lb._gpt_table[0].number) |
| # First make sure we get the directory name when we actually mount. |
| self.assertEqual(dirname, lb._Mount(lb._gpt_table[0], ("ro",))) |
| self.mount_mock.assert_called_once_with( |
| devname, |
| dirname, |
| makedirs=True, |
| skip_mtab=False, |
| sudo=True, |
| mount_opts=("ro",), |
| ) |
| # Then make sure we get it when we call it again. |
| self.assertEqual( |
| dirname, lb._Mount(lb._gpt_table[0], ("remount", "rw")) |
| ) |
| self.assertEqual( |
| mock.call( |
| devname, |
| dirname, |
| makedirs=True, |
| skip_mtab=False, |
| sudo=True, |
| mount_opts=("remount", "rw"), |
| ), |
| self.mount_mock.call_args, |
| ) |
| lb.close() |
| |
| def testGetPartitionDevName(self) -> None: |
| """Test GetPartitionDevName().""" |
| lb = image_lib.LoopbackPartitions(FAKE_PATH) |
| lb.Attach() |
| for part in LOOP_PARTITION_INFO: |
| self.assertEqual( |
| "%sp%d" % (LOOP_DEV, part.number), |
| lb.GetPartitionDevName(part.number), |
| ) |
| if part.name != "reserved": |
| self.assertEqual( |
| "%sp%d" % (LOOP_DEV, part.number), |
| lb.GetPartitionDevName(part.name), |
| ) |
| lb.close() |
| |
| def test_GetMountPointAndSymlink(self) -> None: |
| """Test _GetMountPointAndSymlink().""" |
| lb = image_lib.LoopbackPartitions(FAKE_PATH, destination=self.tempdir) |
| lb.Attach() |
| for part in LOOP_PARTITION_INFO: |
| expected = [ |
| os.path.join(lb.destination, "dir-%s" % n) |
| for n in (part.number, part.name) |
| ] |
| self.assertEqual(expected, list(lb._GetMountPointAndSymlink(part))) |
| lb.close() |
| |
| def testIsExt2OnVarious(self) -> None: |
| """Test _IsExt2 works with the various partition types.""" |
| # STATE, ROOT-A, and OEM generally have ext2 filesystems. |
| FS_PARTITIONS = (1, 3, 8) |
| |
| def ext_mock(path, offset=0): # pylint: disable=unused-argument |
| for num in FS_PARTITIONS: |
| if path.endswith(f"p{num}"): |
| return True |
| return False |
| |
| self.PatchObject(image_lib, "IsExt2Image", side_effect=ext_mock) |
| |
| lb = image_lib.LoopbackPartitions(FAKE_PATH, destination=self.tempdir) |
| lb.Attach() |
| # We expect that only the partitions in FS_PARTITIONS are ext2. |
| self.assertEqual( |
| [part.number in FS_PARTITIONS for part in LOOP_PARTITION_INFO], |
| [lb._IsExt2(part.name) for part in LOOP_PARTITION_INFO], |
| ) |
| lb.close() |
| |
| |
| class LsbUtilsTest(cros_test_lib.RunCommandTempDirTestCase): |
| """Tests the various LSB utilities.""" |
| |
| def setUp(self) -> None: |
| # Patch osutils.IsRootUser() to pretend running as root, so |
| # reading/writing the lsb-release file doesn't require escalated |
| # privileges and the test can clean itself up correctly. |
| self.PatchObject(os_util, "is_root_user", return_value=True) |
| |
| def testWriteLsbRelease(self) -> None: |
| """Tests writing out the lsb_release file using WriteLsbRelease(..).""" |
| fields = collections.OrderedDict( |
| ( |
| ("x", "1"), |
| ("y", "2"), |
| ("foo", "bar"), |
| ) |
| ) |
| image_lib.WriteLsbRelease(self.tempdir, fields) |
| lsb_release_file = os.path.join(self.tempdir, "etc", "lsb-release") |
| expected_content = "x=1\ny=2\nfoo=bar\n" |
| self.assertFileContents(lsb_release_file, expected_content) |
| self.rc.assertCommandCalled( |
| [ |
| "setfattr", |
| "-n", |
| "security.selinux", |
| "-v", |
| "u:object_r:cros_conf_file:s0", |
| os.path.join(self.tempdir, "etc/lsb-release"), |
| ] |
| ) |
| |
| def testOverwriteLsbRelease(self) -> None: |
| """Tests overwriting the lsb_release file using WriteLsbRelease(..).""" |
| lsb_release_file = os.path.join(self.tempdir, "etc", "lsb-release") |
| |
| image_lib.WriteLsbRelease( |
| self.tempdir, |
| collections.OrderedDict( |
| ( |
| ("x", "1"), |
| ("y", "2"), |
| ("foo", "bar"), |
| ) |
| ), |
| ) |
| |
| # Test that WriteLsbRelease(..) correctly handles an existing file. |
| fields = collections.OrderedDict( |
| ( |
| ("newkey1", "value1"), |
| ("newkey2", "value2"), |
| ("a", "3"), |
| ("b", "4"), |
| ) |
| ) |
| image_lib.WriteLsbRelease(self.tempdir, fields) |
| expected_content = ( |
| "x=1\ny=2\nfoo=bar\nnewkey1=value1\nnewkey2=value2\na=3\nb=4\n" |
| ) |
| self.assertFileContents(lsb_release_file, expected_content) |
| self.rc.assertCommandCalled( |
| [ |
| "setfattr", |
| "-n", |
| "security.selinux", |
| "-v", |
| "u:object_r:cros_conf_file:s0", |
| os.path.join(self.tempdir, "etc/lsb-release"), |
| ] |
| ) |
| |
| |
| class BuildImagePathTest(cros_test_lib.MockTempDirTestCase): |
| """BuildImagePath tests.""" |
| |
| def setUp(self) -> None: |
| self.board = "board" |
| self.board_dir = os.path.join(self.tempdir, self.board) |
| |
| D = cros_test_lib.Directory |
| filesystem = ( |
| D(self.board, ("recovery_image.bin", "other_image.bin")), |
| "full_path_image.bin", |
| ) |
| cros_test_lib.CreateOnDiskHierarchy(self.tempdir, filesystem) |
| |
| self.full_path = os.path.join(self.tempdir, "full_path_image.bin") |
| |
| def testBuildImagePath(self) -> None: |
| """BuildImagePath tests.""" |
| self.PatchObject( |
| image_lib, |
| "GetLatestImageLink", |
| return_value=os.path.join(self.tempdir, self.board), |
| ) |
| |
| # Board and full image path provided. |
| result = image_lib.BuildImagePath(self.board, self.full_path) |
| self.assertEqual(self.full_path, result) |
| |
| # Only full image path provided. |
| result = image_lib.BuildImagePath(None, self.full_path) |
| self.assertEqual(self.full_path, result) |
| |
| # Full image path provided that does not exist. |
| with self.assertRaises(image_lib.ImageDoesNotExistError): |
| image_lib.BuildImagePath(self.board, "/does/not/exist") |
| with self.assertRaises(image_lib.ImageDoesNotExistError): |
| image_lib.BuildImagePath(None, "/does/not/exist") |
| |
| # Default image is used. |
| result = image_lib.BuildImagePath(self.board, None) |
| self.assertEqual( |
| os.path.join(self.board_dir, "recovery_image.bin"), result |
| ) |
| |
| # Image basename provided. |
| result = image_lib.BuildImagePath(self.board, "other_image.bin") |
| self.assertEqual( |
| os.path.join(self.board_dir, "other_image.bin"), result |
| ) |
| |
| # Image basename provided that does not exist. |
| with self.assertRaises(image_lib.ImageDoesNotExistError): |
| image_lib.BuildImagePath(self.board, "does_not_exist.bin") |
| |
| default_mock = self.PatchObject(cros_build_lib, "GetDefaultBoard") |
| |
| # Nothing provided, and no default. |
| default_mock.return_value = None |
| with self.assertRaises(image_lib.ImageDoesNotExistError): |
| image_lib.BuildImagePath(None, None) |
| |
| # Nothing provided, with default. |
| default_mock.return_value = "board" |
| result = image_lib.BuildImagePath(None, None) |
| self.assertEqual( |
| os.path.join(self.board_dir, "recovery_image.bin"), result |
| ) |
| |
| |
| class SecurityTestConfigTest(cros_test_lib.RunCommandTempDirTestCase): |
| """SecurityTestConfig class tests.""" |
| |
| # pylint: disable=protected-access |
| |
| def setUp(self) -> None: |
| self.image = "/path/to/image.bin" |
| self.baselines = "/path/to/baselines" |
| self.vboot_hash = "abc123" |
| self.config = image_lib.SecurityTestConfig( |
| self.image, self.baselines, self.vboot_hash, self.tempdir |
| ) |
| |
| def testVbootCheckout(self) -> None: |
| """Test normal flow - clone and checkout.""" |
| clone_patch = self.PatchObject(git, "Clone") |
| self.config._VbootCheckout() |
| clone_patch.assert_called_once() |
| self.assertCommandContains(["git", "checkout", self.vboot_hash]) |
| |
| # Make sure it doesn't try to clone & checkout again after already |
| # having done so successfully. |
| clone_patch = self.PatchObject(git, "Clone") |
| self.config._VbootCheckout() |
| clone_patch.assert_not_called() |
| |
| def testVbootCheckoutError(self) -> None: |
| """Test exceptions in a git command.""" |
| rce = cros_build_lib.RunCommandError("error") |
| self.PatchObject(git, "Clone", side_effect=rce) |
| with self.assertRaises(image_lib.VbootCheckoutError): |
| self.config._VbootCheckout() |
| |
| def testVbootCheckoutNoDirectory(self) -> None: |
| """Test the error handling when the directory does not exist.""" |
| # Test directory that does not exist. |
| self.config.directory = "/DOES/NOT/EXIST" |
| with self.assertRaises(image_lib.SecurityConfigDirectoryError): |
| self.config._VbootCheckout() |
| |
| def testRunCheck(self) -> None: |
| """RunCheck tests.""" |
| # No config argument when running check. |
| self.config.RunCheck("check1", False) |
| check1 = os.path.join(self.config._checks_dir, "ensure_check1.sh") |
| config1 = os.path.join(self.baselines, "ensure_check1.config") |
| self.assertCommandContains([check1, self.image]) |
| self.assertCommandContains([config1], expected=False) |
| |
| # Include config argument when running check. |
| self.config.RunCheck("check2", True) |
| check2 = os.path.join(self.config._checks_dir, "ensure_check2.sh") |
| config2 = os.path.join(self.baselines, "ensure_check2.config") |
| self.assertCommandContains([check2, self.image, config2]) |
| |
| |
| class GetImageDiskPartitionInfoTests(cros_test_lib.RunCommandTempDirTestCase): |
| """Tests the GetImageDiskPartitionInfo function.""" |
| |
| SAMPLE_PARTED = """/foo/chromiumos_qemu_image.bin:\ |
| 2271240192B:file:512:512:gpt::; |
| 11:32768B:8421375B:8388608B::RWFW:; |
| 6:8421376B:8421887B:512B::KERN-C:; |
| 7:8421888B:8422399B:512B::ROOT-C:; |
| 9:8422400B:8422911B:512B::reserved:; |
| 10:8422912B:8423423B:512B::reserved:; |
| 2:10485760B:27262975B:16777216B::KERN-A:; |
| 4:27262976B:44040191B:16777216B::KERN-B:; |
| 8:44040192B:60817407B:16777216B:ext4:OEM:msftdata; |
| 12:127926272B:161480703B:33554432B:fat16:EFI-SYSTEM:boot, esp; |
| 5:161480704B:163577855B:2097152B::ROOT-B:; |
| 3:163577856B:2260729855B:2097152000B:ext2:ROOT-A:; |
| 1:2260729856B:2271215615B:10485760B:ext2:STATE:msftdata; |
| """ |
| |
| SAMPLE_CGPT = """\ |
| start size part contents |
| 0 1 PMBR (Boot GUID: 88FB7EB8-2B3F-B943-B933-\ |
| EEC571FFB6E1) |
| 1 1 Pri GPT header |
| 2 32 Pri GPT table |
| 1921024 2097152 1 Label: "STATE" |
| Type: 0FC63DAF-8483-4772-8E79-3D69D8477DE4 |
| UUID: EEBD83BE-397E-BD44-878B-0DDDD5A5C510 |
| Attr: [0] |
| 20480 32768 2 Label: "KERN-A" |
| Type: FE3A2A5D-4F32-41A7-B725-ACCC3285A309 |
| UUID: 7007C2F3-08E5-AB40-A4BC-FF5B01F5460D |
| Attr: [101] |
| 1101824 819200 3 Label: "ROOT-A" |
| Type: 3CB8E202-3B7E-47DD-8A3C-7FF2A13CFCEC |
| UUID: F4C5C3AD-027F-894B-80CD-3DEC57932948 |
| Attr: [0] |
| 53248 32768 4 Label: "KERN-B" |
| Type: FE3A2A5D-4F32-41A7-B725-ACCC3285A309 |
| UUID: C85FB478-404C-8741-ADB8-11312A35880D |
| Attr: [0] |
| 282624 819200 5 Label: "ROOT-B" |
| Type: 3CB8E202-3B7E-47DD-8A3C-7FF2A13CFCEC |
| UUID: A99F4231-1EC3-C542-AC0C-DF3729F5DB07 |
| Attr: [0] |
| 16448 1 6 Label: "KERN-C" |
| Type: FE3A2A5D-4F32-41A7-B725-ACCC3285A309 |
| UUID: 81F0E336-FAC9-174D-A08C-864FE627B637 |
| Attr: [0] |
| 16449 1 7 Label: "ROOT-C" |
| Type: 3CB8E202-3B7E-47DD-8A3C-7FF2A13CFCEC |
| UUID: 9E127FCA-30C1-044E-A5F2-DF74E6932692 |
| Attr: [0] |
| 86016 32768 8 Label: "OEM" |
| Type: 0FC63DAF-8483-4772-8E79-3D69D8477DE4 |
| UUID: 72986347-A37C-684F-9A19-4DBAF41C55A9 |
| Attr: [0] |
| 16450 1 9 Label: "reserved" |
| Type: 2E0A753D-9E48-43B0-8337-B15192CB1B5E |
| UUID: BA85A0A7-1850-964D-8EF8-6707AC106C3A |
| Attr: [0] |
| 16451 1 10 Label: "reserved" |
| Type: 2E0A753D-9E48-43B0-8337-B15192CB1B5E |
| UUID: 16C9EC9B-50FA-DD46-98DC-F781360817B4 |
| Attr: [0] |
| 64 16384 11 Label: "RWFW" |
| Type: CAB6E88E-ABF3-4102-A07A-D4BB9BE3C1D3 |
| UUID: BE8AECB9-4F78-7C44-8F23-5A9273B7EC8F |
| Attr: [0] |
| 249856 32768 12 Label: "EFI-SYSTEM" |
| Type: C12A7328-F81F-11D2-BA4B-00A0C93EC93B |
| UUID: 88FB7EB8-2B3F-B943-B933-EEC571FFB6E1 |
| Attr: [0] |
| 4050847 32 Sec GPT table |
| 4050879 1 Sec GPT header |
| """ |
| |
| def setUp(self) -> None: |
| self.WriteTempFile("foo", "non-empty file") |
| self._image_path = os.path.join(self.tempdir, "foo") |
| |
| def testCgpt(self) -> None: |
| """Tests that we can list all partitions with `cgpt` correctly.""" |
| self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=True) |
| self.rc.AddCmdResult(partial_mock.Ignore(), stdout=self.SAMPLE_CGPT) |
| partitions = image_lib.GetImageDiskPartitionInfo(self._image_path) |
| part_dict = {p.name: p for p in partitions} |
| self.assertEqual(part_dict["STATE"].start, 983564288) |
| self.assertEqual(part_dict["STATE"].size, 1073741824) |
| self.assertEqual(part_dict["STATE"].number, 1) |
| self.assertEqual(part_dict["STATE"].name, "STATE") |
| self.assertEqual(part_dict["EFI-SYSTEM"].start, 249856 * 512) |
| self.assertEqual(part_dict["EFI-SYSTEM"].size, 32768 * 512) |
| self.assertEqual(part_dict["EFI-SYSTEM"].number, 12) |
| self.assertEqual(part_dict["EFI-SYSTEM"].name, "EFI-SYSTEM") |
| self.assertEqual(12, len(partitions)) |
| |
| def testNormalPath(self) -> None: |
| self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False) |
| self.rc.AddCmdResult(partial_mock.Ignore(), stdout=self.SAMPLE_PARTED) |
| partitions = image_lib.GetImageDiskPartitionInfo(self._image_path) |
| part_dict = {p.name: p for p in partitions} |
| self.assertEqual(12, len(partitions)) |
| self.assertEqual(1, part_dict["STATE"].number) |
| self.assertEqual(2097152000, part_dict["ROOT-A"].size) |
| |
| def testKeyedByNumber(self) -> None: |
| self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False) |
| self.rc.AddCmdResult(partial_mock.Ignore(), stdout=self.SAMPLE_PARTED) |
| partitions = image_lib.GetImageDiskPartitionInfo(self._image_path) |
| part_dict = {p.number: p for p in partitions} |
| self.assertEqual(12, len(part_dict)) |
| self.assertEqual("STATE", part_dict[1].name) |
| self.assertEqual(2097152000, part_dict[3].size) |
| self.assertEqual("reserved", part_dict[9].name) |
| self.assertEqual("reserved", part_dict[10].name) |
| |
| def testChangeUnitInsideChroot(self) -> None: |
| self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=True) |
| self.rc.AddCmdResult(partial_mock.Ignore(), stdout=self.SAMPLE_CGPT) |
| partitions = image_lib.GetImageDiskPartitionInfo(self._image_path) |
| part_dict = {p.name: p for p in partitions} |
| self.assertEqual(part_dict["STATE"].start, 983564288) |
| self.assertEqual(part_dict["STATE"].size, 1073741824) |
| |
| |
| class GetImagesToBuildTests(cros_test_lib.MockTestCase): |
| """Tests the GetImagesToBuild function.""" |
| |
| def testExpectedInput(self) -> None: |
| """Pass in all expected image types and check expected image names.""" |
| # TODO(b/236161656): Fix. |
| # pylint: disable-next=consider-using-dict-items |
| for k in constants.IMAGE_TYPE_TO_NAME: |
| image = image_lib.GetImagesToBuild([k]) |
| self.assertEqual(len(image), 1) |
| self.assertTrue(constants.IMAGE_TYPE_TO_NAME[k] in image) |
| |
| def testInvalidInput(self) -> None: |
| """Pass in an invalid image type and check for ValueError.""" |
| with self.assertRaises(ValueError): |
| image_lib.GetImagesToBuild([constants.IMAGE_TYPE_DEV, "invalid"]) |
| |
| def testInvalidImageCombination(self) -> None: |
| """Verify an invalid image type combination raises a ValueError.""" |
| with self.assertRaises(ValueError): |
| image_lib.GetImagesToBuild( |
| [constants.IMAGE_TYPE_DEV, constants.FACTORY_IMAGE_BIN] |
| ) |
| |
| |
| class GetBuildImageEnvvarTests(cros_test_lib.MockTestCase): |
| """Tests the GetBuildImageEnvvars function.""" |
| |
| def setUp(self) -> None: |
| self.use_flag_mock = self.PatchObject( |
| portage_util, "GetBoardUseFlags", return_value=[] |
| ) |
| |
| def testStandardImage(self) -> None: |
| """Test with standard base/dev/test image name.""" |
| expected_envvar = { |
| "INSTALL_MASK": ( |
| "\n".join(install_mask.DEFAULT) |
| + "\n" |
| + "\n".join(install_mask.SYSTEMD) |
| ), |
| "PRISTINE_IMAGE_NAME": constants.BASE_IMAGE_BIN, |
| "BASE_PACKAGE": "virtual/target-os", |
| } |
| image_to_test = [ |
| constants.BASE_IMAGE_BIN, |
| constants.DEV_IMAGE_BIN, |
| constants.TEST_IMAGE_BIN, |
| ] |
| for image in image_to_test: |
| envar = image_lib.GetBuildImageEnvvars(set([image]), "test_board") |
| self.assertDictEqual(envar, expected_envvar) |
| |
| # Validate scenario with systemd in USE flag |
| self.use_flag_mock.return_value = ["cros_debug", "systemd"] |
| expected_envvar["INSTALL_MASK"] = "\n".join(install_mask.DEFAULT) |
| for image in image_to_test: |
| envar = image_lib.GetBuildImageEnvvars(set([image]), "test_board") |
| self.assertDictEqual(envar, expected_envvar) |
| |
| def testFactoryImage(self) -> None: |
| """Test with factory image name.""" |
| expected_envvar = { |
| "INSTALL_MASK": ( |
| "\n".join(install_mask.FACTORY_SHIM) |
| + "\n" |
| + "\n".join(install_mask.SYSTEMD) |
| ), |
| "USE": image_lib._FACTORY_SHIM_USE_FLAGS, |
| "PRISTINE_IMAGE_NAME": constants.FACTORY_IMAGE_BIN, |
| "BASE_PACKAGE": "virtual/target-os-factory-shim", |
| } |
| envar = image_lib.GetBuildImageEnvvars( |
| set([constants.FACTORY_IMAGE_BIN]), "betty" |
| ) |
| self.assertDictEqual(envar, expected_envvar) |
| |
| # Validate scenario with systemd in USE flag |
| self.use_flag_mock.return_value = ["cros_debug", "systemd"] |
| expected_envvar["INSTALL_MASK"] = "\n".join(install_mask.FACTORY_SHIM) |
| envar.clear() |
| envar = image_lib.GetBuildImageEnvvars( |
| set([constants.FACTORY_IMAGE_BIN]), "betty" |
| ) |
| print(envar) |
| self.assertDictEqual(envar, expected_envvar) |
| |
| # Validate if extra environment variable is passed |
| extra_env = { |
| "USE": "test test1", |
| "ENV": "TEST_VALUE", |
| } |
| expected_envvar["USE"] = extra_env["USE"] + " " + expected_envvar["USE"] |
| expected_envvar["ENV"] = extra_env["ENV"] |
| envar.clear() |
| envar = image_lib.GetBuildImageEnvvars( |
| set([constants.FACTORY_IMAGE_BIN]), "betty", env_var_init=extra_env |
| ) |
| self.assertDictEqual(envar, expected_envvar) |
| |
| def testChromeOSVersion(self) -> None: |
| """Test ChromeOS version environment variable.""" |
| version_info = chromeos_version.VersionInfo( |
| version_string="1.2.3", chrome_branch="4" |
| ) |
| envar = image_lib.GetBuildImageEnvvars( |
| set([constants.BASE_IMAGE_BIN]), "betty", version_info=version_info |
| ) |
| |
| self.assertEqual(envar["CHROME_BRANCH"], "4") |
| self.assertEqual(envar["CHROMEOS_BUILD"], "1") |
| self.assertEqual(envar["CHROMEOS_BRANCH"], "2") |
| self.assertEqual(envar["CHROMEOS_PATCH"], "3") |
| self.assertEqual(envar["CHROMEOS_VERSION_STRING"], "1.2.3") |
| |
| def testBuildAndOutputDir(self) -> None: |
| """Test BUILD_DIR and OUTPUT_DIR environment variable.""" |
| build_dir = "build/dir" |
| output_dir = Path("ouput/dir") |
| envar = image_lib.GetBuildImageEnvvars( |
| set([constants.BASE_IMAGE_BIN]), |
| "betty", |
| build_dir=build_dir, |
| output_dir=output_dir, |
| ) |
| |
| self.assertEqual(envar["BUILD_DIR"], build_dir) |
| self.assertEqual(envar["OUTPUT_DIR"], str(output_dir)) |
| |
| |
| class CreateBuildDirTests(cros_test_lib.MockTempDirTestCase): |
| """Test CreateBuildDir.""" |
| |
| def setUp(self) -> None: |
| self.PatchObject( |
| chromeos_version.VersionInfo, |
| "_GetDateTime", |
| return_value=FAKE_DATE_STRING, |
| ) |
| self.build_top_dir = self.tempdir / "build" |
| self.output_top_dir = self.tempdir / "output" |
| self.testBoard = "TestBoard" |
| self.version_info = chromeos_version.VersionInfo( |
| version_string="1.2.3", chrome_branch="4" |
| ) |
| self.attempt = 5 |
| self.result_build_dir = self.build_top_dir / self.testBoard |
| self.result_output_dir = self.output_top_dir / self.testBoard |
| self.image_dir = ( |
| f"R{self.version_info.chrome_branch}-" |
| + f"{self.version_info.VersionString()}" |
| ) |
| self.image_dir_attempt = self.image_dir + f"-a{self.attempt}" |
| self.image_dir_date = ( |
| f"R{self.version_info.chrome_branch}-" |
| + f"{self.version_info.VersionStringWithDateTime()}" |
| ) |
| self.image_dir_date_attempt = f"{self.image_dir_date}-a{self.attempt}" |
| self.symlink = "latest" |
| |
| def testChromeBranchVersion(self) -> None: |
| """Test with chrome_branch and version string.""" |
| build_dir, output_dir, symlink_dir = image_lib.CreateBuildDir( |
| self.build_top_dir, |
| self.output_top_dir, |
| self.version_info.chrome_branch, |
| self.version_info.VersionString(), |
| self.testBoard, |
| self.symlink, |
| ) |
| |
| self.assertExists(build_dir) |
| self.assertExists(output_dir) |
| self.assertEqual(build_dir, self.result_build_dir / self.image_dir) |
| self.assertExists(output_dir, self.result_output_dir / self.image_dir) |
| self.assertTrue(symlink_dir.is_symlink()) |
| self.assertEqual(self.image_dir, os.readlink(symlink_dir)) |
| |
| # Now test the case where the build directory already exists. |
| build_dir, output_dir, symlink_dir = image_lib.CreateBuildDir( |
| self.build_top_dir, |
| self.output_top_dir, |
| self.version_info.chrome_branch, |
| self.version_info.VersionString(), |
| self.testBoard, |
| self.symlink, |
| replace=True, |
| ) |
| |
| self.assertExists(build_dir) |
| self.assertExists(output_dir) |
| self.assertEqual(build_dir, self.result_build_dir / self.image_dir) |
| self.assertExists(output_dir, self.result_output_dir / self.image_dir) |
| self.assertTrue(symlink_dir.is_symlink()) |
| self.assertEqual(self.image_dir, os.readlink(symlink_dir)) |
| |
| # Now test the case where the build directory already exists with |
| # replace as false. |
| with self.assertRaises(FileExistsError): |
| build_dir, output_dir, symlink_dir = image_lib.CreateBuildDir( |
| self.build_top_dir, |
| self.output_top_dir, |
| self.version_info.chrome_branch, |
| self.version_info.VersionString(), |
| self.testBoard, |
| self.symlink, |
| ) |
| |
| def testChromeBranchVersionDate(self) -> None: |
| """Test with chrome_branch and version string with date.""" |
| build_dir, output_dir, symlink_dir = image_lib.CreateBuildDir( |
| self.build_top_dir, |
| self.output_top_dir, |
| self.version_info.chrome_branch, |
| self.version_info.VersionStringWithDateTime(), |
| self.testBoard, |
| self.symlink, |
| ) |
| |
| self.assertExists(build_dir) |
| self.assertExists(output_dir) |
| self.assertEqual(build_dir, self.result_build_dir / self.image_dir_date) |
| self.assertExists( |
| output_dir, self.result_output_dir / self.image_dir_date |
| ) |
| self.assertTrue(symlink_dir.is_symlink()) |
| self.assertEqual(self.image_dir_date, os.readlink(symlink_dir)) |
| |
| def testBuildAttempt(self) -> None: |
| """Test with chrome_branch, version string and build attempt.""" |
| build_dir, output_dir, symlink_dir = image_lib.CreateBuildDir( |
| self.build_top_dir, |
| self.output_top_dir, |
| self.version_info.chrome_branch, |
| self.version_info.VersionString(), |
| self.testBoard, |
| self.symlink, |
| build_attempt=self.attempt, |
| ) |
| |
| self.assertExists(build_dir) |
| self.assertExists(output_dir) |
| self.assertEqual( |
| build_dir, self.result_build_dir / self.image_dir_attempt |
| ) |
| self.assertExists( |
| output_dir, self.result_output_dir / self.image_dir_attempt |
| ) |
| self.assertTrue(symlink_dir.is_symlink()) |
| self.assertEqual(self.image_dir_attempt, os.readlink(symlink_dir)) |
| |
| def testBuildAttemptDate(self) -> None: |
| """Test with chrome_branch, version string, date, and build attempt.""" |
| build_dir, output_dir, symlink_dir = image_lib.CreateBuildDir( |
| self.build_top_dir, |
| self.output_top_dir, |
| self.version_info.chrome_branch, |
| self.version_info.VersionStringWithDateTime(), |
| self.testBoard, |
| self.symlink, |
| build_attempt=self.attempt, |
| ) |
| |
| self.assertExists(build_dir) |
| self.assertExists(output_dir) |
| self.assertEqual( |
| build_dir, self.result_build_dir / self.image_dir_date_attempt |
| ) |
| self.assertExists( |
| output_dir, self.result_output_dir / self.image_dir_date_attempt |
| ) |
| self.assertTrue(symlink_dir.is_symlink()) |
| self.assertEqual(self.image_dir_date_attempt, os.readlink(symlink_dir)) |
| |
| def testOutputSuffix(self) -> None: |
| """Test with output suffix.""" |
| output_suffix = "test-suffix" |
| build_dir, output_dir, symlink_dir = image_lib.CreateBuildDir( |
| self.build_top_dir, |
| self.output_top_dir, |
| self.version_info.chrome_branch, |
| self.version_info.VersionString(), |
| self.testBoard, |
| self.symlink, |
| build_attempt=self.attempt, |
| output_suffix=output_suffix, |
| ) |
| |
| self.assertExists(build_dir) |
| self.assertExists(output_dir) |
| self.assertEqual( |
| build_dir, |
| self.result_build_dir |
| / (self.image_dir_attempt + "-" + output_suffix), |
| ) |
| self.assertExists( |
| output_dir, |
| self.result_output_dir |
| / (self.image_dir_attempt + "-" + output_suffix), |
| ) |
| self.assertTrue(symlink_dir.is_symlink()) |
| self.assertEqual( |
| self.image_dir_attempt + "-" + output_suffix, |
| os.readlink(symlink_dir), |
| ) |
| |
| # Test the case without build_attempt. |
| build_dir, output_dir, symlink_dir = image_lib.CreateBuildDir( |
| self.build_top_dir, |
| self.output_top_dir, |
| self.version_info.chrome_branch, |
| self.version_info.VersionString(), |
| self.testBoard, |
| self.symlink, |
| output_suffix=output_suffix, |
| ) |
| |
| self.assertExists(build_dir) |
| self.assertExists(output_dir) |
| self.assertEqual( |
| build_dir, |
| self.result_build_dir / (self.image_dir + "-" + output_suffix), |
| ) |
| self.assertExists( |
| output_dir, |
| self.result_output_dir / (self.image_dir + "-" + output_suffix), |
| ) |
| self.assertTrue(symlink_dir.is_symlink()) |
| self.assertEqual( |
| self.image_dir + "-" + output_suffix, os.readlink(symlink_dir) |
| ) |
| |
| def testOutputSuffixWithDate(self) -> None: |
| """Test with output suffix with date.""" |
| output_suffix = "test-suffix" |
| build_dir, output_dir, symlink_dir = image_lib.CreateBuildDir( |
| self.build_top_dir, |
| self.output_top_dir, |
| self.version_info.chrome_branch, |
| self.version_info.VersionStringWithDateTime(), |
| self.testBoard, |
| self.symlink, |
| build_attempt=self.attempt, |
| output_suffix=output_suffix, |
| ) |
| |
| self.assertExists(build_dir) |
| self.assertExists(output_dir) |
| self.assertEqual( |
| build_dir, |
| self.result_build_dir |
| / f"{self.image_dir_date_attempt}-{output_suffix}", |
| ) |
| self.assertExists( |
| output_dir, |
| self.result_output_dir |
| / f"{self.image_dir_date_attempt}-{output_suffix}", |
| ) |
| self.assertTrue(symlink_dir.is_symlink()) |
| self.assertEqual( |
| f"{self.image_dir_date_attempt}-{output_suffix}", |
| os.readlink(symlink_dir), |
| ) |
| |
| # Test the case without build_attempt. |
| build_dir, output_dir, symlink_dir = image_lib.CreateBuildDir( |
| self.build_top_dir, |
| self.output_top_dir, |
| self.version_info.chrome_branch, |
| self.version_info.VersionStringWithDateTime(), |
| self.testBoard, |
| self.symlink, |
| output_suffix=output_suffix, |
| ) |
| |
| self.assertExists(build_dir) |
| self.assertExists(output_dir) |
| self.assertEqual( |
| build_dir, |
| self.result_build_dir / f"{self.image_dir_date}-{output_suffix}", |
| ) |
| self.assertExists( |
| output_dir, |
| self.result_output_dir / f"{self.image_dir_date}-{output_suffix}", |
| ) |
| self.assertTrue(symlink_dir.is_symlink()) |
| self.assertEqual( |
| f"{self.image_dir_date}-{output_suffix}", os.readlink(symlink_dir) |
| ) |
| |
| |
| class UtilsTests(cros_test_lib.TempDirTestCase): |
| """Test simple util funcs.""" |
| |
| def testIsSquashfsImageFails(self) -> None: |
| """Test SquashFS identification on non-images.""" |
| image = self.tempdir / "img.squashfs" |
| osutils.AllocateFile(image, 1024 * 1024) |
| self.assertFalse(image_lib.IsSquashfsImage(image)) |
| |
| def testIsSquashfsImage(self) -> None: |
| """Tests we correctly identify a SquashFS image.""" |
| image = self.tempdir / "img.squashfs" |
| root = self.tempdir / "root" |
| root.mkdir() |
| cros_build_lib.run( |
| ["mksquashfs", root.name, image.name], cwd=self.tempdir |
| ) |
| self.assertTrue(image_lib.IsSquashfsImage(image)) |
| |
| def testIsExt4Image(self) -> None: |
| """Tests we correctly identify an Ext4 image.""" |
| for ver in (2, 3, 4): |
| image = self.tempdir / f"rootfs.ext{ver}" |
| # 2 MiB is big enough for ext3/ext4 specific features. |
| osutils.AllocateFile(image, 2 * 1024 * 1024) |
| |
| # Tests failure to identify. |
| self.assertFalse(image_lib.IsExt2Image(image)) |
| |
| # Make a real ext2/ext3/ext4 images. |
| cros_build_lib.run( |
| [f"mkfs.ext{ver}", image], |
| extra_env={"PATH": "/sbin:/usr/sbin:%s" % os.environ["PATH"]}, |
| ) |
| self.assertTrue(image_lib.IsExt2Image(image)) |
