🏡 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) 2021 Red Hat, Inc.
# SPDX-License-Identifier: LGPL-2.1-or-later


import subprocess

import packagelib
import storagelib
import testlib

PV_SIZE = 4000  # 4 GB in MB

# List of images that support Stratis and create V1 pools by
# default. When this gets empty, Cockpit itself can be cleaned up to
# drop support for Stratis API revisions before "r8".
#
V1_POOL_IMAGES = ["rhel-9-10", "rhel-9-9", "rhel-9-8", "rhel-9-7", "centos-9-bootc", "centos-9", "fedora-eln"]


def get_stratis_stop_type_opt(execute):
    """Get `stratis stop pool` required option for a pool name

    The CLI changed in an incompatible way in Fedora 40, it needs an extra --name option
    which cannot be provided in earlier versions.
    """
    try:
        if '--name' in execute("stratis pool stop --help"):
            return "--name"
    except subprocess.CalledProcessError:
        # on RHEL 8 this fails with "error: invalid choice"
        pass
    return ""


def create_pool_key(machine, keyname, passphrase):
    # this is a bit complicated, see https://bugzilla.redhat.com/show_bug.cgi?id=2246923
    machine.execute(f"echo -n '{passphrase}' | stratis key set --keyfile-path /dev/stdin {keyname}")


def create_legacy_pool(machine, disks, name="pool0", keydesc=None, tang=None):
    keydesc_opt = "--key-desc " + keydesc if keydesc else ""
    clevis_opt = "--clevis tang --trust-url --tang-url " + tang if tang else ""
    machine.execute(f"yes | stratisd-tools stratis-legacy-pool {keydesc_opt} {clevis_opt} {name} {' '.join(disks)}")
    testlib.wait(lambda: name in machine.execute("stratis pool list"))


