[Stable-11.0.4 103/120] iotests: test O_TRUNC behavior for fuse exports

From: Fiona Ebner <[email protected]>

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]>
Message-ID: <[email protected]>
[kwolf: Catch OSError when probing sudo support]
Reviewed-by: Kevin Wolf <[email protected]>
Signed-off-by: Kevin Wolf <[email protected]>
(cherry picked from commit 7d06bbb2d9ba77e45b5a3131c9e7f934df99d5bc)
Signed-off-by: Michael Tokarev <[email protected]>

diff --git a/tests/qemu-iotests/tests/fuse-truncate b/tests/qemu-iotests/tests/fuse-truncate
new file mode 100755
index 00000000000..7eee35111c8
--- /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 (OSError, subprocess.CalledProcessError):
+        return False
+    try:
+        subprocess.run(['sudo', '-n', 'chmod', '--version'],
+                       capture_output=True, check=True)
+    except (OSError, 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 00000000000..8d7e9967009
--- /dev/null
+++ b/tests/qemu-iotests/tests/fuse-truncate.out
@@ -0,0 +1,5 @@
+...
+----------------------------------------------------------------------
+Ran 3 tests
+
+OK
-- 
2.47.3