🏡 index : quine-global-labs/mirror-cockpit-project-cockpit.git

#!/usr/bin/python3 -cimport os, sys; os.execv(os.path.dirname(sys.argv[1]) + "/../common/pywrap", sys.argv)

#
# Copyright (C) 2015 Red Hat, Inc.
# SPDX-License-Identifier: LGPL-2.1-or-later


import storagelib
import testlib


@testlib.nondestructive
class TestStorageMounting(storagelib.StorageCase):

    def _navigate_root_subvolume(self):
        b = self.browser

        b.wait_visible(self.card("btrfs filesystem"))
        self.click_card_row("btrfs filesystem", name="top-level")
        b.wait_visible(self.card("btrfs subvolume"))

    def testMounting(self):
        self._testMounting()

    @testlib.skipImage('no btrfs support', 'rhel-*', 'centos-*', 'fedora-eln')
    def testMountingBtrfs(self):
        self._testMounting("btrfs", "btrfs subvolume")

    def _testMounting(self, fstype="ext4", filesystem="ext4 filesystem"):
        m = self.machine
        b = self.browser

        mount_point_foo = f"{self.mnt_dir}/foo"
        mount_point_bar = f"{self.mnt_dir}/bar"

        self.login_and_go("/storage")

        # Add a disk
        disk_size = 128
        disk = self.add_ram_disk(size=disk_size)
        self.click_card_row("Storage", name=disk)

        # Format it

        self.click_card_dropdown("Unformatted data", "Format")
        self.dialog_wait_open()
        self.dialog_set_val("type", fstype)
        self.dialog_set_val("name", "FILESYSTEM")
        self.dialog_set_val("mount_point", "")
        self.dialog_apply()
        self.dialog_wait_error("mount_point", "Mount point cannot be empty")
        self.dialog_set_val("mount_point", mount_point_foo)
        self.dialog_apply()
        self.dialog_wait_close()
        self.assert_in_configuration("/dev/sda", "fstab", "dir", mount_point_foo)
        if fstype == "btrfs":
            self._navigate_root_subvolume()
        else:
            b.wait_text(self.card_desc(filesystem, "Name"), "FILESYSTEM")

        b.wait_in_text(self.card_desc(filesystem, "Mount point"), mount_point_foo)

        # Keep the mount point busy
        sleep_pid = m.spawn(f"cd {mount_point_foo}; sleep infinity", "sleep")
        self.write_file("/etc/systemd/system/keep-mnt-busy.service",
                        f"""
[Unit]
Description=Test Service

[Service]
WorkingDirectory={mount_point_foo}
ExecStart=/usr/bin/sleep infinity
""")
        m.execute("systemctl start keep-mnt-busy")

        b.click(self.card_button(filesystem, "Unmount"))
        b.wait_in_text("#dialog", str(sleep_pid))
        b.wait_in_text("#dialog", "sleep infinity")
        b.wait_in_text("#dialog", "keep-mnt-busy")
        b.wait_in_text("#dialog", "Test Service")
        b.wait_in_text("#dialog", "/usr/bin/sleep infinity")
        b.wait_in_text("#dialog", "The listed processes and services will be forcefully stopped.")
        if fstype != "btrfs":
            b.assert_pixels("#dialog", "busy-unmount", mock={"td[data-label='PID']": "1234",
                                                             "td[data-label='Started']": "a little while ago"})
        self.confirm()
        b.wait_in_text(self.card_desc(filesystem, "Mount point"), "The filesystem is not mounted")

        m.execute("! systemctl --quiet is-active keep-mnt-busy")

        if fstype != "btrfs":  # tested in test-storage-btrfs
            b.click(self.card_desc_action(filesystem, "Name"))
            self.dialog({"name": "filesystem"})
            b.wait_text(self.card_desc(filesystem, "Name"), "filesystem")

        b.click(self.card_desc(filesystem, "Mount point") + " button")
        self.dialog(expect={"mount_point": mount_point_foo},
                    values={"mount_point": mount_point_bar})
        self.assert_in_configuration("/dev/sda", "fstab", "dir", mount_point_bar)
        b.wait_in_text(self.card_desc(filesystem, "Mount point"), mount_point_bar)

        b.click(self.card_button(filesystem, "Mount"))
        self.dialog_wait_open()
        self.dialog_wait_val("mount_point", mount_point_bar)
        self.dialog_set_val("mount_point", "")
        self.dialog_apply()
        self.dialog_wait_error("mount_point", "Mount point cannot be empty")
        self.dialog_set_val("mount_point", mount_point_bar)
        self.dialog_apply()
        self.dialog_wait_close()
        b.wait_not_in_text(self.card_desc(filesystem, "Mount point"), "The filesystem is not mounted")

        # Set the "Never unlock at boot option"
        b.click(self.card_desc(filesystem, "Mount point") + " button")
        self.dialog({"at_boot": "never"})
        self.assertIn("noauto", m.execute(f"findmnt -s -n -o OPTIONS {mount_point_bar}"))
        self.assertIn("x-cockpit-never-auto", m.execute(f"findmnt -s -n -o OPTIONS {mount_point_bar}"))

        # Go to overview page and check that the filesystem usage is
        # displayed correctly.

        if fstype == "btrfs":  # TODO: Cockpit/UDisks reports the wrong usage for btrfs subvolumes
            return

        def wait_ratio_in_range(sel, low, high):
            b.wait_js_func("""(function (sel, low, high) {
              var text = ph_text(sel);
              var match = text.match('([0-9.]+) / ([0-9]+)');
              if (!match)
                return false;
              var ratio = parseFloat(match[1]) / parseFloat(match[2]);
              return low <= ratio && ratio <= high;
            })""", sel, low, high)

        b.go("#/")
        b.wait_visible(self.card_row("Storage", location=mount_point_bar))
        bar_selector = self.card_row("Storage", location=mount_point_bar) + " td:nth-child(5)"
        wait_ratio_in_range(bar_selector, 0.0, 0.1)
        dd_count = disk_size // 2 + 10
        m.execute(f"dd if=/dev/zero of={mount_point_bar}/zero bs=1M count={dd_count} status=none")
        wait_ratio_in_range(bar_selector, 0.5, 1.0)
        m.execute(f"rm {mount_point_bar}/zero")
        wait_ratio_in_range(bar_selector, 0.0, 0.1)

        self.click_card_row("Storage", location=mount_point_bar)
        # On rhel-9 we don't use the 'vpd_use_hostno=0' option so S/N is different
        serial_num = "8000" if b.machine.image.startswith("rhel-9") else "2000"
        b.wait_in_text(self.card_desc("Solid State Drive", "Serial number"), serial_num)  # scsi_debug serial

        # Remove fstab entry

        b.click(self.card_desc(filesystem, "Mount point") + " button")
        self.dialog(expect={"mount_point": mount_point_bar},
                    values={"mount_point": ""})
        self.assertEqual(self.configuration_field("/dev/sda", "fstab", "dir"), "")
        b.wait_in_text(self.card_desc(filesystem, "Mount point"), "The filesystem has no permanent mount point.")

    @testlib.skipImage('no btrfs support', 'rhel-*', 'centos-*', 'fedora-eln')
    def testMountingHelpBtrfs(self):
        self._testMountingHelp("btrfs", "btrfs subvolume")

    def testMountingHelp(self):
        self._testMountingHelp()

    def _testMountingHelp(self, fstype="ext4", filesystem="ext4 filesystem"):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        # Add a disk
        disk = self.add_ram_disk(size=128)
        self.click_card_row("Storage", name=disk)

        # Format it

        self.click_card_dropdown("Unformatted data", "Format")
        self.dialog_wait_open()
        self.dialog_set_val("type", fstype)
        self.dialog_set_val("name", "FILESYSTEM")
        self.dialog_set_val("mount_point", f"{self.mnt_dir}/foo")
        self.dialog_apply()
        self.dialog_wait_close()
        if fstype == "btrfs":
            self._navigate_root_subvolume()
            b.wait_in_text(self.card_desc("btrfs subvolume", "Mount point"), f"{self.mnt_dir}/foo")
            self.addCleanupMount(f"{self.mnt_dir}/foo")
        else:
            b.wait_text(self.card_desc(filesystem, "Name"), "FILESYSTEM")
            b.wait_in_text(self.card_desc(filesystem, "Mount point"), f"{self.mnt_dir}/foo")

        # Unmount externally, remount with Cockpit

        m.execute(f"while mountpoint -q {self.mnt_dir}/foo && ! umount {self.mnt_dir}/foo; do sleep 0.2; done;")
        b.click(self.card_button(filesystem, "Mount now"))
        b.wait_not_present(self.card_button(filesystem, "Mount now"))
        b.wait_not_in_text(self.card_desc(filesystem, "Mount point"), "The filesystem is not mounted")

        # Unmount externally, adjust fstab with Cockpit

        m.execute(f"while mountpoint -q {self.mnt_dir}/foo && ! umount {self.mnt_dir}/foo; do sleep 0.2; done;")
        b.click(self.card_button(filesystem, "Do not mount automatically on boot"))
        b.wait_not_present(self.card_button(filesystem, "Do not mount automatically on boot"))

        # Mount somewhere else externally while "noauto", unmount with Cockpit

        m.execute(f"mkdir -p {self.mnt_dir}/bar; mount {disk} {self.mnt_dir}/bar")
        b.click(self.card_button(filesystem, "Unmount now"))
        b.wait_not_present(self.card_button(filesystem, "Unmount now"))

        # Mount externally, unmount with Cockpit

        m.execute(f"mount {self.mnt_dir}/foo")
        b.click(self.card_button(filesystem, "Unmount now"))
        b.wait_not_present(self.card_button(filesystem, "Unmount now"))

        # Mount externally, adjust fstab with Cockpit

        m.execute(f"mount {self.mnt_dir}/foo")
        b.click(self.card_button(filesystem, "Mount also automatically on boot"))
        b.wait_not_present(self.card_button(filesystem, "Mount also automatically on boot"))

        # Move mount point externally, move back with Cockpit

        m.execute(f"while mountpoint -q {self.mnt_dir}/foo && ! umount {self.mnt_dir}/foo; do sleep 0.2; done;")
        m.execute(f"mkdir -p {self.mnt_dir}/bar; mount {disk} {self.mnt_dir}/bar")
        b.click(self.card_button(filesystem, f"Mount on {self.mnt_dir}/foo now"))
        b.wait_not_present(self.card_button(filesystem, f"Mount on {self.mnt_dir}/foo now"))

        # Move mount point externally, adjust fstab with Cockpit

        m.execute(f"while mountpoint -q {self.mnt_dir}/foo && ! umount {self.mnt_dir}/foo; do sleep 0.2; done;")
        m.execute(f"mkdir -p {self.mnt_dir}/bar; mount {disk} {self.mnt_dir}/bar")
        b.click(self.card_button(filesystem, f"Mount automatically on {self.mnt_dir}/bar on boot"))
        b.wait_not_present(self.card_button(filesystem, f"Mount automatically on {self.mnt_dir}/bar on boot"))

        # Using noauto,x-systemd.automount should not show a warning
        m.execute("sed -i -e 's/auto nofail/auto nofail,noauto/' /etc/fstab")
        b.wait_visible(self.card_button(filesystem, "Mount also automatically on boot"))
        m.execute("sed -i -e 's/noauto/noauto,x-systemd.automount/' /etc/fstab")
        b.wait_not_present(self.card_button(filesystem, "Mount also automatically on boot"))

        # Without fstab entry, mount and try to unmount
        m.execute(f"sed -i '{self.mnt_dir.replace('/', '\\/')}\\/bar/d' /etc/fstab")
        b.wait_visible(self.card_button(filesystem, f"Mount automatically on {self.mnt_dir}/bar on boot"))
        b.click(self.card_button(filesystem, "Unmount now"))
        b.wait_not_present(self.card_button(filesystem, f"Mount automatically on {self.mnt_dir}/bar on boot"))

    def testFstabOptions(self):
        self._testFstabOptions()

    @testlib.skipImage('no btrfs support', 'rhel-*', 'centos-*', 'fedora-eln')
    def testFsabOptionsBtrfs(self):
        self._testFstabOptions("btrfs", "btrfs subvolume")

    def _testFstabOptions(self, filesystem="ext4", desc="ext4 filesystem"):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        disk = self.add_ram_disk(size=128)
        self.click_card_row("Storage", name=disk)

        m.execute(f"mkfs.{filesystem} -L testfs {disk}")
        if filesystem == "btrfs":
            self._navigate_root_subvolume()
        else:
            b.wait_visible(self.card(desc))

        m.execute("! grep /run/data /etc/fstab")
        b.click(self.card_button(desc, "Mount"))
        self.dialog({"mount_point": "/run/data",
                     "mount_options.extra": "x-foo"})
        m.execute("grep /run/data /etc/fstab")
        m.execute("grep 'x-foo' /etc/fstab")

        b.wait_in_text(self.card_desc(desc, "Mount point"), "/run/data (ignore failure, x-foo)")

        # absent mntopts and fsck columns implies "defaults"
        if filesystem == "btrfs":
            m.execute(r"sed -i '/run\/data/ s/auto.*$/auto subvol=\//' /etc/fstab")
        else:
            m.execute(r"sed -i '/run\/data/ s/auto.*$/auto/' /etc/fstab")
        b.wait_in_text(self.card_desc(desc, "Mount point"), "/run/data (stop boot on failure)")

    @testlib.skipImage("FIXME: ought to work, investigate", "rhel-8-10")
    def testBadOption(self):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        disk = self.add_ram_disk()
        self.click_card_row("Storage", name=disk)

        self.click_card_dropdown("Unformatted data", "Format")
        self.dialog({"type": "ext4",
                     "mount_point": f"{self.mnt_dir}/foo"},
                    secondary=True)
        b.wait_in_text(self.card_desc("ext4 filesystem", "Mount point"), "The filesystem is not mounted")
        self.assertIn("noauto", m.execute(f"findmnt --fstab -n -o OPTIONS {self.mnt_dir}/foo"))

        b.click(self.card_button("ext4 filesystem", "Mount"))
        self.dialog_wait_open()
        self.dialog_set_val("mount_options.extra", "hurr")
        self.dialog_apply()
        self.dialog_wait_alert("Unknown parameter 'hurr'", "bad option")
        self.dialog_cancel()
        self.dialog_wait_close()

        # No changes should have been done to fstab, and the
        # filesystem should not be mounted.
        self.assertIn("noauto", m.execute(f"findmnt --fstab -n -o OPTIONS {self.mnt_dir}/foo"))
        self.assertNotIn("hurr", m.execute(f"findmnt --fstab -n -o OPTIONS {self.mnt_dir}/foo"))
        b.wait_in_text(self.card_desc("ext4 filesystem", "Mount point"), "The filesystem is not mounted")

        # Mount
        b.click(self.card_button("ext4 filesystem", "Mount"))
        self.dialog({})

        # Apply the dialog without changes and verify that the
        # filesystem is still mounted afterwards.  Cockpit used to
        # have a bug where this would accidentally unmount the
        # filesystem.

        b.click(self.card_desc("ext4 filesystem", "Mount point") + " button")
        self.dialog({})
        b.wait_not_in_text(self.card_desc("ext4 filesystem", "Mount point"), "The filesystem is not mounted")

        # Try to set a bad option while the filesystem is mounted.
        b.click(self.card_desc("ext4 filesystem", "Mount point") + " button")
        self.dialog_wait_open()
        self.dialog_set_val("mount_options.extra", "hurr")
        self.dialog_apply()
        self.dialog_wait_alert("Unknown parameter 'hurr'", "bad option")
        self.dialog_cancel()
        self.dialog_wait_close()

        self.assertNotIn("hurr", m.execute(f"findmnt --fstab -n -o OPTIONS {self.mnt_dir}/foo"))
        b.wait_not_in_text(self.card_desc("ext4 filesystem", "Mount point"), "The filesystem is not mounted")

    def testAtBoot(self):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        disk = self.add_ram_disk()
        self.click_card_row("Storage", name=disk)

        self.click_card_dropdown("Unformatted data", "Format")
        self.dialog({"type": "ext4",
                     "mount_point": f"{self.mnt_dir}/foo",
                     "at_boot": "local",
                     "crypto": "luks1",
                     "passphrase": "foobarfoo",
                     "passphrase2": "foobarfoo"},
                    secondary=True)
        b.wait_in_text(self.card_desc("Filesystem", "Mount point"), "The filesystem is not mounted")
        self.assertIn("noauto", m.execute(f"findmnt --fstab -n -o OPTIONS {self.mnt_dir}/foo"))
        self.assert_in_configuration(disk, "crypttab", "options", "noauto")

        def mount(expected_at_boot, at_boot):
            b.click(self.card_button("Filesystem", "Mount"))
            self.dialog_wait_open()
            self.dialog_wait_val("at_boot", expected_at_boot)
            self.dialog_set_val("at_boot", at_boot)
            self.dialog_set_val("passphrase", "foobarfoo")
            self.dialog_apply()
            self.dialog_wait_close()

        def unmount():
            b.click(self.card_button("ext4 filesystem", "Unmount"))
            self.dialog_wait_open()
            self.dialog_apply()
            self.dialog_wait_close()
            b.wait_in_text(self.card_desc("Filesystem", "Mount point"), "The filesystem is not mounted")

        mount("local", "nofail")
        b.wait_in_text(self.card_desc("ext4 filesystem", "Mount point"), "ignore failure")
        self.assertIn("nofail", m.execute(f"findmnt --fstab -n -o OPTIONS {self.mnt_dir}/foo"))
        self.assert_in_configuration(disk, "crypttab", "options", "nofail")
        unmount()

        mount("nofail", "netdev")
        b.wait_in_text(self.card_desc("ext4 filesystem", "Mount point"), "after network")
        self.assertIn("_netdev", m.execute(f"findmnt --fstab -n -o OPTIONS {self.mnt_dir}/foo"))
        self.assert_in_configuration(disk, "crypttab", "options", "_netdev")
        unmount()

        mount("netdev", "never")
        b.wait_in_text(self.card_desc("ext4 filesystem", "Mount point"), "never mount")
        self.assertIn("x-cockpit-never-auto", m.execute(f"findmnt --fstab -n -o OPTIONS {self.mnt_dir}/foo"))
        self.assert_in_configuration(disk, "crypttab", "options", "noauto")
        unmount()