@testlib.skipImage("No Stratis", "debian-*", "ubuntu-*", "arch")
@testlib.skipImage("stratis not installed", "*-bootc")
@testlib.skipImage("commit 817c957899a4 removed Stratis 2 support", "rhel-8-*")
@testlib.nondestructive
class TestStorageStratis(storagelib.StorageCase):
    def setUp(self):
        super().setUp()
        exe = self.machine.execute

        exe("systemctl start stratisd")
        self.addCleanup(exe, "systemctl stop stratisd")

        self.stop_type_opt = get_stratis_stop_type_opt(exe)

    def tearDown(self):
        # we need to clean up stratis *first*, before add_loopback_disk()'s addCleanup() handlers
        # tearDown() runs before addCleanup()
        exe = self.machine.execute
        exe("mount | grep mapper/stratis | awk '{print $1}' | xargs --no-run-if-empty umount")
        exe(f"stratis report | jq -r '.pools[] | .name' | xargs -n1 --no-run-if-empty stratis pool stop {self.stop_type_opt}")
        exe("stratis report | jq -r '.pools[] | .name' | xargs -n1 --no-run-if-empty stratis pool destroy")
        exe("dmsetup ls | cut -f1 | grep ^stratis | xargs -n1 --no-run-if-empty dmsetup remove --force")

        super().tearDown()

    def testBasic(self, legacy=False):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        # use fixed names, to avoid grabbing loop1 and loop12 (which losetup sometimes likes to do)
        # as that clashes with :contains(loop1) below; also, fix names for pixel test
        dev_1 = self.add_loopback_disk(PV_SIZE, name="loop10")
        dev_2 = self.add_loopback_disk(PV_SIZE, name="loop11")
        dev_3 = self.add_loopback_disk(PV_SIZE, name="loop12")
        dev_4 = self.add_loopback_disk(PV_SIZE, name="loop13")
        dev_5 = self.add_loopback_disk(PV_SIZE, name="loop14")
        b.wait_visible(self.card_row("Storage", name=dev_1))
        b.wait_visible(self.card_row("Storage", name=dev_2))
        b.wait_visible(self.card_row("Storage", name=dev_3))
        b.wait_visible(self.card_row("Storage", name=dev_4))
        b.wait_visible(self.card_row("Storage", name=dev_5))

        # Create a pool
        if not legacy:
            self.dialog_open_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                                        expect=lambda: (self.dialog_is_present('disks', dev_1) and
                                                        self.dialog_is_present('disks', dev_2) and
                                                        self.dialog_check({"name": "pool0"})))
            self.dialog_set_val("disks", {dev_1: True, dev_2: True})
            b.assert_pixels("#dialog", "create-pool")
            self.dialog_apply()
            self.dialog_wait_close()
        else:
            create_legacy_pool(m, disks=[dev_1, dev_2])

        b.wait_visible(self.card_row("Storage", name="pool0"))
        b.wait_not_present(self.card_row("Storage", name="pool0") + " .ct-icon-exclamation-triangle")

        # Check that the next name is "pool1"
        self.click_devices_dropdown("Create Stratis pool")
        self.dialog_wait_open()
        self.dialog_wait_val("name", "pool1")
        self.dialog_cancel()
        self.dialog_wait_close()

        # Stop the pool
        m.execute(f"stratis pool stop {self.stop_type_opt} pool0")
        b.wait_in_text(self.card_row("Storage", name="pool0"), "Stratis pool (stopped)")

        # Start it
        self.click_dropdown(self.card_row("Storage", name="pool0"), "Start")
        b.wait_in_text(self.card_row("Storage", name="pool0"), "Stratis filesystems")

        self.click_card_row("Storage", name="pool0")
        b.wait_text(self.card_desc("Stratis pool", "Name"), "pool0")
        b.wait_in_text(self.card_desc("Stratis pool", "Usage"), "8 GB")
        b.wait_not_present('.pf-v6-c-alert')

        udisk_contains_stratis_private = "physical-originsub" in m.execute("udisksctl dump")

        # Create two filesystems
        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog_wait_open()
        self.dialog_set_val('name', 'fsys1')
        self.dialog_set_val('mount_point', '/run/fsys1')
        if not legacy:
            b.assert_pixels("#dialog", "create-fsys")
        self.dialog_apply()
        self.dialog_wait_close()
        self.addCleanupMount("/run/fsys1")

        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1")
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1")

        self.assertEqual(self.inode(m.execute("findmnt -n -o SOURCE /run/fsys1").strip()),
                         self.inode("/dev/stratis/pool0/fsys1"))

        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog({'name': 'fsys2',
                     'mount_point': '/run/fsys2'})
        self.addCleanupMount("/run/fsys2")
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 1), "fsys2")
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 3), "/run/fsys2")
        if not legacy:
            b.assert_pixels(self.card("Stratis filesystems"), "fsys-rows")
        self.assertEqual(self.inode(m.execute("findmnt -n -o SOURCE /run/fsys2").strip()),
                         self.inode("/dev/stratis/pool0/fsys2"))
        m.write("/run/fsys2/FILE", "Hello Stratis!")

        # Check that they have entries in fstab
        self.assertNotEqual(m.execute("grep /run/fsys1 /etc/fstab"), "")
        self.assertNotEqual(m.execute("grep /run/fsys2 /etc/fstab"), "")

        # Rename one filesystem
        self.click_card_row("Stratis filesystems", 1)
        b.click(self.card_desc_action("Stratis filesystem", "Name"))
        self.dialog({'name': "fsys1-renamed"})
        b.wait_text(self.card_desc("Stratis filesystem", "Name"), "fsys1-renamed")
        b.click(self.card_parent_link())
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1-renamed")

        # Destroy one filesystem
        self.click_card_row("Stratis filesystems", 1)
        self.click_card_dropdown("Stratis filesystem", "Delete")
        self.dialog_wait_open()
        if not legacy:
            b.assert_pixels("#dialog", "delete-fsys")
        self.dialog_apply()
        self.dialog_wait_close()
        b.wait_visible(self.card("Stratis filesystems"))
        b.wait_not_present(self.card_row("Stratis filesystems", name="fsys1-renamed"))

        # Unmount and remount the other filesystem
        self.click_dropdown(self.card_row("Stratis filesystems", 1), "Unmount")
        self.confirm()
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys2 (not mounted)")
        self.click_dropdown(self.card_row("Stratis filesystems", 1), "Mount")
        self.dialog({})
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys2")

        # Make a copy of the filesystem
        self.click_dropdown(self.card_row("Stratis filesystems", 1), "Snapshot")
        self.dialog_wait_open()
        self.dialog_set_val('name', 'fsys2-copy')
        self.dialog_set_val('mount_point', '/run/fsys2-copy')
        self.dialog_set_val('at_boot', 'never')
        if not legacy:
            b.assert_pixels("#dialog", "copy-fsys")
        self.dialog_apply()
        self.dialog_wait_close()
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 1), "fsys2-copy")
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 3), "/run/fsys2-copy")

        self.assertEqual("Hello Stratis!", m.execute("cat /run/fsys2-copy/FILE"))

        # Delete the copy
        self.click_card_row("Stratis filesystems", 2)
        self.click_card_dropdown("Stratis filesystem", "Delete")
        self.confirm()
        b.wait_visible(self.card("Stratis filesystems"))
        b.wait_not_present(self.card_row("Stratis filesystems", name="fsys2-copy"))

        # Make an unmounted copy of the filesystem
        self.click_dropdown(self.card_row("Stratis filesystems", 1), "Snapshot")
        self.dialog_wait_open()
        self.dialog_set_val('name', 'fsys2-copy')
        self.dialog_set_val('at_boot', 'never')
        self.dialog_apply_secondary()
        self.dialog_wait_close()
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 1), "fsys2-copy")
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 3), "(not mounted)")

        # Delete the copy
        self.click_card_row("Stratis filesystems", 2)
        self.click_card_dropdown("Stratis filesystem", "Delete")
        self.confirm()
        b.wait_visible(self.card("Stratis filesystems"))
        b.wait_not_present(self.card_row("Stratis filesystems", name="fsys2-copy"))

        # Create an unmounted filesystem
        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog_wait_open()
        self.dialog_set_val('name', 'fsys-unmounted')
        self.dialog_apply_secondary()
        self.dialog_wait_close()

        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys-unmounted")
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "(not mounted)")

        # Delete the unmounted filesystem
        self.click_card_row("Stratis filesystems", 1)
        self.click_card_dropdown("Stratis filesystem", "Delete")
        self.confirm()
        b.wait_visible(self.card("Stratis filesystems"))
        b.wait_not_present(self.card_row("Stratis filesystems", name="fsys-unmounted"))

        # Add a data blockdev
        b.click(self.card_button("Stratis pool", "Add block device"))
        self.dialog_wait_open()
        self.dialog_apply()
        self.dialog_wait_error("disks", "At least one")
        self.dialog_set_val('disks', {dev_3: True})
        if not legacy:
            b.assert_pixels("#dialog", "add-disk")
        self.dialog_apply()
        self.dialog_wait_close()

        b.wait_visible(self.card_row("Stratis pool", name=dev_3))
        b.wait_in_text(self.card_desc("Stratis pool", "Usage"), "12 GB")

        # Add a cache blockdev
        b.click(self.card_button("Stratis pool", "Add block device"))
        self.dialog({'tier': "cache",
                     'disks': {dev_4: True}})
        b.wait_in_text(self.card_row("Stratis pool", name=dev_4), "cache")

        # Add a second cache blockdev, this uses a different code path
        b.click(self.card_button("Stratis pool", "Add block device"))
        self.dialog({'tier': "cache",
                     'disks': {dev_5: True}})
        b.wait_in_text(self.card_row("Stratis pool", name=dev_5), "cache")

        # Rename the pool
        b.click(self.card_desc_action("Stratis pool", "Name"))
        self.dialog({'name': "pool0-renamed"})
        b.wait_text(self.card_desc("Stratis pool", "Name"), "pool0-renamed")

        # Create another filesystem
        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog({'name': 'fsys3',
                     'mount_point': '/run/fsys3'})
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 1), "fsys3")
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 3), "/run/fsys3")
        self.assertEqual(self.inode(m.execute("findmnt -n -o SOURCE /run/fsys3").strip()),
                         self.inode("/dev/stratis/pool0-renamed/fsys3"))

        # Destroy the pool
        self.click_card_dropdown("Stratis pool", "Delete")
        self.dialog_wait_open()
        if not legacy:
            b.assert_pixels('#dialog', "delete-pool")
        self.dialog_apply()
        self.dialog_wait_close()
        b.wait_visible(self.card("Storage"))
        b.wait_not_present(self.card_row("Storage", name="pool0-renamed"))

        # Check that the entries have disappeared from fstab
        self.assertEqual(m.execute("grep /run/fsys1 /etc/fstab || true"), "")
        self.assertEqual(m.execute("grep /run/fsys2 /etc/fstab || true"), "")
        self.assertEqual(m.execute("grep /run/fsys3 /etc/fstab || true"), "")

        m.execute("! findmnt /run/fsys1")
        m.execute("! findmnt /run/fsys2")
        m.execute("! findmnt /run/fsys2-copy")
        m.execute("! findmnt /run/fsys3")

        # https://bugzilla.redhat.com/show_bug.cgi?id=2183084
        # Do this assertion in the end so that the previous checks still run.
        # After the stratis pool is deleted we can't check this, so use the value from earlier.
        self.assertFalse(udisk_contains_stratis_private)

    @testlib.skipImage("Stratis too old for legacy tests", *V1_POOL_IMAGES)
    def testBasicLegacy(self):
        self.testBasic(legacy=True)

    @testlib.skipImage("Stratis too old", "rhel-8-*")
    def testAlerts(self):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        dev_1 = self.add_loopback_disk(PV_SIZE)
        dev_2 = self.add_loopback_disk(PV_SIZE)
        b.wait_visible(self.card_row("Storage", name=dev_1))
        b.wait_visible(self.card_row("Storage", name=dev_2))

        # Create an encrypted V1 pool with two block devices

        if m.image not in V1_POOL_IMAGES:
            create_pool_key(m, "pool0", "foodeeboodeebar")
            create_legacy_pool(m, disks=[dev_1, dev_2], keydesc="pool0")
            m.execute("stratis key unset pool0")
        else:
            self.click_devices_dropdown("Create Stratis pool")
            self.dialog_wait_open()
            self.dialog_set_val("encrypt_pass.on", val=True)
            self.dialog_set_val("passphrase", "foodeeboodeebar")
            self.dialog_set_val("passphrase2", "foodeeboodeebar")
            self.dialog_set_val("disks", {dev_1: True})
            self.dialog_set_val("disks", {dev_2: True})
            self.dialog_apply()
            self.dialog_wait_close()

        b.wait_visible(self.card_row("Storage", name="pool0"))
        b.wait_not_present(self.card_row("Storage", name="pool0") + " .ct-icon-exclamation-triangle")

        # Check that there is no alert on the details page
        self.click_card_row("Storage", name="pool0")
        b.wait_visible(self.card("Encrypted Stratis pool"))
        b.wait_not_present('.pf-v6-c-alert')

        m.execute(f"""
JSON=$(sudo cryptsetup token export --token-id=1 {dev_1} \
       | jq '.key_description = "stratis-1-key-no-other-is-the-same"')
sudo cryptsetup token remove --token-id=1 {dev_1}
echo $JSON | sudo cryptsetup token import --token-id=1 {dev_1}
systemctl restart stratisd
        """)

        b.go('#/')
        b.wait_visible(self.card_row("Storage", name="pool0") + " .ct-icon-exclamation-triangle")

        self.click_card_row("Storage", name="pool0")
        b.wait_visible('.pf-v6-c-alert:contains("This pool is in a degraded state")')

        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog_wait_open()
        self.dialog_set_val("name", "fsys1")
        self.dialog_set_val("mount_point", "/run/fsys1")
        self.dialog_apply()
        self.dialog_wait_alert("Pool is in state NoRequests where this action cannot be performed until the issue is resolved manually")

    @testlib.nondestructive
    def testCli(self):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        dev_1 = self.add_loopback_disk(PV_SIZE)
        dev_2 = self.add_loopback_disk(PV_SIZE)
        b.wait_visible(self.card_row("Storage", name=dev_1))
        b.wait_visible(self.card_row("Storage", name=dev_2))

        # Create a pool outside of Cockpit
        m.execute(f"stratis pool create TEST1 {dev_1} {dev_2}")
        b.wait_visible(self.card_row("Storage", name="TEST1"))
        b.wait_in_text(self.card_row("Storage", name=dev_1), "Stratis block device")
        b.wait_in_text(self.card_row("Storage", name=dev_1), "TEST1")
        b.wait_in_text(self.card_row("Storage", name=dev_2), "Stratis block device")
        b.wait_in_text(self.card_row("Storage", name=dev_1), "TEST1")

        # Create two filesystems outside of Cockpit
        m.execute("stratis filesystem create TEST1 fsys1")
        b.wait_visible(self.card_row("Storage", name="fsys1"))
        m.execute("stratis filesystem create TEST1 fsys2")
        b.wait_visible(self.card_row("Storage", name="fsys2"))

        mount = f"{self.vm_tmpdir}/fsys1"

        # Mount externally, adjust fstab with Cockpit
        self.click_card_row("Storage", name="fsys1")
        m.execute(f"mkdir {mount}; mount /dev/stratis/TEST1/fsys1 {mount}")
        b.click(self.card_button("Stratis filesystem", f"Mount automatically on {mount} on boot"))
        b.wait_not_present(self.card_button("Stratis filesystem", f"Mount automatically on {mount} on boot"))
        self.assertIn("stratis-fstab-setup", m.execute(f"grep {mount} /etc/fstab"))

        # Unmount externally, adjust fstab with Cockpit
        m.execute(f"umount {mount}")
        b.click(self.card_button("Stratis filesystem", "Do not mount automatically on boot"))
        b.wait_not_present(self.card_button("Stratis filesystem", "Do not mount automatically on boot"))
        self.assertIn("noauto", m.execute(f"grep {mount} /etc/fstab"))

        # Destroy them outside of Cockpit
        b.click(self.card_parent_link())
        b.wait_visible(self.card("Stratis filesystems"))
        m.execute("stratis filesystem destroy TEST1 fsys1")
        b.wait_not_present(self.card_row("Stratis filesystems", name="fsys1"))
        m.execute("stratis filesystem destroy TEST1 fsys2")
        b.wait_not_present(self.card_row("Stratis filesystems", name="fsys2"))

        # Destroy the pool outside of Cockpit
        m.execute("stratis pool destroy TEST1")
        b.wait_in_text("main", "Not found")

        b.go("#/")
        b.wait_visible(self.card("Storage"))
        b.wait_not_present(self.card_row("Storage", name="TEST1"))

    def testOverprovisioning(self, legacy=False):
        b = self.browser

        self.login_and_go("/storage")

        dev_1 = self.add_loopback_disk(PV_SIZE)
        b.wait_visible(self.card_row("Storage", name=dev_1))

        # Create a pool with overprov off
        if not legacy:
            self.dialog_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                                   expect=lambda: (self.dialog_is_present('disks', dev_1) and
                                                   self.dialog_check({"name": "pool0"})),
                                   values={
                                       "disks": {dev_1: True},
                                       "overprov.on": False
                                   })
        else:
            create_legacy_pool(self.machine, disks=[dev_1])
            self.machine.execute("stratis pool overprovision pool0 no")

        self.click_card_row("Storage", name="pool0")

        overprov_toggle = "input[aria-label='Allow overprovisioning']"
        b.wait_visible(f"{overprov_toggle}:not(:checked)")

        # Create filesystem with defaults.  Initial virtual size
        # should be appropriate.

        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog_wait_open()
        self.dialog_wait_val('set_custom_size.enabled', val=True)
        self.dialog_set_val('name', 'fsys1')
        self.dialog_apply_secondary()
        self.dialog_wait_close()

        # Snapshot should be disabled now

        self.check_dropdown_action_disabled(self.card_row("Stratis filesystems", name="fsys1"), "Snapshot", "Not enough free space")

        # Switch overprov on

        b.click(overprov_toggle)
        b.wait_visible(f"{overprov_toggle}:checked")

        # Snapshot succeeds

        self.click_dropdown(self.card_row("Stratis filesystems", name="fsys1"), "Snapshot")
        self.dialog_wait_open()
        self.dialog_set_val('name', 'snap1')
        self.dialog_apply_secondary()
        self.dialog_wait_close()

        # Filesystem creation does not ask for file size by default

        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog_wait_open()
        self.dialog_wait_val('set_custom_size.enabled', val=False)
        self.dialog_cancel()
        self.dialog_wait_close()

        # Turning overprov off should be disabled now

        b.wait_visible(overprov_toggle + "[disabled]")

        # Delete snapshot

        self.click_dropdown(self.card_row("Stratis filesystems", name="snap1"), "Delete")
        self.confirm()

        # Switch overprov off

        b.click(overprov_toggle)
        b.wait_visible(f"{overprov_toggle}:not(:checked)")

    @testlib.skipImage("Stratis too old for legacy tests", *V1_POOL_IMAGES)
    def testOverprovisioningLegacy(self):
        self.testOverprovisioning(legacy=True)

    def testFilesystemSizes(self, legacy=False):
        b = self.browser

        self.login_and_go("/storage")

        dev_1 = self.add_loopback_disk(PV_SIZE)
        b.wait_visible(self.card_row("Storage", name=dev_1))

        # Create a pool
        if not legacy:
            self.dialog_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                                   expect=lambda: (self.dialog_is_present('disks', dev_1) and
                                                   self.dialog_check({"name": "pool0"})),
                                   values={"disks": {dev_1: True}})
        else:
            create_legacy_pool(self.machine, disks=[dev_1])

        self.click_card_row("Storage", name="pool0")

        # Create filesystem with initial virtual size and a limit

        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog_wait_open()
        self.dialog_set_val('name', 'fsys1')
        self.dialog_set_val('set_custom_size.enabled', val=True)
        self.dialog_set_val('size', 800)
        self.dialog_set_val('set_custom_limit.enabled', val=True)
        self.dialog_set_val('limit', 1600)
        self.dialog_apply_secondary()
        self.dialog_wait_close()

        self.click_card_row("Stratis filesystems", name="fsys1")

        b.wait_text(self.card_desc("Stratis filesystem", "Virtual size"), "800 MB")
        b.wait_text(self.card_desc("Stratis filesystem", "Virtual size limit"), "1.60 GB")

        # Raise the limit

        b.click(self.card_desc_action("Stratis filesystem", "Virtual size limit"))
        self.dialog_wait_open()
        self.dialog_wait_val('size_options.custom_limit', val=True)
        self.dialog_wait_val('limit', 1600)
        self.dialog_set_val('limit', 3200)
        self.dialog_apply()
        self.dialog_wait_close()

        b.wait_text(self.card_desc("Stratis filesystem", "Virtual size limit"), "3.20 GB")

        # Remove limit

        b.click(self.card_desc_action("Stratis filesystem", "Virtual size limit"))
        self.dialog_wait_open()
        self.dialog_set_val('size_options.custom_limit', val=False)
        self.dialog_apply()
        self.dialog_wait_close()

        b.wait_text(self.card_desc("Stratis filesystem", "Virtual size limit"), "none")

    @testlib.skipImage("Stratis too old for legacy tests", *V1_POOL_IMAGES)
    def testFilesystemSizesLegacy(self):
        self.testFilesystemSizes(legacy=True)


