path: root/tests/unittests/test_handler/test_handler_resizefs.py

# 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