The test cases for the blockdev-based export and for the file-based
export with growable=on work before commit a94a1d7699 ("fuse: Manually
process requests (without libfuse)"), then are broken until commit
"block/export/fuse: fix regression with O_TRUNC when export is
growable".The test case for the blockdev-based export requires passwordless sudo for losetup and chmod similar to test 108. Signed-off-by: Fiona Ebner <[email protected]> --- tests/qemu-iotests/tests/fuse-truncate | 171 +++++++++++++++++++++ tests/qemu-iotests/tests/fuse-truncate.out | 5 + 2 files changed, 176 insertions(+) create mode 100755 tests/qemu-iotests/tests/fuse-truncate create mode 100644 tests/qemu-iotests/tests/fuse-truncate.out diff --git a/tests/qemu-iotests/tests/fuse-truncate b/tests/qemu-iotests/tests/fuse-truncate new file mode 100755 index 0000000000..10bc1175c0 --- /dev/null +++ b/tests/qemu-iotests/tests/fuse-truncate @@ -0,0 +1,171 @@ +#!/usr/bin/env python3 +# group: rw +# +# Test how fuse exports behave with regard to O_TRUNC. +# +# Copyright (C) 2026 Proxmox Server Solutions GmbH +# +# SPDX-License-Identifier: GPL-2.0-or-later + +import os +import subprocess +from pathlib import Path + +import iotests +from iotests import qemu_img, QemuStorageDaemon + +fuse_mount_point = os.path.join(iotests.test_dir, 'export.fuse') +image_size = 1 * 1024 * 1024 +image = os.path.join(iotests.test_dir, 'image.' + iotests.imgfmt) + +def check_fuse_support(): + Path(fuse_mount_point).touch() + test_qsd = QemuStorageDaemon('--blockdev', 'null-co,node-name=node0', + qmp=True) + res = test_qsd.qmp('block-export-add', { + 'id': 'exp0', + 'type': 'fuse', + 'node-name': 'node0', + 'mountpoint': fuse_mount_point, + 'allow-other': 'off' + }) + test_qsd.stop() + os.remove(fuse_mount_point) + if 'error' in res: + assert (res['error']['desc'] == + "Parameter 'type' does not accept value 'fuse'") + iotests.notrun('No FUSE support') + +def check_sudo_support(): + try: + subprocess.run(['sudo', '-n', 'losetup', '--version'], + capture_output=True, check=True) + except subprocess.CalledProcessError: + return False + try: + subprocess.run(['sudo', '-n', 'chmod', '--version'], + capture_output=True, check=True) + except subprocess.CalledProcessError: + return False + return True + +check_fuse_support() + +class TestTruncateBase(iotests.QMPTestCase): + growable = False + supports_preconditions = True + + def evaluate_preconditions(self): + return + + def add_blockdev(self): + qemu_img('create', '-f', iotests.imgfmt, image, str(image_size)) + self.qsd.cmd('blockdev-add', { + 'node-name': 'node0', + 'driver': iotests.imgfmt, + 'file': { + 'driver': 'file', + 'filename': image + } + }) + + def cleanup_blockdev(self): + os.remove(image) + + def add_export(self): + self.qsd.cmd('block-export-add', { + 'id': 'exp0', + 'type': 'fuse', + 'node-name': 'node0', + 'mountpoint': fuse_mount_point, + 'growable': self.growable, + 'writable': True, + 'allow-other': 'off' + }) + + def stop_qsd(self): + if self.qsd: + self.qsd.cmd('block-export-del', {'id': 'exp0'}) + self.qsd.stop() + self.qsd = None + + def setUp(self): + self.evaluate_preconditions() + if not self.supports_preconditions: + return + Path(fuse_mount_point).touch() + self.qsd = QemuStorageDaemon(qmp=True) + self.add_blockdev() + self.add_export() + + def tearDown(self): + if not self.supports_preconditions: + return + self.stop_qsd() + self.cleanup_blockdev() + os.remove(fuse_mount_point) + +class TestTruncateFileGrowable(TestTruncateBase): + growable = True + + def test_o_trunc(self): + with open(fuse_mount_point, 'w+b') as file: + file.seek(0, os.SEEK_END) + self.assertEqual(file.tell(), 0) + file.write(b"test") + self.stop_qsd() + +class TestTruncateFileNotGrowable(TestTruncateBase): + growable = False + + def test_o_trunc(self): + with open(fuse_mount_point, 'w+b') as file: + file.seek(0, os.SEEK_END) + self.assertEqual(file.tell(), image_size) + file.seek(0, os.SEEK_SET) + file.write(b"test") + self.stop_qsd() + +class TestTruncateBlockdev(TestTruncateBase): + growable = False + + def evaluate_preconditions(self): + self.supports_preconditions = check_sudo_support() + + def add_blockdev(self): + qemu_img('create', '-f', iotests.imgfmt, image, str(image_size)) + res = subprocess.run(['sudo', '-n', 'losetup', '--show', '-f', image], + check=True, capture_output=True, text=True) + self.loopdev = res.stdout.strip() + subprocess.run(['sudo', '-n', 'chmod', 'go+rw', self.loopdev], + check=True, capture_output=True) + self.qsd.cmd('blockdev-add', { + 'node-name': 'node0', + 'driver': iotests.imgfmt, + 'file': { + 'driver': 'host_device', + 'filename': self.loopdev + } + }) + + def cleanup_blockdev(self): + subprocess.run(['sudo', '-n', 'losetup', '--detach', self.loopdev], + check=True, capture_output=True) + os.remove(image) + + def test_o_trunc(self): + if not self.supports_preconditions: + iotests.case_notrun('No passwordless sudo for losetup and chmod') + return + + with open(fuse_mount_point, 'w+b') as file: + file.seek(0, os.SEEK_END) + self.assertEqual(file.tell(), image_size) + file.seek(0, os.SEEK_SET) + file.write(b"test") + self.stop_qsd() + +if __name__ == '__main__': + iotests.main(supported_fmts=['raw'], + supported_protocols=['file'], + supported_platforms=['linux']) diff --git a/tests/qemu-iotests/tests/fuse-truncate.out b/tests/qemu-iotests/tests/fuse-truncate.out new file mode 100644 index 0000000000..8d7e996700 --- /dev/null +++ b/tests/qemu-iotests/tests/fuse-truncate.out @@ -0,0 +1,5 @@ +... +---------------------------------------------------------------------- +Ran 3 tests + +OK -- 2.47.3