@testlib.nondestructive
@testlib.skipImage("cryptsetup uses too much memory, OOM on our test VMs", "rhel-8-*")
class TestStorageMountingLUKS(storagelib.StorageCase):
    def testEncryptedMountingHelp(self):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        # Add a disk
        disk = self.add_ram_disk()
        self.click_card_row("Storage", name=disk)

        # Format it with encryption

        self.click_card_dropdown("Unformatted data", "Format")
        self.dialog_wait_open()
        self.dialog_set_val("type", "ext4")
        self.dialog_set_val("name", "FILESYSTEM")
        self.dialog_set_val("crypto", self.default_crypto_type)
        self.dialog_set_val("crypto_options", "xxx")
        self.dialog_set_val("passphrase", "vainu-reku-toma-rolle-kaja")
        self.dialog_set_val("passphrase2", "vainu-reku-toma-rolle-kaja")
        self.dialog_set_val("mount_point", f"{self.mnt_dir}/foo")
        self.dialog_set_val("at_boot", "netdev")
        self.dialog_apply()
        self.dialog_wait_close()
        b.wait_text(self.card_desc("ext4 filesystem", "Name"), "FILESYSTEM")
        b.wait_in_text(self.card_desc("ext4 filesystem", "Mount point"), f"{self.mnt_dir}/foo")

        # Unmount and lock externally, unlock and remount with Cockpit

        m.execute(f"while mountpoint -q {self.mnt_dir}/foo && ! umount {self.mnt_dir}/foo; do sleep 0.2; done;")
        m.execute(f"udisksctl lock -b {disk}")
        # wait until the UI updated to the locking
        b.wait_text(self.card_desc("Encryption", "Cleartext device"), "-")
        b.click(self.card_button("Filesystem", "Mount now"))
        self.dialog({"passphrase": "vainu-reku-toma-rolle-kaja"})
        b.wait_not_in_text(self.card_desc("ext4 filesystem", "Mount point"),
                           "The filesystem is not mounted")
        b.wait_not_present(self.card_button("ext4 filesystem", "Mount now"))

        # Unmount and lock externally, adjust fstab with Cockpit

        m.execute(f"while mountpoint -q {self.mnt_dir}/foo && ! umount {self.mnt_dir}/foo; do sleep 0.2; done;")
        m.execute(f"udisksctl lock -b {disk}")
        # wait until the UI updated to the locking
        b.wait_text(self.card_desc("Encryption", "Cleartext device"), "-")
        b.click(self.card_button("Filesystem", "Do not mount automatically on boot"))
        b.wait_not_present(self.card_button("Filesystem", "Do not mount automatically on boot"))

        # Unlock and mount externally, unmount and lock with Cockpit

        m.execute(f"echo -n vainu-reku-toma-rolle-kaja | udisksctl unlock --key-file /dev/stdin -b {disk}")
        m.execute(f"mount {self.mnt_dir}/foo")
        b.click(self.card_button("ext4 filesystem", "Unmount now"))
        b.wait_visible(self.card("Filesystem"))
        b.wait_not_present(self.card_button("Filesystem", "Unmount now"))

        # Unlock and mount externally, adjust fstab with Cockpit

        m.execute(f"echo -n vainu-reku-toma-rolle-kaja | udisksctl unlock --key-file /dev/stdin -b {disk}")
        m.execute(f"mount {self.mnt_dir}/foo")
        b.click(self.card_button("ext4 filesystem", "Mount also automatically on boot"))
        b.wait_not_present(self.card_button("ext4 filesystem", "Mount also automatically on boot"))
        b.wait_visible(self.card("ext4 filesystem"))

        # Add noauto to crypttab (but not fstab), remove with Cockpit

        m.execute("sed -i -e 's/xxx/xxx,noauto/' /etc/crypttab")
        b.click(self.card_button("ext4 filesystem", "Unlock automatically on boot"))
        b.wait_not_present(self.card_button("ext4 filesystem", "Unlock automatically on boot"))
        b.wait_visible(self.card("ext4 filesystem"))

        # Add noauto to crypttab (but not fstab), add also to fstab with Cockpit

        m.execute("sed -i -e 's/xxx/xxx,noauto/' /etc/crypttab")
        b.click(self.card_button("ext4 filesystem", "Do not mount automatically on boot"))
        b.wait_not_present(self.card_button("ext4 filesystem", "Do not mount automatically on boot"))
        b.wait_visible(self.card("ext4 filesystem"))

    def testDuplicateMountPoints(self):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        self.addCleanupVG("test")
        self.addCleanup(m.execute, "cryptsetup close $(lsblk -lno NAME /dev/test/one | tail -1) || true")
        self.addCleanupMount("/run/data")

        # Quickly make two logical volumes
        disk = self.add_ram_disk()
        b.wait_visible(self.card_row("Storage", name=disk))
        m.execute(f"vgcreate test {disk}; lvcreate test -n one -L 20M; lvcreate test -n two -L 20M")
        self.click_card_row("Storage", name="test")
        b.wait_text(self.card_row_col("LVM2 logical volumes", 1, 1), "one")
        b.wait_text(self.card_row_col("LVM2 logical volumes", 2, 1), "two")

        # Encrypt and format the first and give it /run/data as the mount point
        self.click_card_row("LVM2 logical volumes", 1)
        self.click_card_dropdown("Unformatted data", "Format")
        self.dialog({"type": "ext4",
                     "crypto": self.default_crypto_type,
                     "passphrase": "vainu-reku-toma-rolle-kaja",
                     "passphrase2": "vainu-reku-toma-rolle-kaja",
                     "mount_point": "/run/data"})
        b.click(self.card_parent_link())
        b.wait_text(self.card_row_col("LVM2 logical volumes", 1, 2), "ext4 filesystem (encrypted)")
        b.wait_text(self.card_row_col("LVM2 logical volumes", 1, 3), "/run/data")

        # Format the second and also try to use /run/data as the mount point
        self.click_card_row("LVM2 logical volumes", 2)
        self.click_card_dropdown("Unformatted data", "Format")
        self.dialog_wait_open()
        self.dialog_set_val("type", "ext4")
        self.dialog_set_val("mount_point", "/run/data")
        self.dialog_apply()
        self.dialog_wait_error("mount_point", "Mount point is already used for /dev/test/one")
        self.dialog_set_val("mount_point", "/run/data")  # to clear the error
        self.dialog_apply_secondary()
        self.dialog_wait_error("mount_point", "Mount point is already used for /dev/test/one")
        self.dialog_cancel()
        self.dialog_wait_close()
        b.click(self.card_parent_link())

        # Format the first and re-use /run/data as the mount point.
        # This should be allowed.
        self.click_card_row("LVM2 logical volumes", 1)
        self.click_card_dropdown("ext4 filesystem", "Format")
        self.dialog({"type": "ext4",
                     "mount_point": "/run/data"})
        b.click(self.card_parent_link())
        b.wait_text(self.card_row_col("LVM2 logical volumes", 1, 2), "ext4 filesystem (encrypted)")
        b.wait_text(self.card_row_col("LVM2 logical volumes", 1, 3), "/run/data")

    def testNeverAuto(self):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        # Add a disk and format it with luks and a filesystem, but with "Never unlock at boot"
        disk = self.add_ram_disk()
        self.click_card_row("Storage", name=disk)

        self.click_card_dropdown("Unformatted data", "Format")
        self.dialog({"type": "ext4",
                     "crypto": self.default_crypto_type,
                     "passphrase": "vainu-reku-toma-rolle-kaja",
                     "passphrase2": "vainu-reku-toma-rolle-kaja",
                     "at_boot": "never",
                     "mount_point": f"{self.mnt_dir}/foo"})
        b.wait_in_text(self.card_desc("ext4 filesystem", "Mount point"), "never mount at boot")

        # The filesystem should be mounted but have the "noauto"
        # option in both fstab and crypttab
        self.wait_mounted("ext4 filesystem")
        self.assertIn("noauto", m.execute(f"grep {self.mnt_dir}/foo /etc/fstab || true"))
        self.assertNotEqual(m.execute("grep noauto /etc/crypttab"), "")

        # Unmounting should keep the noauto option, as always
        b.click(self.card_button("ext4 filesystem", "Unmount"))
        self.confirm()
        self.wait_not_mounted("Filesystem")
        self.assertIn("noauto", m.execute(f"grep {self.mnt_dir}/foo /etc/fstab || true"))
        self.assertNotEqual(m.execute("grep noauto /etc/crypttab"), "")

        # Mounting should also keep the "noauto", but it should not show up in the extra options
        b.click(self.card_button("Filesystem", "Mount"))
        self.dialog_check({"mount_options.extra": False})
        self.dialog_set_val("passphrase", "vainu-reku-toma-rolle-kaja")
        self.dialog_apply()
        self.dialog_wait_close()
        self.wait_mounted("ext4 filesystem")
        self.assertIn("noauto", m.execute(f"grep {self.mnt_dir}/foo /etc/fstab || true"))
        self.assertNotEqual(m.execute("grep noauto /etc/crypttab"), "")

        # As should updating the mount information
        b.click(self.card_desc("ext4 filesystem", "Mount point") + " button")
        self.dialog_check({"mount_options.extra": False})
        self.dialog_apply()
        self.dialog_wait_close()
        self.assertIn("noauto", m.execute(f"grep {self.mnt_dir}/foo /etc/fstab || true"))
        self.assertNotEqual(m.execute("grep noauto /etc/crypttab"), "")

        # Removing "noauto" from fstab but not from crypttab externally should show a warning
        m.execute("sed -i -e 's/noauto//' /etc/fstab")
        b.wait_in_text(self.card("ext4 filesystem"),
                       "The filesystem is configured to be automatically mounted on boot but its encryption container will not be unlocked at that time.")
        b.click(self.card_button("ext4 filesystem", "Do not mount automatically"))
        b.wait_not_present(self.card_button("ext4 filesystem", "Do not mount automatically"))

    @testlib.skipImage("cryptsetup uses too much memory, OOM on our test VMs", "fedora-45")
    def testOverMounting(self):
        m = self.machine
        b = self.browser

        # Disable pre-loading packagekit, otherwise udisks oom's.
        self.disable_preload("packagekit")

        self.login_and_go("/storage")

        # Add a disk and make two partitions on it, one on self.mnt_dir/foo
        # and one on self.mnt_dir/foo/bar

        disk = self.add_ram_disk(100)
        self.click_card_row("Storage", name=disk)

        self.click_card_dropdown("Solid State Drive", "Create partition table")
        self.dialog({"type": "gpt"})
        b.wait_text(self.card_row_col("GPT partitions", 1, 1), "Free space")

        self.click_dropdown(self.card_row("GPT partitions", 1), "Create partition")
        self.dialog({"type": "ext4",
                     "size": 50,
                     "crypto": self.default_crypto_type,
                     "passphrase": "vainu-reku-toma-rolle-kaja",
                     "passphrase2": "vainu-reku-toma-rolle-kaja",
                     "mount_point": f"{self.mnt_dir}/foo"},
                    secondary=True)
        b.wait_text(self.card_row_col("GPT partitions", 1, 3), f"{self.mnt_dir}/foo (not mounted)")

        self.click_dropdown(self.card_row("GPT partitions", 2), "Create partition")
        self.dialog({"type": "ext4",
                     "crypto": self.default_crypto_type,
                     "passphrase": "vainu-reku-toma-rolle-kaja",
                     "passphrase2": "vainu-reku-toma-rolle-kaja",
                     "mount_point": f"{self.mnt_dir}/foo/bar"},
                    secondary=True)
        b.wait_text(self.card_row_col("GPT partitions", 2, 3), f"{self.mnt_dir}/foo/bar (not mounted)")

        # Mount self.mnt_dir/foo/bar first and check that mounting self.mnt_dir/foo is
        # rejected

        self.click_dropdown(self.card_row("GPT partitions", 2), "Mount")
        self.dialog({"passphrase": "vainu-reku-toma-rolle-kaja"})
        b.wait_text(self.card_row_col("GPT partitions", 2, 3), f"{self.mnt_dir}/foo/bar")

        self.click_dropdown(self.card_row("GPT partitions", 1), "Mount")
        self.dialog_wait_open()
        self.dialog_set_val("passphrase", "vainu-reku-toma-rolle-kaja")
        self.dialog_apply()
        self.dialog_wait_error("mount_point", "Filesystems are already mounted below this mountpoint.")
        b.assert_pixels("#dialog", "overmounting-rejection")
        self.dialog_cancel()
        self.dialog_wait_close()

        # Unmount self.mnt_dir/foo/bar, mount self.mnt_dir/foo, mount self.mnt_dir/foo/bar
        # again and then check that unmounting self.mnt_dir/foo will also
        # unmount self.mnt_dir/foo/bar.

        self.click_dropdown(self.card_row("GPT partitions", 2), "Unmount")
        self.confirm()

        self.click_dropdown(self.card_row("GPT partitions", 1), "Mount")
        self.dialog({"passphrase": "vainu-reku-toma-rolle-kaja"})
        b.wait_text(self.card_row_col("GPT partitions", 1, 3), f"{self.mnt_dir}/foo")

        self.click_dropdown(self.card_row("GPT partitions", 2), "Mount")
        self.dialog({"passphrase": "vainu-reku-toma-rolle-kaja"})
        b.wait_text(self.card_row_col("GPT partitions", 2, 3), f"{self.mnt_dir}/foo/bar")

        self.click_dropdown(self.card_row("GPT partitions", 1), "Unmount")
        self.dialog_wait_open()
        b.wait_in_text(f"#dialog tr:contains('{self.mnt_dir}/foo/bar')", "unmount")
        self.dialog_apply()
        self.dialog_wait_close()

        # Now self.mnt_dir/foo/bar should be noauto.
        self.assert_in_configuration(disk + "2", "crypttab", "options", "noauto")
        self.assertIn("noauto", m.execute(f"findmnt --fstab -n -o OPTIONS {self.mnt_dir}/foo/bar"))

        # Mount them again and check that initializing the disk will
        # unmount self.mnt_dir/foo/bar first.

        self.click_dropdown(self.card_row("GPT partitions", 1), "Mount")
        self.dialog({"passphrase": "vainu-reku-toma-rolle-kaja"})
        b.wait_text(self.card_row_col("GPT partitions", 1, 3), f"{self.mnt_dir}/foo")

        self.click_dropdown(self.card_row("GPT partitions", 2), "Mount")
        self.dialog({"passphrase": "vainu-reku-toma-rolle-kaja"})
        b.wait_text(self.card_row_col("GPT partitions", 2, 3), f"{self.mnt_dir}/foo/bar")

        # Sometimes a block device is still held open by
        # something immediately after locking it. This
        # prevents the kernel from reading the new partition
        # table. Let's just retry in that case.

        def first_setup():
            b.wait_text("#dialog tbody:nth-of-type(1) td[data-label=Location]", f"{self.mnt_dir}/foo/bar")
            b.wait_text("#dialog tbody:nth-of-type(2) td[data-label=Location]", f"{self.mnt_dir}/foo")

        self.dialog_with_error_retry(trigger=lambda: self.click_card_dropdown("Solid State Drive",
                                                                              "Create partition table"),
                                     first_setup=first_setup,
                                     errors=["Timed out waiting for object"])

        b.wait_text(self.card_row_col("GPT partitions", 1, 1), "Free space")


if __name__ == '__main__':
    testlib.test_main()