@testlib.skipImage("No Stratis", "debian-*", "ubuntu-*", "arch")
@testlib.skipImage("stratis not installed", "*-bootc")
@testlib.skipImage("commit 817c957899a4 removed Stratis 2 support", "rhel-8-*")
class TestStorageStratisReboot(storagelib.StorageCase):
    def setUp(self):
        super().setUp()
        exe = self.machine.execute

        if self.image == "arch":
            # Arch Linux does not enable systemd units by default
            exe("systemctl enable --now stratisd")
            self.addCleanup(self.machine.execute, "systemctl disable --now stratisd")

        # these tests reboot
        self.allow_restart_journal_messages()

    def testEncrypted(self, legacy=False):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        dev_1 = "/dev/sda"
        m.add_disk("4G", serial="DISK1")
        b.wait_visible(self.card_row("Storage", name=dev_1))

        dev_2 = "/dev/sdb"
        m.add_disk("4G", serial="DISK2")
        b.wait_visible(self.card_row("Storage", name=dev_2))

        dev_3 = "/dev/sdc"
        m.add_disk("4G", serial="DISK3")
        b.wait_visible(self.card_row("Storage", name=dev_3))

        passphrase = "foodeeboodeebar"

        # Create an encrypted pool with a filesystem, but don't mount
        # it.  Cockpit will chose a key description for the pool and
        # we occupy its first choice in order to force Cockpit to use
        # something else.
        create_pool_key(m, "pool0", "not-the-passphrase")
        if not legacy:
            self.dialog_open_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                                        expect=lambda: (self.dialog_is_present('disks', dev_1) and
                                                        self.dialog_check({"name": "pool0"})))
            self.dialog_set_val("encrypt_pass.on", val=True)
            self.dialog_set_val("passphrase", passphrase)
            self.dialog_set_val("passphrase2", passphrase)
            self.dialog_set_val("disks", {dev_1: True})
            b.assert_pixels("#dialog", "create-encrypted-pool",
                            # The small checkbox ticks render inconsistently
                            ignore=["input[type=checkbox]"])
            self.dialog_apply()
            self.dialog_wait_close()
        else:
            create_pool_key(m, "pool0.1", passphrase)
            create_legacy_pool(m, disks=[dev_1], keydesc="pool0.1")
            m.execute("stratis key unset pool0.1")
        m.execute("stratis key unset pool0")

        v1_pool = legacy or m.image in V1_POOL_IMAGES

        self.click_card_row("Storage", name="pool0")
        b.wait_visible(self.card("Encrypted Stratis pool"))

        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog({'name': 'fsys1',
                     'mount_point': '/run/fsys1',
                     'at_boot': 'local'},
                    secondary=True)
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1")
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1 (not mounted)")

        # Check that it has an entry in fstab and that it is "noauto"
        self.assertIn("noauto", m.execute("grep /run/fsys1 /etc/fstab"))

        # Add a data blockdev
        b.click(self.card_button("Encrypted Stratis pool", "Add block device"))
        self.dialog_wait_open()
        self.dialog_set_val('disks', {dev_2: True})
        self.dialog_apply()
        if v1_pool:
            self.dialog_wait_error("pool_passphrase", "Passphrase cannot be empty")
            self.dialog_set_val('pool_passphrase', passphrase)
            self.dialog_apply()
        self.dialog_wait_close()
        b.wait_in_text(self.card_row("Encrypted Stratis pool", name=dev_2), "data")

        # Change the passphrase
        if v1_pool:
            b.wait_visible(self.card_desc("Encryption tokens", "Passphrase") + " button:contains('Remove'):disabled")
            b.click(self.card_desc("Encryption tokens", "Passphrase") + " button:contains('Change')")
        else:
            self.click_dropdown(self.card_row("Encryption tokens", index=1), "Change")
        self.dialog({'old_passphrase': passphrase,
                     'new_passphrase': "boodeefoodeebar",
                     'new_passphrase2': "boodeefoodeebar"})
        # do it again, with the old passphrase in the keyring
        create_pool_key(m, "pool0", "boodeefoodeebar")
        if v1_pool:
            b.click(self.card_desc("Encryption tokens", "Passphrase") + " button:contains('Change')")
        else:
            self.click_dropdown(self.card_row("Encryption tokens", index=1), "Change")
        self.dialog({'new_passphrase': passphrase,
                     'new_passphrase2': passphrase})
        m.execute("stratis key unset pool0")

        if not v1_pool:
            # Add a second passphrase
            b.wait_text(self.card_row_col("Encryption tokens", 1, 2), "Passphrase")
            b.click(self.card_button("Encryption tokens", "Add passphrase"))
            self.dialog({'pool_passphrase': passphrase,
                         'passphrase': "boodeefoodeebar",
                         'passphrase2': "boodeefoodeebar"})
            b.wait_text(self.card_row_col("Encryption tokens", 2, 2), "Passphrase")
            # And remove it again
            self.click_dropdown(self.card_row("Encryption tokens", index=2), "Remove")
            self.confirm()

        # Add a cache blockdev
        b.click(self.card_button("Encrypted Stratis pool", "Add block device"))
        self.dialog_wait_open()
        self.dialog_set_val('tier', "cache")
        self.dialog_set_val('disks', {dev_3: True})
        if v1_pool:
            self.dialog_set_val('pool_passphrase', passphrase)
        self.dialog_apply()
        self.dialog_wait_close()
        b.wait_in_text(self.card_row("Encrypted Stratis pool", name=dev_3), "cache")

        self.reboot()
        m.start_cockpit()
        b.relogin()
        b.enter_page("/storage")

        b.wait_visible(self.card("Stratis pool"))
        b.wait_in_text(self.card("Stratis pool"), "DISK1")
        b.wait_in_text(self.card("Stratis pool"), "DISK2")

        # Unlock the pool
        b.click(self.card_button("Stratis pool", "Start"))
        self.dialog_wait_open()
        self.dialog_set_val('passphrase', "wrong-passphrase")
        self.dialog_apply()
        with b.wait_timeout(60):
            b.wait_visible("#dialog .pf-v6-c-alert.pf-m-danger")
        self.dialog_set_val('passphrase', passphrase)
        self.dialog_apply()
        self.dialog_wait_close()
        b.wait_text(self.card_desc("Encrypted Stratis pool", "Name"), "pool0")

        # Mount the filesystem
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1 (not mounted)")
        self.click_dropdown(self.card_row("Stratis filesystems", 1), "Mount")
        self.dialog({})
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1")

        # Reboot (this requires the passphrase)
        self.setup_systemd_password_agent(passphrase)
        self.reboot()
        m.start_cockpit()
        b.relogin()
        b.enter_page("/storage")
        b.wait_text(self.card_desc("Encrypted Stratis pool", "Name"), "pool0")

        # Filesystem should be mounted now
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1")

        # Destroy the pool
        self.click_card_dropdown("Encrypted Stratis pool", "Delete")
        self.confirm()
        b.wait_visible(self.card("Storage"))
        b.wait_not_present(self.card_row("Storage", name="pool0"))

        # Check that the entry has disappeared from fstab
        self.assertEqual(m.execute("grep /run/fsys1 /etc/fstab || true"), "")

    @testlib.skipImage("Stratis too old for legacy tests", *V1_POOL_IMAGES)
    def testEncryptedLegacy(self):
        self.testEncrypted(legacy=True)

    def testReboot(self, legacy=False):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        dev_1 = "/dev/sda"
        m.add_disk("4G", serial="DISK1")
        b.wait_visible(self.card_row("Storage", name=dev_1))

        # Create a pool
        if not legacy:
            self.dialog_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                                   expect=lambda: (self.dialog_is_present('disks', dev_1) and
                                                   self.dialog_check({"name": "pool0"})),
                                   values={"disks": {dev_1: True}})
        else:
            create_legacy_pool(m, disks=[dev_1])
        self.click_card_row("Storage", name="pool0")

        # Create a filesystems
        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog({'name': 'fsys1',
                     'mount_point': '/run/fsys1'})
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1")
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1")

        self.reboot()
        m.start_cockpit()
        b.relogin()
        b.enter_page("/storage")

        # Filesystem should be mounted now
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1")
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1")

    @testlib.skipImage("Stratis too old for legacy tests", *V1_POOL_IMAGES)
    def testRebootLegacy(self):
        self.testReboot(legacy=True)

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

        self.login_and_go("/storage")

        dev_1 = "/dev/sda"
        m.add_disk("4G", serial="DISK1")
        b.wait_visible(self.card_row("Storage", name=dev_1))

        # Create a pool
        self.dialog_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                               expect=lambda: (self.dialog_is_present('disks', dev_1) and
                                               self.dialog_check({"name": "pool0"})),
                               values={"disks": {dev_1: True}})
        self.click_card_row("Storage", name="pool0")

        def create(at_boot):
            b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
            self.dialog({'name': 'fsys1',
                         'mount_point': f'{self.mnt_dir}/foo',
                         'at_boot': at_boot})
            b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1")
            b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), f"{self.mnt_dir}/foo")

        def destroy():
            self.click_card_row("Stratis filesystems", 1)
            self.click_card_dropdown("Stratis filesystem", "Delete")
            self.dialog_wait_open()
            self.dialog_apply_with_retry("Device or resource busy")
            b.wait_visible(self.card("Stratis filesystems"))
            b.wait_not_present(self.card_row("Stratis filesystems", name="fsys1"))

        create("local")
        self.assertNotIn("noauto", m.execute(f"findmnt --fstab -n -o OPTIONS {self.mnt_dir}/foo"))
        destroy()

        create("nofail")
        self.assertIn("nofail", m.execute(f"findmnt --fstab -n -o OPTIONS {self.mnt_dir}/foo"))
        destroy()

        create("netdev")
        self.assertIn("_netdev", m.execute(f"findmnt --fstab -n -o OPTIONS {self.mnt_dir}/foo"))
        destroy()

        create("never")
        self.assertIn("x-cockpit-never-auto", m.execute(f"findmnt --fstab -n -o OPTIONS {self.mnt_dir}/foo"))
        self.assertIn("noauto", m.execute(f"findmnt --fstab -n -o OPTIONS {self.mnt_dir}/foo"))
        destroy()

    @testlib.skipImage("Stratis too old", "rhel-8-*")
    def testPoolResize(self, legacy=False):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        dev = "/dev/sda"
        m.add_disk("4G", serial="DISK1")
        b.wait_visible(self.card_row("Storage", name=dev))

        # Create a logical volume that we will later grow
        m.execute(f"vgcreate vgroup0 {dev}; lvcreate vgroup0 -n lvol0 -L 1500000256b")
        b.wait_visible(self.card_row("Storage", name="lvol0"))

        # Create a pool
        if not legacy:
            self.dialog_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                                   expect=lambda: self.dialog_is_present('disks', "/dev/vgroup0/lvol0"),
                                   values={"disks": {"/dev/vgroup0/lvol0": True}})
        else:
            create_legacy_pool(m, disks=["/dev/vgroup0/lvol0"])

        b.wait_in_text(self.card_row("Storage", name="pool0"), "1.5 GB")

        # Grow the logical volume in Cockpit, the pool should grow automatically
        self.click_card_row("Storage", name="lvol0")
        b.click(self.card_button("LVM2 logical volume", "Grow"))
        self.dialog({"size": 1600})
        b.go("#/")
        b.wait_in_text(self.card_row("Storage", name="pool0"), "1.6 GB")

        # Grow the logical volume from outside of Cockpit, the pool should complain
        m.execute("lvresize vgroup0/lvol0 -L +100000256b")
        b.wait_visible(self.card_row("Storage", name="pool0") + ' .ct-icon-exclamation-triangle')
        self.click_card_row("Storage", name="pool0")
        b.wait_visible('.pf-v6-c-alert:contains("This pool does not use all the space")')
        b.click('button:contains("Grow the pool")')
        b.wait_not_present('.pf-v6-c-alert')
        b.wait_in_text(self.card_desc("Stratis pool", "Usage"), "1.7 GB")

        b.go("#/")

        # Grow the logical volume from outside of Cockpit, the logical volume should also complain
        m.execute("lvresize vgroup0/lvol0 -L +100000256b")
        b.wait_visible(self.card_row("Storage", name="lvol0") + " .ct-icon-exclamation-triangle")
        self.click_card_row("Storage", name="lvol0")
        # First shrink the volume to test whether Cockpit can figure out the right size for that
        b.click(self.card_button("LVM2 logical volume", "Shrink volume"))
        b.wait_in_text(self.card_desc("LVM2 logical volume", "Size"), "1.70 GB")
        b.wait_not_present(self.card_button("LVM2 logical volume", "Shrink volume"))
        # Then enlarge the volume from the outside again and grow the blockdev
        m.execute("lvresize vgroup0/lvol0 -L +100000256b")
        b.click(self.card_button("LVM2 logical volume", "Grow content"))
        b.wait_not_present(self.card_button("LVM2 logical volume", "Grow content"))
        b.go("#/")
        b.wait_in_text(self.card_row("Storage", name="pool0"), "1.8 GB")

    @testlib.skipImage("Stratis too old for legacy tests", *V1_POOL_IMAGES)
    def testPoolResizeLegacy(self):
        self.testPoolResize(legacy=True)


