diff --git a/qubes/tests/storage_file.py b/qubes/tests/storage_file.py index a2a461b6..9742ffa1 100644 --- a/qubes/tests/storage_file.py +++ b/qubes/tests/storage_file.py @@ -25,7 +25,10 @@ import shutil import asyncio import unittest.mock +import subprocess + import qubes.storage +import qubes.utils import qubes.tests.storage from qubes.config import defaults @@ -383,6 +386,73 @@ class TC_01_FileVolumes(qubes.tests.QubesTestCase): volume_data = volume_file.read().strip('\0') self.assertNotEqual(volume_data, 'test data') + def test_023_resize(self): + config = { + 'name': 'root', + 'pool': self.POOL_NAME, + 'rw': True, + 'save_on_stop': True, + 'size': 32 * 1024**2, + } + vm = qubes.tests.storage.TestVM(self) + volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config) + qubes.utils.void_coros_maybe([volume.create()]) + new_size = 64 * 1024 ** 2 + qubes.utils.void_coros_maybe([volume.resize(new_size)]) + self.assertEqual(os.path.getsize(volume.path), new_size) + self.assertEqual(volume.size, new_size) + + def _get_loop_size(self, path): + sudo = [] if os.getuid() == 0 else ['sudo'] + try: + loop_name = subprocess.check_output( + sudo + ['losetup', '--associated', path]).decode().split(':')[0] + if os.getuid() != 0: + return int( + subprocess.check_output( + ['sudo', 'blockdev', '--getsize64', loop_name])) + fd = os.open(loop_name, os.O_RDONLY) + try: + return os.lseek(fd, 0, os.SEEK_END) + finally: + os.close(fd) + except subprocess.CalledProcessError: + return None + + def _setup_loop(self, path): + sudo = [] if os.getuid() == 0 else ['sudo'] + loop_name = subprocess.check_output( + sudo + ['losetup', '--show', '--find', path]).decode().strip() + self.addCleanup(subprocess.call, sudo + ['losetup', '-d', loop_name]) + + def test_007_resize_running(self): + old_size = 32 * 1024**2 + config = { + 'name': 'root', + 'pool': self.POOL_NAME, + 'rw': True, + 'save_on_stop': True, + 'size': old_size, + } + vm = qubes.tests.storage.TestVM(self) + volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config) + qubes.utils.void_coros_maybe([volume.create()]) + qubes.utils.void_coros_maybe([volume.start()]) + self._setup_loop(volume.path) + new_size = 64 * 1024 ** 2 + orig_check_call = subprocess.check_call + with unittest.mock.patch('subprocess.check_call') as mock_subprocess: + sudo = [] if os.getuid() == 0 else ['sudo'] + mock_subprocess.side_effect = (lambda *args, **kwargs: + orig_check_call(sudo + args[0], *args[1:], **kwargs)) + qubes.utils.void_coros_maybe([volume.resize(new_size)]) + self.assertEqual(os.path.getsize(volume.path), new_size) + self.assertEqual(self._get_loop_size(volume.path), new_size) + self.assertEqual(volume.size, new_size) + qubes.utils.void_coros_maybe([volume.stop()]) + self.assertEqual(os.path.getsize(volume.path), new_size) + self.assertEqual(volume.size, new_size) + def assertVolumePath(self, vm, dev_name, expected, rw=True): # :pylint: disable=invalid-name volumes = vm.volumes