# This file is part of cloud-init. See LICENSE file for license information.
from cloudinit.config.cc_resizefs import (
can_skip_resize, handle, maybe_get_writable_device_path, _resize_btrfs,
_resize_zfs, _resize_xfs, _resize_ext, _resize_ufs)
from collections import namedtuple
import logging
import textwrap
from cloudinit.tests.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.config.cc_resizefs._get_dumpfs_output')
@mock.patch('cloudinit.config.cc_resizefs._get_gpart_output')
def test_skip_ufs_resize(self, gpart_out, dumpfs_out):
fs_type = "ufs"
resize_what = "/"
devpth = "/dev/da0p2"
dumpfs_out.return_value = (
"# newfs command for / (/dev/label/rootfs)\n"
"newfs -O 2 -U -a 4 -b 32768 -d 32768 -e 4096 "
"-f 4096 -g 16384 -h 64 -i 8192 -j -k 6408 -m 8 "
"-o time -s 58719232 /dev/label/rootfs\n")
gpart_out.return_value = textwrap.dedent("""\
=> 40 62914480 da0 GPT (30G)
40 1024 1 freebsd-boot (512K)
1064 58719232 2 freebsd-ufs (28G)
58720296 3145728 3 freebsd-swap (1.5G)
61866024 1048496 - free - (512M)
""")
res = can_skip_resize(fs_type, resize_what, devpth)
self.assertTrue(res)
@mock.patch('cloudinit.config.cc_resizefs._get_dumpfs_output')
@mock.patch('cloudinit.config.cc_resizefs._get_gpart_output')
def test_skip_ufs_resize_roundup(self, gpart_out, dumpfs_out):
fs_type = "ufs"
resize_what = "/"
devpth = "/dev/da0p2"
dumpfs_out.return_value = (
"# newfs command for / (/dev/label/rootfs)\n"
"newfs -O 2 -U -a 4 -b 32768 -d 32768 -e 4096 "
"-f 4096 -g 16384 -h 64 -i 8192 -j -k 368 -m 8 "
"-o time -s 297080 /dev/label/rootfs\n")
gpart_out.return_value = textwrap.dedent("""\
=> 34 297086 da0 GPT (145M)
34 297086 1 freebsd-ufs (145M)
""")
res = can_skip_resize(fs_type, resize_what, devpth)
self.assertTrue(res)
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 = {'resize_rootfs': False}
handle('cc_resizefs', cfg, _cloud=None, log=LOG, args=[])
self.assertIn(
'DEBUG: Skipping module named cc_resizefs, resizing disabled\n',
self.logs.getvalue())
@skipUnlessJsonSchema()
def test_handle_schema_validation_logs_invalid_resize_rootfs_value(self):
"""The handle reports json schema violations as a warning.
Invalid values for resize_rootfs result in disabling the module.
"""
cfg = {'resize_rootfs': 'junk'}
handle('cc_resizefs', cfg, _cloud=None, log=LOG, args=[])
logs = self.logs.getvalue()
self.assertIn(
"WARNING: Invalid config:\nresize_rootfs: 'junk' is not one of"
" [True, False, 'noblock']",
logs)
self.assertIn(
'DEBUG: Skipping module named cc_resizefs, resizing disabled\n',
logs)
@mock.patch('cloudinit.config.cc_resizefs.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 = {'resize_rootfs': True}
handle('cc_resizefs', cfg, _cloud=None, log=LOG, args=[])
logs = self.logs.getvalue()
self.assertNotIn("WARNING: Invalid config:\nresize_rootfs:", 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 = {'resize_rootfs': True}
exists_mock_path = 'cloudinit.config.cc_resizefs.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.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', 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 = {'resize_rootfs': True}
with mock.patch('cloudinit.config.cc_resizefs.do_resize') as dresize:
handle('cc_resizefs', 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 = {'resize_rootfs': 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.do_resize') as dresize:
with mock.patch('cloudinit.config.cc_resizefs.os.stat') as m_stat:
m_stat.side_effect = fake_stat
handle('cc_resizefs', 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.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.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.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.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.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',
{'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.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.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',
{'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.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.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