@testlib.skipImage("No Stratis", "debian-*", "ubuntu-*", "arch")
@testlib.skipOstree("no package installation on OSTree")
class TestStoragePackagesStratis(packagelib.PackageCase, storagelib.StorageCase):

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

        # RHEL 8 should not offer installation of Stratis from Cockpit
        # itself.
        #
        ondemand_stratis = "rhel-8" not in m.image

        m.execute("systemctl stop stratisd && dnf remove -y stratisd stratis")
        if ondemand_stratis:
            self.addPackageSet("stratis")
            self.enableRepo()

        self.login_and_go("/storage")

        dev_1 = "/dev/sda"
        m.add_disk("4G", serial="DISK1")
        b.wait_visible(self.card_row("Storage", name=dev_1))

        if ondemand_stratis:
            self.click_devices_dropdown("Create Stratis pool")
            self.dialog_wait_open()
            b.wait_in_text("#dialog", "The stratisd package must be installed")
            self.dialog_apply()
            with b.wait_timeout(60):
                self.dialog_wait_val("name", "pool0")
            self.dialog_set_val("disks", {dev_1: True})
            self.dialog_apply()
            self.dialog_wait_close()
            b.wait_visible(self.card_row("Storage", name="pool0"))
        else:
            dropdown_toggle = self.dropdown_toggle(self.card_header("Storage"))
            raid_action = self.dropdown_action("Create MDRAID device")
            stratis_action = self.dropdown_action("Create Stratis pool")
            b.click(dropdown_toggle)
            b.wait_visible(raid_action)
            b.wait_not_present(stratis_action)


