diff options
Diffstat (limited to 'tests/unittests/config/test_cc_resizefs_vyos.py')
| -rw-r--r-- | tests/unittests/config/test_cc_resizefs_vyos.py | 490 |
1 files changed, 490 insertions, 0 deletions
diff --git a/tests/unittests/config/test_cc_resizefs_vyos.py b/tests/unittests/config/test_cc_resizefs_vyos.py new file mode 100644 index 00000000..c995e6aa --- /dev/null +++ b/tests/unittests/config/test_cc_resizefs_vyos.py @@ -0,0 +1,490 @@ +# This file is part of cloud-init. See LICENSE file for license information. + +import logging +from collections import namedtuple + +from cloudinit.config.cc_resizefs_vyos import ( + _resize_btrfs, + _resize_ext, + _resize_ufs, + _resize_xfs, + _resize_zfs, + can_skip_resize, + handle, + maybe_get_writable_device_path, +) +from cloudinit.subp import ProcessExecutionError +from tests.unittests.helpers import ( + CiTestCase, + mock, + skipUnlessJsonSchema, + util, + wrap_and_call, +) + +LOG = logging.getLogger(__name__) + + +class TestResizefs(CiTestCase): + with_logs = True + + def setUp(self): + super(TestResizefs, self).setUp() + self.name = "resizefs" + + @mock.patch("cloudinit.subp.subp") + def test_skip_ufs_resize(self, m_subp): + fs_type = "ufs" + resize_what = "/" + devpth = "/dev/da0p2" + err = ( + "growfs: requested size 2.0GB is not larger than the " + "current filesystem size 2.0GB\n" + ) + exception = ProcessExecutionError(stderr=err, exit_code=1) + m_subp.side_effect = exception + res = can_skip_resize(fs_type, resize_what, devpth) + self.assertTrue(res) + + @mock.patch("cloudinit.subp.subp") + def test_cannot_skip_ufs_resize(self, m_subp): + fs_type = "ufs" + resize_what = "/" + devpth = "/dev/da0p2" + m_subp.return_value = ( + "stdout: super-block backups (for fsck_ffs -b #) at:\n\n", + "growfs: no room to allocate last cylinder group; " + "leaving 364KB unused\n", + ) + res = can_skip_resize(fs_type, resize_what, devpth) + self.assertFalse(res) + + @mock.patch("cloudinit.subp.subp") + def test_cannot_skip_ufs_growfs_exception(self, m_subp): + fs_type = "ufs" + resize_what = "/" + devpth = "/dev/da0p2" + err = "growfs: /dev/da0p2 is not clean - run fsck.\n" + exception = ProcessExecutionError(stderr=err, exit_code=1) + m_subp.side_effect = exception + with self.assertRaises(ProcessExecutionError): + can_skip_resize(fs_type, resize_what, devpth) + + def test_can_skip_resize_ext(self): + self.assertFalse(can_skip_resize("ext", "/", "/dev/sda1")) + + def test_handle_noops_on_disabled(self): + """The handle function logs when the configuration disables resize.""" + cfg = {"resizefs_enabled": False} + handle("cc_resizefs_vyos", cfg, _cloud=None, log=LOG, args=[]) + self.assertIn( + "DEBUG: Skipping module named cc_resizefs_vyos, resizing disabled\n", + self.logs.getvalue(), + ) + + @skipUnlessJsonSchema() + def test_handle_schema_validation_logs_invalid_resizefs_enabled_value(self): + """The handle reports json schema violations as a warning. + + Invalid values for resizefs_enabled result in disabling the module. + """ + cfg = {"resizefs_enabled": "junk"} + handle("cc_resizefs_vyos", cfg, _cloud=None, log=LOG, args=[]) + logs = self.logs.getvalue() + self.assertIn( + "WARNING: Invalid cloud-config provided:\nresizefs_enabled: 'junk' is" + " not one of [True, False, 'noblock']", + logs, + ) + self.assertIn( + "DEBUG: Skipping module named cc_resizefs_vyos, resizing disabled\n", + logs, + ) + + @mock.patch("cloudinit.config.cc_resizefs_vyos.util.get_mount_info") + def test_handle_warns_on_unknown_mount_info(self, m_get_mount_info): + """handle warns when get_mount_info sees unknown filesystem for /.""" + m_get_mount_info.return_value = None + cfg = {"resizefs_enabled": True} + handle("cc_resizefs_vyos", cfg, _cloud=None, log=LOG, args=[]) + logs = self.logs.getvalue() + self.assertNotIn( + "WARNING: Invalid cloud-config provided:\nresizefs_enabled:", logs + ) + self.assertIn( + "WARNING: Could not determine filesystem type of /\n", logs + ) + self.assertEqual( + [mock.call("/", LOG)], m_get_mount_info.call_args_list + ) + + def test_handle_warns_on_undiscoverable_root_path_in_commandline(self): + """handle noops when the root path is not found on the commandline.""" + cfg = {"resizefs_enabled": True} + exists_mock_path = "cloudinit.config.cc_resizefs_vyos.os.path.exists" + + def fake_mount_info(path, log): + self.assertEqual("/", path) + self.assertEqual(LOG, log) + return ("/dev/root", "ext4", "/") + + with mock.patch(exists_mock_path) as m_exists: + m_exists.return_value = False + wrap_and_call( + "cloudinit.config.cc_resizefs_vyos.util", + { + "is_container": {"return_value": False}, + "get_mount_info": {"side_effect": fake_mount_info}, + "get_cmdline": {"return_value": "BOOT_IMAGE=/vmlinuz.efi"}, + }, + handle, + "cc_resizefs_vyos", + cfg, + _cloud=None, + log=LOG, + args=[], + ) + logs = self.logs.getvalue() + self.assertIn("WARNING: Unable to find device '/dev/root'", logs) + + def test_resize_zfs_cmd_return(self): + zpool = "zroot" + devpth = "gpt/system" + self.assertEqual( + ("zpool", "online", "-e", zpool, devpth), + _resize_zfs(zpool, devpth), + ) + + def test_resize_xfs_cmd_return(self): + mount_point = "/mnt/test" + devpth = "/dev/sda1" + self.assertEqual( + ("xfs_growfs", mount_point), _resize_xfs(mount_point, devpth) + ) + + def test_resize_ext_cmd_return(self): + mount_point = "/" + devpth = "/dev/sdb1" + self.assertEqual( + ("resize2fs", devpth), _resize_ext(mount_point, devpth) + ) + + def test_resize_ufs_cmd_return(self): + mount_point = "/" + devpth = "/dev/sda2" + self.assertEqual( + ("growfs", "-y", mount_point), _resize_ufs(mount_point, devpth) + ) + + @mock.patch("cloudinit.util.is_container", return_value=False) + @mock.patch("cloudinit.util.parse_mount") + @mock.patch("cloudinit.util.get_device_info_from_zpool") + @mock.patch("cloudinit.util.get_mount_info") + def test_handle_zfs_root( + self, mount_info, zpool_info, parse_mount, is_container + ): + devpth = "vmzroot/ROOT/freebsd" + disk = "gpt/system" + fs_type = "zfs" + mount_point = "/" + + mount_info.return_value = (devpth, fs_type, mount_point) + zpool_info.return_value = disk + parse_mount.return_value = (devpth, fs_type, mount_point) + + cfg = {"resizefs_enabled": True} + + with mock.patch("cloudinit.config.cc_resizefs_vyos.do_resize") as dresize: + handle("cc_resizefs_vyos", cfg, _cloud=None, log=LOG, args=[]) + ret = dresize.call_args[0][0] + + self.assertEqual(("zpool", "online", "-e", "vmzroot", disk), ret) + + @mock.patch("cloudinit.util.is_container", return_value=False) + @mock.patch("cloudinit.util.get_mount_info") + @mock.patch("cloudinit.util.get_device_info_from_zpool") + @mock.patch("cloudinit.util.parse_mount") + def test_handle_modern_zfsroot( + self, mount_info, zpool_info, parse_mount, is_container + ): + devpth = "zroot/ROOT/default" + disk = "da0p3" + fs_type = "zfs" + mount_point = "/" + + mount_info.return_value = (devpth, fs_type, mount_point) + zpool_info.return_value = disk + parse_mount.return_value = (devpth, fs_type, mount_point) + + cfg = {"resizefs_enabled": True} + + def fake_stat(devpath): + if devpath == disk: + raise OSError("not here") + FakeStat = namedtuple( + "FakeStat", ["st_mode", "st_size", "st_mtime"] + ) # minimal stat + return FakeStat(25008, 0, 1) # fake char block device + + with mock.patch("cloudinit.config.cc_resizefs_vyos.do_resize") as dresize: + with mock.patch("cloudinit.config.cc_resizefs_vyos.os.stat") as m_stat: + m_stat.side_effect = fake_stat + handle("cc_resizefs_vyos", cfg, _cloud=None, log=LOG, args=[]) + + self.assertEqual( + ("zpool", "online", "-e", "zroot", "/dev/" + disk), + dresize.call_args[0][0], + ) + + +class TestRootDevFromCmdline(CiTestCase): + def test_rootdev_from_cmdline_with_no_root(self): + """Return None from rootdev_from_cmdline when root is not present.""" + invalid_cases = [ + "BOOT_IMAGE=/adsf asdfa werasef root adf", + "BOOT_IMAGE=/adsf", + "", + ] + for case in invalid_cases: + self.assertIsNone(util.rootdev_from_cmdline(case)) + + def test_rootdev_from_cmdline_with_root_startswith_dev(self): + """Return the cmdline root when the path starts with /dev.""" + self.assertEqual( + "/dev/this", util.rootdev_from_cmdline("asdf root=/dev/this") + ) + + def test_rootdev_from_cmdline_with_root_without_dev_prefix(self): + """Add /dev prefix to cmdline root when the path lacks the prefix.""" + self.assertEqual( + "/dev/this", util.rootdev_from_cmdline("asdf root=this") + ) + + def test_rootdev_from_cmdline_with_root_with_label(self): + """When cmdline root contains a LABEL, our root is disk/by-label.""" + self.assertEqual( + "/dev/disk/by-label/unique", + util.rootdev_from_cmdline("asdf root=LABEL=unique"), + ) + + def test_rootdev_from_cmdline_with_root_with_uuid(self): + """When cmdline root contains a UUID, our root is disk/by-uuid.""" + self.assertEqual( + "/dev/disk/by-uuid/adsfdsaf-adsf", + util.rootdev_from_cmdline("asdf root=UUID=adsfdsaf-adsf"), + ) + + +class TestMaybeGetDevicePathAsWritableBlock(CiTestCase): + + with_logs = True + + def test_maybe_get_writable_device_path_none_on_overlayroot(self): + """When devpath is overlayroot (on MAAS), is_dev_writable is False.""" + info = "does not matter" + devpath = wrap_and_call( + "cloudinit.config.cc_resizefs_vyos.util", + {"is_container": {"return_value": False}}, + maybe_get_writable_device_path, + "overlayroot", + info, + LOG, + ) + self.assertIsNone(devpath) + self.assertIn( + "Not attempting to resize devpath 'overlayroot'", + self.logs.getvalue(), + ) + + def test_maybe_get_writable_device_path_warns_missing_cmdline_root(self): + """When root does not exist isn't in the cmdline, log warning.""" + info = "does not matter" + + def fake_mount_info(path, log): + self.assertEqual("/", path) + self.assertEqual(LOG, log) + return ("/dev/root", "ext4", "/") + + exists_mock_path = "cloudinit.config.cc_resizefs_vyos.os.path.exists" + with mock.patch(exists_mock_path) as m_exists: + m_exists.return_value = False + devpath = wrap_and_call( + "cloudinit.config.cc_resizefs_vyos.util", + { + "is_container": {"return_value": False}, + "get_mount_info": {"side_effect": fake_mount_info}, + "get_cmdline": {"return_value": "BOOT_IMAGE=/vmlinuz.efi"}, + }, + maybe_get_writable_device_path, + "/dev/root", + info, + LOG, + ) + self.assertIsNone(devpath) + logs = self.logs.getvalue() + self.assertIn("WARNING: Unable to find device '/dev/root'", logs) + + def test_maybe_get_writable_device_path_does_not_exist(self): + """When devpath does not exist, a warning is logged.""" + info = "dev=/dev/I/dont/exist mnt_point=/ path=/dev/none" + devpath = wrap_and_call( + "cloudinit.config.cc_resizefs_vyos.util", + {"is_container": {"return_value": False}}, + maybe_get_writable_device_path, + "/dev/I/dont/exist", + info, + LOG, + ) + self.assertIsNone(devpath) + self.assertIn( + "WARNING: Device '/dev/I/dont/exist' did not exist." + " cannot resize: %s" % info, + self.logs.getvalue(), + ) + + def test_maybe_get_writable_device_path_does_not_exist_in_container(self): + """When devpath does not exist in a container, log a debug message.""" + info = "dev=/dev/I/dont/exist mnt_point=/ path=/dev/none" + devpath = wrap_and_call( + "cloudinit.config.cc_resizefs_vyos.util", + {"is_container": {"return_value": True}}, + maybe_get_writable_device_path, + "/dev/I/dont/exist", + info, + LOG, + ) + self.assertIsNone(devpath) + self.assertIn( + "DEBUG: Device '/dev/I/dont/exist' did not exist in container." + " cannot resize: %s" % info, + self.logs.getvalue(), + ) + + def test_maybe_get_writable_device_path_raises_oserror(self): + """When unexpected OSError is raises by os.stat it is reraised.""" + info = "dev=/dev/I/dont/exist mnt_point=/ path=/dev/none" + with self.assertRaises(OSError) as context_manager: + wrap_and_call( + "cloudinit.config.cc_resizefs_vyos", + { + "util.is_container": {"return_value": True}, + "os.stat": { + "side_effect": OSError("Something unexpected") + }, + }, + maybe_get_writable_device_path, + "/dev/I/dont/exist", + info, + LOG, + ) + self.assertEqual( + "Something unexpected", str(context_manager.exception) + ) + + def test_maybe_get_writable_device_path_non_block(self): + """When device is not a block device, emit warning return False.""" + fake_devpath = self.tmp_path("dev/readwrite") + util.write_file(fake_devpath, "", mode=0o600) # read-write + info = "dev=/dev/root mnt_point=/ path={0}".format(fake_devpath) + + devpath = wrap_and_call( + "cloudinit.config.cc_resizefs_vyos.util", + {"is_container": {"return_value": False}}, + maybe_get_writable_device_path, + fake_devpath, + info, + LOG, + ) + self.assertIsNone(devpath) + self.assertIn( + "WARNING: device '{0}' not a block device. cannot resize".format( + fake_devpath + ), + self.logs.getvalue(), + ) + + def test_maybe_get_writable_device_path_non_block_on_container(self): + """When device is non-block device in container, emit debug log.""" + fake_devpath = self.tmp_path("dev/readwrite") + util.write_file(fake_devpath, "", mode=0o600) # read-write + info = "dev=/dev/root mnt_point=/ path={0}".format(fake_devpath) + + devpath = wrap_and_call( + "cloudinit.config.cc_resizefs_vyos.util", + {"is_container": {"return_value": True}}, + maybe_get_writable_device_path, + fake_devpath, + info, + LOG, + ) + self.assertIsNone(devpath) + self.assertIn( + "DEBUG: device '{0}' not a block device in container." + " cannot resize".format(fake_devpath), + self.logs.getvalue(), + ) + + def test_maybe_get_writable_device_path_returns_cmdline_root(self): + """When root device is UUID in kernel commandline, update devpath.""" + # XXX Long-term we want to use FilesystemMocking test to avoid + # touching os.stat. + FakeStat = namedtuple( + "FakeStat", ["st_mode", "st_size", "st_mtime"] + ) # minimal def. + info = "dev=/dev/root mnt_point=/ path=/does/not/matter" + devpath = wrap_and_call( + "cloudinit.config.cc_resizefs_vyos", + { + "util.get_cmdline": {"return_value": "asdf root=UUID=my-uuid"}, + "util.is_container": False, + "os.path.exists": False, # /dev/root doesn't exist + "os.stat": { + "return_value": FakeStat(25008, 0, 1) + }, # char block device + }, + maybe_get_writable_device_path, + "/dev/root", + info, + LOG, + ) + self.assertEqual("/dev/disk/by-uuid/my-uuid", devpath) + self.assertIn( + "DEBUG: Converted /dev/root to '/dev/disk/by-uuid/my-uuid'" + " per kernel cmdline", + self.logs.getvalue(), + ) + + @mock.patch("cloudinit.util.mount_is_read_write") + @mock.patch("cloudinit.config.cc_resizefs_vyos.os.path.isdir") + def test_resize_btrfs_mount_is_ro(self, m_is_dir, m_is_rw): + """Do not resize / directly if it is read-only. (LP: #1734787).""" + m_is_rw.return_value = False + m_is_dir.return_value = True + self.assertEqual( + ("btrfs", "filesystem", "resize", "max", "//.snapshots"), + _resize_btrfs("/", "/dev/sda1"), + ) + + @mock.patch("cloudinit.util.mount_is_read_write") + @mock.patch("cloudinit.config.cc_resizefs_vyos.os.path.isdir") + def test_resize_btrfs_mount_is_rw(self, m_is_dir, m_is_rw): + """Do not resize / directly if it is read-only. (LP: #1734787).""" + m_is_rw.return_value = True + m_is_dir.return_value = True + self.assertEqual( + ("btrfs", "filesystem", "resize", "max", "/"), + _resize_btrfs("/", "/dev/sda1"), + ) + + @mock.patch("cloudinit.util.is_container", return_value=True) + @mock.patch("cloudinit.util.is_FreeBSD") + def test_maybe_get_writable_device_path_zfs_freebsd( + self, freebsd, m_is_container + ): + freebsd.return_value = True + info = "dev=gpt/system mnt_point=/ path=/" + devpth = maybe_get_writable_device_path("gpt/system", info, LOG) + self.assertEqual("gpt/system", devpth) + + +# vi: ts=4 expandtab |