@testlib.skipImage("No Stratis", "debian-*", "ubuntu-*", "arch")
@testlib.skipImage("stratis not installed", "*-bootc")
@testlib.skipImage("Stratis too old", "rhel-8-*")
class TestStorageStratisNBDE(packagelib.PackageCase, storagelib.StorageCase):
    provision = {
        "0": {"address": "10.111.112.1/20", "memory_mb": 2048},
        "tang": {"address": "10.111.112.5/20"}
    }

    def setUp(self):
        super().setUp()

        if self.image == "arch":
            # Arch Linux does not enable systemd units by default
            self.machine.execute("systemctl enable --now stratisd")
            self.addCleanup(self.machine.execute, "systemctl disable --now stratisd")

        self.stop_type_opt = get_stratis_stop_type_opt(self.machine.execute)

    def testBasic(self, legacy=False):
        m = self.machine
        b = self.browser

        tang_m = self.machines["tang"]
        tang_m.execute("systemctl start tangd.socket")
        tang_m.execute("firewall-cmd --add-port 80/tcp")

        self.login_and_go("/storage")

        dev_1 = "/dev/sda"
        m.add_disk("4G", serial="DISK1")
        b.wait_visible(self.card_row("Storage", name=dev_1))

        dev_2 = "/dev/sdb"
        m.add_disk("5G", serial="DISK2")
        b.wait_visible(self.card_row("Storage", name=dev_2))

        # Create an encrypted pool with both a passphrase and a keyserver
        if not legacy:
            self.dialog_open_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                                        expect=lambda: (self.dialog_is_present('disks', dev_1) and
                                                        self.dialog_check({"name": "pool0"})))
            self.dialog_set_val("encrypt_pass.on", val=True)
            self.dialog_set_val("passphrase", "foodeeboodeebar")
            self.dialog_set_val("passphrase2", "foodeeboodeebar")
            self.dialog_set_val("encrypt_tang.on", val=True)
            self.dialog_set_val("tang_url", "10.111.112.5")
            self.dialog_set_val("disks", {dev_1: True})
            self.dialog_apply()
            b.wait_in_text("#dialog", "Check the key hash")
            b.wait_in_text("#dialog", tang_m.execute("tang-show-keys").strip())
            self.dialog_apply()
            with b.wait_timeout(60):
                self.dialog_wait_close()
        else:
            create_pool_key(m, "pool0", "foodeeboodeebar")
            create_legacy_pool(m, disks=[dev_1], keydesc="pool0", tang="10.111.112.5")
            m.execute("stratis key unset pool0")

        v1_pool = legacy or m.image in V1_POOL_IMAGES

        self.click_card_row("Storage", name="pool0")
        if v1_pool:
            b.wait_visible(self.card_desc("Encryption tokens", "Passphrase"))
            b.wait_in_text(self.card_desc("Encryption tokens", "Keyserver"), "10.111.112.5")
        else:
            b.wait_visible(self.card_row("Encryption tokens", location="passphrase"))
            b.wait_visible(self.card_row("Encryption tokens", location="10.111.112.5"))

        if not legacy:
            b.assert_pixels(self.card("Encrypted Stratis pool"), "header",
                            ignore=['.pf-v6-c-description-list__group:contains("UUID")'])

        # Remove passphrase
        if v1_pool:
            b.click(self.card_desc("Encryption tokens", "Passphrase") + " button:contains('Remove')")
            self.confirm()
            b.wait_visible(self.card_desc("Encryption tokens", "Passphrase") + " button:contains('Add passphrase')")
            b.wait_visible(self.card_desc("Encryption tokens", "Keyserver") + " button:contains('Remove'):disabled")
        else:
            self.click_dropdown(self.card_row("Encryption tokens", index=1), "Remove")
            self.confirm()
            row = self.card_row("Encryption tokens", location="10.111.112.5")
            b.wait_visible(row)
            b.click(self.dropdown_toggle(row))
            b.wait_visible(self.dropdown_action("Remove") + "[disabled]")
            b.click(self.dropdown_toggle(row))

        # Stop the pool and start it again.  This should not ask
        # for the passphrase (since there isn't any)
        m.execute(f"stratis pool stop {self.stop_type_opt} pool0")
        b.wait_visible(self.card("Stratis pool"))
        tang_m.execute("systemctl stop tangd.socket")
        b.click(self.card_button("Stratis pool", "Start"))
        self.dialog_wait_open()
        # stratis' error message for unreachable tang server is very poor:
        # https://bugzilla.redhat.com/show_bug.cgi?id=2246920 Version < 3.6.0 said
        # "Error communicating with server 10.111.112.5", check this again after fixing
        b.wait_in_text("#dialog", "Error")
        self.dialog_cancel()
        self.dialog_wait_close()

        tang_m.execute("systemctl start tangd.socket")
        b.click(self.card_button("Stratis pool", "Start"))
        b.wait_visible(self.card("Encrypted Stratis pool"))

        # Put passphrase back and do the stopping starting again, but
        # without tangd running.  This should try clevis but then fall
        # back to asking for a passphrase.

        b.click(self.card_button("Encryption tokens", "Add passphrase"))
        self.dialog({'passphrase': "foodeeboodeebar",
                     'passphrase2': "foodeeboodeebar"})
        if v1_pool:
            b.wait_visible(self.card_desc("Encryption tokens", "Passphrase") + " button:contains('Remove'):not(:disabled)")
        else:
            b.wait_visible(self.card_row("Encryption tokens", location="passphrase"))
            b.click(self.dropdown_toggle(row))
            b.wait_visible(self.dropdown_action("Remove") + ":not([disabled])")
            b.click(self.dropdown_toggle(row))

        m.execute(f"stratis pool stop {self.stop_type_opt} pool0")
        tang_m.execute("systemctl stop tangd.socket")
        b.click(self.card_button("Stratis pool", "Start"))
        self.dialog_wait_open()
        self.dialog_set_val("passphrase", "foobar")
        self.dialog_cancel()
        self.dialog_wait_close()

        # Finally start tang
        tang_m.execute("systemctl start tangd.socket")
        b.click(self.card_button("Stratis pool", "Start"))
        b.wait_visible(self.card("Encrypted Stratis pool"))

        # Add a blockdevice.

        b.click(self.card_button("Encrypted Stratis pool", "Add block device"))
        if v1_pool:
            self.dialog({'disks': {dev_2: True}, 'pool_passphrase': "foodeeboodeebar"})
        else:
            self.dialog({'disks': {dev_2: True}})

        # Remove the keyserver and add it back

        if v1_pool:
            b.click(self.card_desc("Encryption tokens", "Keyserver") + " button:contains('Remove')")
        else:
            self.click_dropdown(self.card_row("Encryption tokens", location="10.111.112.5"), "Remove")
        self.confirm()

        if v1_pool:
            b.wait_not_in_text(self.card_desc("Encryption tokens", "Keyserver"), "10.111.112.5")
        else:
            b.wait_not_present(self.card_row("Encryption tokens", location="10.111.112.5"))

        b.click(self.card_button("Encryption tokens", "Add keyserver"))
        self.dialog_wait_open()
        self.dialog_set_val("tang_url", "10.111.112.5")
        self.dialog_set_val("pool_passphrase", "foodeeboodeebar")
        self.dialog_apply()
        b.wait_in_text("#dialog", "Check the key hash")
        b.wait_in_text("#dialog", tang_m.execute("tang-show-keys").strip())
        self.dialog_apply()
        with b.wait_timeout(60):
            self.dialog_wait_close()

        if v1_pool:
            b.wait_in_text(self.card_desc("Encryption tokens", "Keyserver"), "10.111.112.5")
        else:
            b.wait_visible(self.card_row("Encryption tokens", location="10.111.112.5"))

        # Remove the keyserver and add it back a second time, but try
        # first with the wrong passphrase already in the keyring

        if v1_pool:
            b.click(self.card_desc("Encryption tokens", "Keyserver") + " button:contains('Remove')")
        else:
            self.click_dropdown(self.card_row("Encryption tokens", location="10.111.112.5"), "Remove")
        self.confirm()

        create_pool_key(m, "pool0", "foobar")
        b.click(self.card_button("Encryption tokens", "Add keyserver"))
        self.dialog_wait_open()
        self.dialog_set_val("tang_url", "10.111.112.5")
        self.dialog_apply()
        b.wait_in_text("#dialog", "Check the key hash")
        b.wait_in_text("#dialog", tang_m.execute("tang-show-keys").strip())
        self.dialog_apply()
        with b.wait_timeout(60):
            b.wait_in_text('#dialog', "Command failed")
        m.execute("stratis key unset pool0")
        create_pool_key(m, "pool0", "foodeeboodeebar")
        self.dialog_apply()
        with b.wait_timeout(60):
            self.dialog_wait_close()

        if v1_pool:
            b.wait_in_text(self.card_desc("Encryption tokens", "Keyserver"), "10.111.112.5")
        else:
            b.wait_visible(self.card_row("Encryption tokens", location="10.111.112.5"))

        m.execute("stratis key unset pool0")

        # Create a mounted filesystem and reboot.

        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog({'name': 'fsys1',
                     'mount_point': '/run/fsys1'})
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1")
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1")
        self.reboot()
        m.start_cockpit()
        b.relogin()
        b.enter_page("/storage")
        b.wait_visible(self.card("Stratis pool"))
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1")  # should be started after boot
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1")  # should be mounted after boot

    @testlib.skipImage("Stratis too old for legacy tests", *V1_POOL_IMAGES)
    def testBasicLegacy(self):
        self.testBasic(legacy=True)

    @testlib.skipImage("Stratis too old", *V1_POOL_IMAGES)
    def testEncryptionTokens(self):
        m = self.machine
        b = self.browser

        tang_m = self.machines["tang"]
        tang_m.execute("systemctl start tangd.socket")
        tang_m.execute("firewall-cmd --add-port 80/tcp")

        self.login_and_go("/storage")

        dev_1 = "/dev/sda"
        m.add_disk("4G", serial="DISK1")
        b.wait_visible(self.card_row("Storage", name=dev_1))

        passphrase = "foodeeboodeebar"
        passphrase2 = "boodeefoodeebar"

        # Create a pool with a passphrase

        self.dialog_open_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                                    expect=lambda: (self.dialog_is_present('disks', dev_1) and
                                                    self.dialog_check({"name": "pool0"})))
        self.dialog_set_val("encrypt_pass.on", val=True)
        self.dialog_set_val("passphrase", passphrase)
        self.dialog_set_val("passphrase2", passphrase)
        self.dialog_set_val("disks", {dev_1: True})
        self.dialog_apply()
        self.dialog_wait_close()

        self.click_card_row("Storage", name="pool0")
        b.wait_visible(self.card("Encrypted Stratis pool"))

        # Adding a token needs the existing passphrase.

        b.click(self.card_button("Encryption tokens", "Add keyserver"))
        self.dialog_wait_open()
        b.wait_visible(self.dialog_field("pool_passphrase"))
        b.wait_in_text("#dialog", "Please provide an existing pool passphrase.")
        b.assert_pixels("#dialog", "dialog")
        self.dialog_cancel()
        self.dialog_wait_close()

        # When the only passphrase is already in the keyring, it must
        # work.

        create_pool_key(m, "pool0", passphrase)
        b.click(self.card_button("Encryption tokens", "Add keyserver"))
        self.dialog_wait_open()
        b.wait_in_text("#dialog", "At least one of the non-interactive unlock methods must work.")
        b.wait_not_present(self.dialog_field("pool_passphrase"))
        self.dialog_set_val("tang_url", "10.111.112.5")
        self.dialog_apply()
        b.wait_in_text("#dialog", "Check the key hash")
        self.dialog_apply()
        with b.wait_timeout(60):
            self.dialog_wait_close()

        # A single keyserver will be mentioned explicitly

        m.execute("stratis key unset pool0")
        b.click(self.card_button("Encryption tokens", "Add passphrase"))
        self.dialog_wait_open()
        self.dialog_set_val("passphrase", passphrase2)
        self.dialog_set_val("passphrase2", passphrase2)
        b.wait_visible(self.dialog_field("pool_passphrase"))
        b.wait_in_text("#dialog", "If the keyserver at 10.111.112.5 is not reachable, you can provide an existing passphrase.")
        self.dialog_apply()
        with b.wait_timeout(60):
            self.dialog_wait_close()

        # When only one passphrase is in the keyring, the other one can be provided.

        create_pool_key(m, "pool0", passphrase)
        b.click(self.card_button("Encryption tokens", "Add keyserver"))
        self.dialog_wait_open()
        b.wait_visible(self.dialog_field("pool_passphrase"))
        b.wait_in_text("#dialog", "If none of the non-interactive unlock methods works, you can provide an existing passphrase.")
        self.dialog_cancel()
        self.dialog_wait_close()

        # When there is only one tang, it is mentioned explicitly and it must work

        m.execute("stratis key unset pool0")
        self.click_dropdown(self.card_row("Encryption tokens", index=1), "Remove")
        self.confirm()
        self.click_dropdown(self.card_row("Encryption tokens", location="passphrase"), "Remove")
        self.confirm()
        b.click(self.card_button("Encryption tokens", "Add keyserver"))
        self.dialog_wait_open()
        b.wait_in_text("#dialog", "The keyserver at 10.111.112.5 must be reachable.")
        b.wait_not_present(self.dialog_field("pool_passphrase"))
        self.dialog_set_val("tang_url", "10.111.112.5")
        self.dialog_apply()
        b.wait_in_text("#dialog", "Check the key hash")
        self.dialog_apply()
        with b.wait_timeout(60):
            self.dialog_wait_close()

        # With two, one must work

        b.click(self.card_button("Encryption tokens", "Add passphrase"))
        self.dialog_wait_open()
        self.dialog_set_val("passphrase", passphrase)
        self.dialog_set_val("passphrase2", passphrase)
        b.wait_in_text("#dialog", "At least one keyserver must be reachable.")
        b.wait_not_present(self.dialog_field("pool_passphrase"))
        self.dialog_apply()
        with b.wait_timeout(60):
            self.dialog_wait_close()

        # Add TPM2 while letting Tang unlock the tokens for us

        m.execute("stratis pool bind tpm2 pool0")
        b.wait_visible(self.card_row("Encryption tokens", location="tpm2"))

        b.assert_pixels(self.card("Encryption tokens"), "tokens")

        # Remove TPM2 with Cockpit

        self.click_dropdown(self.card_row("Encryption tokens", location="tpm2"), "Remove")
        self.confirm()

        # Change the passphrase

        self.click_dropdown(self.card_row("Encryption tokens", location="passphrase"), "Change")
        self.dialog({'old_passphrase': passphrase,
                     'new_passphrase': passphrase2,
                     'new_passphrase2': passphrase2})


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