# Copyright (C) 2022 Red Hat, Inc. # SPDX-License-Identifier: LGPL-2.1-or-later import json import os import re import sys import time HELPERS_DIR = os.path.dirname(__file__) sys.path.append(HELPERS_DIR) BOTS_DIR = f'{HELPERS_DIR}/../../bots' sys.path.append(BOTS_DIR) from step_logger import log_step from steps import CUSTOM_MOUNT_POINT, INSTALLATION_METHOD from testlib import Error, wait STORAGE_SERVICE = "org.fedoraproject.Anaconda.Modules.Storage" STORAGE_INTERFACE = STORAGE_SERVICE DISK_INITIALIZATION_INTERFACE = "org.fedoraproject.Anaconda.Modules.Storage.DiskInitialization" STORAGE_OBJECT_PATH = "/org/fedoraproject/Anaconda/Modules/Storage" DISK_INITIALIZATION_OBJECT_PATH = "/org/fedoraproject/Anaconda/Modules/Storage/DiskInitialization" # Default boot partition size for partition preparation (GiB format) BOOT_PARTITION_DEFAULT_SIZE = '2GiB' # Default boot partition size for review display (GB format) BOOT_PARTITION_DEFAULT_SIZE_GB = '2.15 GB' # Default EFI system partition size for partition preparation (MiB format) EFI_PARTITION_DEFAULT_SIZE = '500MiB' # Default EFI system partition size for review display (MB format) EFI_PARTITION_DEFAULT_SIZE_MB = '524 MB' class StorageDestination(): def __init__(self, browser, machine): self.browser = browser self.machine = machine @log_step(snapshot_before=True) def check_disk_selected(self, disk, selected=True, size=None): if selected: self.browser.wait_visible(f"#{INSTALLATION_METHOD}-target-disk-{disk}") else: self.browser.wait_not_present(f"#{INSTALLATION_METHOD}-target-disk-{disk}") if size is not None: self.browser.wait_in_text(f"#{INSTALLATION_METHOD}-target-disk-{disk}", size) def check_os_detected(self, disk, value, detected=True): disk_sel_selector = f"#{INSTALLATION_METHOD}-target-disk-{disk}" if detected: self.browser.wait_in_text(disk_sel_selector, value) else: self.browser.wait_not_in_text(disk_sel_selector, value) @log_step() def wait_no_disks(self): self.browser.wait_in_text("#next-helper-text", "To continue, select the devices to install to.") @log_step(snapshots=True) def rescan_disks(self, expected_disks=None, expect_failure=False): b = self.browser b.click(f"#{INSTALLATION_METHOD}-change-destination-button") b.click(f"#{INSTALLATION_METHOD}-rescan-disks") b.wait_visible(f"#{INSTALLATION_METHOD}-rescan-disks[aria-disabled='true']") # Default 15 seconds is not always enough for re-scanning disks with b.wait_timeout(60): b.wait_not_present(f"#{INSTALLATION_METHOD}-rescan-disks[aria-disabled='true']") for disk in expected_disks or []: b.wait_visible(f"#{INSTALLATION_METHOD}-disk-selection-menu-item-{disk}") if not expect_failure: b.click(f"#{INSTALLATION_METHOD}-change-destination-modal button:contains('Cancel')") def check_constraint(self, constraint, required=True): if required: self.browser.wait_in_text("ul.cockpit-storage-integration-requirements-hint-list:first-of-type", constraint) else: self.browser.wait_in_text("ul.cockpit-storage-integration-requirements-hint-list:nth-of-type(2)", constraint) def confirm_entering_cockpit_storage(self): self.browser.click("#cockpit-storage-integration-enter-storage-confirm") def return_to_installation(self, error=None): self.browser.click("#cockpit-storage-integration-return-to-installation-button") # FIXME: https://github.com/rhinstaller/anaconda/pull/6234 # On Fedora-42 re-scanning takes long when there are LVM devices # This extra timeout can be removed once the above PR is merged with self.browser.wait_timeout(90): if error: self.browser.wait_in_text("#cockpit-storage-integration-check-storage-dialog", error) self.browser.wait_visible("#cockpit-storage-integration-check-storage-dialog-continue:not([disabled])") def return_to_installation_confirm(self): # FIXME: testBtrfsTopLevelVolume fails sometimes on CI without this workaround try: with self.browser.wait_timeout(60): self.browser.click("#cockpit-storage-integration-check-storage-dialog-continue") self.browser.wait_not_present("#cockpit-storage-integration-check-storage-dialog") except Error: # Retry the click in case the dialog is still present self.browser.click("#cockpit-storage-integration-check-storage-dialog-continue") self.browser.wait_not_present("#cockpit-storage-integration-check-storage-dialog") def return_to_installation_cancel(self): self.browser.click("#cockpit-storage-integration-check-storage-dialog-return") self.browser.wait_not_present("#cockpit-storage-integration-check-storage-dialog") def modify_storage(self): self.browser.click("#modify-storage") def enter_cockpit_storage(self): b = self.browser self.modify_storage() self.confirm_entering_cockpit_storage() b.wait_visible(".cockpit-storage-integration-sidebar") b._wait_present("iframe[name='cockpit-storage']") b.switch_to_frame("cockpit-storage") b._wait_present("#storage.ct-page-fill") def exit_cockpit_storage(self): self.browser.switch_to_top() self.return_to_installation() self.return_to_installation_confirm() class StorageEncryption(): encryption_id_prefix = "disk-encryption" def __init__(self, browser, machine): self.browser = browser self.machine = machine @log_step(snapshot_before=True) def check_encryption_selected(self, selected): sel = f"#{self.encryption_id_prefix}-encrypt-devices" if selected: self.browser.wait_visible(f'{sel}:checked') else: self.browser.wait_visible(f'{sel}:not([checked])') @log_step(snapshot_before=True) def set_encryption_selected(self, selected): sel = f"#{self.encryption_id_prefix}-encrypt-devices" self.browser.set_checked(sel, selected) @log_step(docstring=True) def check_post_install_encryption_enabled(self): """ Post-install check, encryption is enabled """ print(self.machine.execute('lsblk')) assert 'is active and is in use' in self.machine.execute('cryptsetup status /dev/mapper/luks*') def unlock_device(self, passphrase, encrypted_devices=None, successfully_unlocked_devices=None): if encrypted_devices is None: encrypted_devices = [] if successfully_unlocked_devices is None: successfully_unlocked_devices = [] # FIXME: https://github.com/patternfly/patternfly-react/issues/9512 b = self.browser for device in encrypted_devices: b.wait_in_text( "#unlock-device-dialog-locked-devices", device, ) b.set_input_text("#unlock-device-dialog-passphrase[type=password]", passphrase) b.click("#unlock-device-dialog-submit-btn") # Wait for the dialog to either close or stop being in progress with b.wait_timeout(45): if successfully_unlocked_devices == encrypted_devices: b.wait_not_present("#unlock-device-dialog") return else: b.wait_visible("#unlock-device-dialog-submit-btn:not([disabled])") # The devices that were successfully unlocked should not not be present # in the 'Locked devices' form field for device in successfully_unlocked_devices: b.wait_not_in_text("#unlock-device-dialog-locked-devices", device) # The locked devices should be present in the 'Locked devices' form field for device in list(set(encrypted_devices) - set(successfully_unlocked_devices)): b.wait_in_text("#unlock-device-dialog-locked-devices", device) # The devices that were successfully unlocked should appear in the info alert if len(successfully_unlocked_devices) > 0: b.wait_in_text( "#unlock-device-dialog .pf-v6-c-alert.pf-m-info", f"Successfully unlocked {', '.join(successfully_unlocked_devices)}." ) # If the user did not unlock any device after submiting the form expect a warning if successfully_unlocked_devices == []: fail_text = "Passphrase did not match any locked device" b.wait_in_text("#unlock-device-dialog .pf-v6-c-helper-text", fail_text) def unlock_all_encrypted(self): self.browser.click(f"#{INSTALLATION_METHOD}-unlock-devices-btn") def check_disk_encryption_keyboard_layout(self, expected_layout): self.browser.wait_val("#disk-encryption-keyboard-layout", expected_layout) class StorageUtils(StorageDestination): def __init__(self, browser, machine): self.browser = browser self.machine = machine @log_step(docstring=True) def unlock_storage_on_boot(self, password): """ Add keyfile to unlock luks encrypted storage on boot """ self.machine.write('/mnt/sysroot/root/keyfile', password, perm='0400') self.machine.write('/mnt/sysroot/root/add_keyfile.sh', ''' awk -v "KEY_FILE=/root/keyfile" '{$3=KEY_FILE; print $0}' /etc/crypttab > crypttab_mod mv -Z crypttab_mod /etc/crypttab chmod 0600 /etc/crypttab kernel_file=`grubby --default-kernel` kernel_version=`rpm -qf $kernel_file --qf '%{VERSION}-%{RELEASE}.%{ARCH}'` initrd_file="/boot/initramfs-${kernel_version}.img" dracut -f -I /root/keyfile $initrd_file $kernel_version if [ -x /sbin/zipl ]; then /sbin/zipl fi ''') self.machine.execute('chroot /mnt/sysroot bash /root/add_keyfile.sh') # partitions_params expected structure: [("size", "file system" {, "other mkfs.fs flags"})] def create_raid_device(self, name, raid_level, devices): self.machine.execute(f""" set -ex mdadm --create --run {name} --level={raid_level} --raid-devices={len(devices)} {' '.join(devices)} udevadm settle """, timeout=90) self._wait_for_mdraid_clean() def _wait_for_mdraid_clean(self): m = self.machine m.execute(""" # Wait for the md array to become active udevadm settle while [ "$(cat /sys/block/md127/md/array_state)" != "clean" ]; do echo "Waiting for RAID array to become active..." echo "$(cat /sys/block/md127/md/array_state)" sleep 0.5 done """, timeout=30) def get_luks_device_name(self, device): ret = self.machine.execute(f'cryptsetup luksUUID {device}') return f"luks-{ret.strip()}" def create_luks_partition(self, device, passphrase, luks_name, fsformat="", close_luks=True): self.machine.execute(f""" set -ex echo {passphrase} | cryptsetup luksFormat {device} echo {passphrase} | cryptsetup luksOpen {device} {luks_name} # Create a filesystem on the LUKS device if [ "{fsformat}" == "xfs" ] || [ "{fsformat}" == "btrfs" ]; then mkfs.{fsformat} -f /dev/mapper/{luks_name} elif [ "{fsformat}" == "ext4" ]; then mkfs.{fsformat} -F /dev/mapper/{luks_name} fi udevadm settle if [ "{close_luks}" == "True" ]; then cryptsetup luksClose {luks_name} fi """) def partition_disk(self, disk, partitions_params, is_mbr=False): command = "set -x\n" command += f"wipefs -a {disk}" label = "dos" if is_mbr else "gpt" partition_commands = [f'label: {label}'] for params in partitions_params: size = params[0] fstype = params[1] # Prepare the size string size_str = f"size={size}" if size else "" if is_mbr: # Determine the type code type_code = { "efi": "ef", "swap": "82", "lvmpv": "8e", "extended": "extended" }.get(fstype, "83") else: # Determine the GPT type name type_code = { "basic-data": "EBD0A0A2-B9E5-4433-87C0-68B6B72699C7", "biosboot": "BIOS boot", "efi": "C12A7328-F81F-11D2-BA4B-00A0C93EC93B", "microsoft-reserved": "E3C9E316-0B5C-4DB8-817D-F92DF00215AE", "microsoft-recovery": "DE94BBA4-06D1-4D40-A16A-BFD50179D6AC", }.get(fstype, "Linux filesystem") # Build the sfdisk line without redundant commas partition_line_elements = [] if size_str: partition_line_elements.append(size_str) if type_code: partition_line_elements.append(f"type=\"{type_code}\"") partition_line = ', '.join(partition_line_elements) partition_commands.append(partition_line) # Prepare the sfdisk script sfdisk_script = '\n'.join(partition_commands) command += f"\necho -e '{sfdisk_script}' | sfdisk {disk}" # Format the partitions partition_number = 1 logical_partition_number = 5 # Logical partitions start from 5 for params in partitions_params: fstype = params[1] # Skip formatting for extended partitions if fstype == "extended": continue # Determine the partition number (logical or primary) if fstype == "logical": device_number = logical_partition_number logical_partition_number += 1 else: device_number = partition_number partition_number += 1 # Construct the device name if "nvme" in disk: device = f"{disk}p{device_number}" else: device = f"{disk}{device_number}" mkfs_args = f"{" ".join(params[2:] if len(params) > 2 else [])} {device}" command += "\nudevadm settle" # Format the partition if fstype == "swap": mkfs = f"mkswap {mkfs_args}" elif fstype == "lvmpv": mkfs = f"pvcreate {mkfs_args}" elif fstype == "efi": mkfs = f"mkfs.vfat {mkfs_args}" elif fstype == "logical": mkfs = f"mkfs.btrfs {mkfs_args}" elif fstype in ["biosboot", "microsoft-reserved", "microsoft-recovery", "basic-data"] or fstype is None: continue else: fs = fstype mkfs = f"mkfs.{fs} {mkfs_args}" command += f"\n{mkfs}" # Execute the commands self.machine.execute(command) def udevadm_settle(self): # Workaround to not have any empty mountpoint labels self.machine.execute(""" systemctl restart systemd-udevd udevadm trigger udevadm settle --timeout=120 """) def get_lsblk_json(self): lsblk = self.machine.execute("lsblk -J") return json.loads(lsblk) class StorageDBus(): def __init__(self, machine): self.machine = machine self._bus_address = self.machine.execute("cat /run/anaconda/bus.address") def dbus_scan_devices(self): task = self.machine.execute(f'busctl --address="{self._bus_address}" \ call \ {STORAGE_SERVICE} \ {STORAGE_OBJECT_PATH} \ {STORAGE_INTERFACE} ScanDevicesWithTask') task = task.splitlines()[-1].split()[-1] self.machine.execute(f'busctl --address="{self._bus_address}" \ call \ {STORAGE_SERVICE} \ {task} \ org.fedoraproject.Anaconda.Task Start') wait(lambda: self.dbus_get_task_status(task) == "false", tries=20, delay=6) def dbus_get_task_status(self, task): ret = self.machine.execute(f'busctl --address="{self._bus_address}" \ get-property \ {STORAGE_SERVICE} \ {task} \ org.fedoraproject.Anaconda.Task IsRunning') return ret.split('b ')[1].strip().strip('"') def dbus_get_usable_disks(self): ret = self.machine.execute(f'busctl --address="{self._bus_address}" \ call \ {STORAGE_SERVICE} \ {STORAGE_OBJECT_PATH}/DiskSelection \ {STORAGE_INTERFACE}.DiskSelection GetUsableDisks') return re.findall('"([^"]*)"', ret) def dbus_reset_selected_disks(self): self.machine.execute(f'busctl --address="{self._bus_address}" \ set-property \ {STORAGE_SERVICE} \ {STORAGE_OBJECT_PATH}/DiskSelection \ {STORAGE_INTERFACE}.DiskSelection SelectedDisks as 0') def dbus_set_selected_disk(self, disk): self.machine.execute(f'busctl --address="{self._bus_address}" \ set-property \ {STORAGE_SERVICE} \ {STORAGE_OBJECT_PATH}/DiskSelection \ {STORAGE_INTERFACE}.DiskSelection SelectedDisks as 1 {disk}') def dbus_reset_scenario(self): self.machine.execute(f'busctl --address="{self._bus_address}" \ call \ {STORAGE_SERVICE} \ {STORAGE_OBJECT_PATH} \ {STORAGE_INTERFACE} ResetPartitioning') def dbus_create_partitioning(self, method="MANUAL"): return self.machine.execute(f'busctl --address="{self._bus_address}" \ call \ {STORAGE_SERVICE} \ {STORAGE_OBJECT_PATH} \ {STORAGE_INTERFACE} CreatePartitioning s {method}') def dbus_get_applied_partitioning(self): ret = self.machine.execute(f'busctl --address="{self._bus_address}" \ get-property \ {STORAGE_SERVICE} \ {STORAGE_OBJECT_PATH} \ {STORAGE_INTERFACE} AppliedPartitioning') return ret.split('s ')[1].strip().strip('"') def dbus_get_created_partitioning(self): ret = self.machine.execute(f'busctl --address="{self._bus_address}" \ get-property \ {STORAGE_SERVICE} \ {STORAGE_OBJECT_PATH} \ {STORAGE_INTERFACE} CreatedPartitioning') data = ret.split(" ") return [i.replace('"', '').replace('\n', '') for i in data[2:]] def dbus_set_initialization_mode(self, value): self.machine.execute(f'busctl --address="{self._bus_address}" \ set-property \ {STORAGE_SERVICE} \ {DISK_INITIALIZATION_OBJECT_PATH} \ {DISK_INITIALIZATION_INTERFACE} InitializationMode i -- {value}') def get_btrfs_volume_ids(self, volume_name): """Get device ids of all volumes with volume_name found.""" # The tool shows unspecified name as "none" volume_name = volume_name or "none" return [ f"BTRFS-{uuid.strip()}" for uuid in self.machine.execute( f"btrfs filesystem show | grep {volume_name} | cut -d ':' -f 3" ).split('\n')[:-1] ] class StorageScenario(): def __init__(self, browser, machine): self.machine = machine self.browser = browser def _scenario_selector(self, scenario): return f"#{INSTALLATION_METHOD}-scenario-{scenario}" def wait_scenario_visible(self, scenario, visible=True): if visible: self.browser.wait_visible(self._scenario_selector(scenario)) else: self.browser.wait_not_present(self._scenario_selector(scenario)) def wait_scenario_available(self, scenario, available=True): if available: self.browser.wait_visible(f"{self._scenario_selector(scenario)}:not([disabled])") else: self.browser.wait_visible(f"{self._scenario_selector(scenario)}:disabled") @log_step(snapshot_before=True) def check_scenario_selected(self, scenario): self.browser.wait_visible(f"{self._scenario_selector(scenario)}:checked") def check_scenario_index(self, scenario, index): self.browser.wait_visible(f".anaconda-screen-method-scenario:nth-child({index}) > {self._scenario_selector(scenario)}") @log_step(snapshot_before=True) def set_scenario(self, scenario): self.browser.click(self._scenario_selector(scenario)) self.browser.wait_visible(f"{self._scenario_selector(scenario)}:checked") self.browser.wait_visible(f"div[data-scenario='{scenario}']") class StorageReclaimDialog(): _TABLE = "#reclaim-space-modal-table" def __init__(self, browser): self.browser = browser def _reclaim_row_selector(self, device): """CSS selector for the table row identified by data-device attribute.""" return f'{self._TABLE} tr[data-device="{device}"]' def _reclaim_row_btn_selector(self, device, aria_label): return f'{self._reclaim_row_selector(device)} button[aria-label="{aria_label}"]' def reclaim_check_device_row(self, device, location=None, name=None, deviceType=None, space=None, locked=False): row = self._reclaim_row_selector(device) self.browser.wait_visible(row) if location is not None: self.browser.wait_in_text(f'{row} td[data-label="Location"]', location) if name is not None: self.browser.wait_in_text(f'{row} td[data-label="Name"]', name) if deviceType is not None: self.browser.wait_in_text(f'{row} td[data-label="Type"]', deviceType) if space is not None: self.browser.wait_in_text(f'{row} td[data-label="Space"]', space) if locked: self.browser.wait_visible(f'{row} .reclaim-space-modal-device-locked') def reclaim_remove_device(self, device): self.browser.click(self._reclaim_row_btn_selector(device, "delete")) def reclaim_check_action_button_present(self, device, action, present=True, disabled=False): sel = self._reclaim_row_btn_selector(device, action) if not present: self.browser.wait_not_present(sel) return self.browser.wait_visible(sel) if disabled: self.browser.wait_js_cond( f'document.querySelector({json.dumps(sel)})?.ariaDisabled === "true"' ) else: self.browser.wait_js_cond( f'document.querySelector({json.dumps(sel)})?.ariaDisabled !== "true"' ) def reclaim_modal_submit_and_check_warning(self, warning): self.browser.click("button:contains('Reclaim space')") self.browser.wait_in_text("#reclaim-space-modal .pf-v6-c-alert", warning) def reclaim_shrink_device(self, device, new_size, current_size=None, ariaLabel="shrink"): self.browser.click(self._reclaim_row_btn_selector(device, ariaLabel)) self.browser.wait_visible("#popover-reclaim-space-modal-shrink-body") if current_size is not None: self.browser.wait_val("#reclaim-space-modal-shrink-input", current_size) # HACK: there is some race here which steals the focus from the input and selects the page text instead for _ in range(3): self.browser.focus('#reclaim-space-modal-shrink-input') time.sleep(1) if self.browser.eval_js('document.activeElement == document.querySelector("#reclaim-space-modal-shrink-input")'): break self.browser.set_input_text("#reclaim-space-modal-shrink-input", new_size) self.browser.click("#reclaim-space-modal-shrink-button") self.browser.wait_not_present("#reclaim-space-modal-shrink-slider") self.reclaim_check_action_present(device, ariaLabel) def reclaim_check_action_present(self, device, action, present=True): row = self._reclaim_row_selector(device) if present: self.browser.wait_in_text(row, action) else: self.browser.wait_not_in_text(row, action) def reclaim_undo_action(self, device): self.browser.click(self._reclaim_row_btn_selector(device, "undo")) def reclaim_check_available_space(self, space): self.browser.wait_text("#reclaim-space-modal-hint-available-free-space", space) def reclaim_check_checkbox(self, value, isDisabled): self.browser.wait_visible("#reclaim-space-checkbox:checked" if value else "#reclaim-space-checkbox:not(:checked)") if isDisabled: self.browser.wait_visible("#reclaim-space-checkbox:disabled") else: self.browser.wait_visible("#reclaim-space-checkbox:not(:disabled)") def reclaim_set_checkbox(self, value): self.browser.set_checked("#reclaim-space-checkbox", value) def reclaim_modal_submit(self): self.browser.click("button.pf-m-warning") self.browser.wait_not_present("#reclaim-space-modal") def reclaim_modal_cancel(self): self.browser.wait_visible("#reclaim-space-modal") self.browser.click("#reclaim-space-modal button:contains('Cancel')") self.browser.wait_not_present("#reclaim-space-modal") class StorageMountPointMapping(StorageDBus, StorageDestination): def __init__(self, browser, machine): self.browser = browser self.machine = machine StorageDBus.__init__(self, machine) StorageDestination.__init__(self, browser, machine) def table_row(self, row): return f"#{CUSTOM_MOUNT_POINT}-table-row-{row}" def disks_loaded(self, disks): usable_disks = self.dbus_get_usable_disks() for disk in usable_disks: disks_dict = dict(disks) if disk not in disks_dict: return False return True def check_mountpoint_row(self, row, mountpoint=None, device=None, reformat=None, format_type=None): if mountpoint: constrained = ["/", "/boot", "/boot/efi", "swap"].count(mountpoint) self.check_mountpoint_row_mountpoint(row, mountpoint, constrained) if device: self.check_mountpoint_row_device(row, device) if reformat: self.check_mountpoint_row_reformat(row, reformat) if format_type: self.check_mountpoint_row_format_type(row, format_type) def select_disks(self, disks): self.browser.wait(lambda: self.disks_loaded(disks)) self.browser.click(f"#{INSTALLATION_METHOD}-change-destination-button") for (disk, selected) in disks: self.browser.set_checked(f"#{INSTALLATION_METHOD}-disk-selection-menu-item-{disk} input[type=checkbox]", selected) self.browser.click(f"#{INSTALLATION_METHOD}-change-destination-modal button:contains('Select')") for (disk, selected) in disks: self.check_disk_selected(disk, selected) def select_mountpoint(self, disks=None): if disks is not None: self.select_disks(disks) self.set_scenario("mount-point-mapping") self.browser.click("button:contains(Next)") self.browser.wait_js_cond(f'window.location.hash === "#/{CUSTOM_MOUNT_POINT}"') with self.browser.wait_timeout(30): self.browser.wait_visible(f"#{CUSTOM_MOUNT_POINT}-table") def select_mountpoint_row_mountpoint(self, row, mountpoint): self.browser.set_input_text(f"{self.table_row(row)} td[data-label='Mount point'] input", mountpoint) def select_mountpoint_row_device(self, row, device, device_id=None): toggle_selector = f"{self.table_row(row)}-device-select-toggle:not([disabled]):not([aria-disabled=true])" self.browser.click(toggle_selector) if device_id: item_selector = f".pf-v6-c-menu li[data-device-id='{device_id}']" else: item_selector = f".pf-v6-c-menu li[data-device-name='{device}']" self.browser.click(item_selector) self.browser.wait_not_present(".pf-v6-c-menu") self.browser.wait_in_text(f"{self.table_row(row)} .pf-v6-c-select__toggle-text", device) def toggle_mountpoint_row_device(self, row): self.browser.click(f"{self.table_row(row)}-device-select-toggle") def check_mountpoint_row_device(self, row, device): self.browser.wait_text(f"{self.table_row(row)} .pf-v6-c-select__toggle-text", device) def check_mountpoint_row_mountpoint(self, row, mountpoint, constrained=True): if constrained: self.browser.wait_text(f"{self.table_row(row)}-mountpoint", mountpoint) else: self.browser.wait_val(f"{self.table_row(row)}-mountpoint", mountpoint) def check_mountpoint_row_format_type(self, row, format_type): self.toggle_mountpoint_row_device(row) self.browser.wait_in_text(".pf-v6-c-menu__item.pf-m-selected", format_type) self.toggle_mountpoint_row_device(row) def check_mountpoint_row_device_available(self, row, device, available=True, disabled=False): disabled_selector = "[aria-disabled='true']" if disabled else ":not([aria-disabled='true'])" self.toggle_mountpoint_row_device(row) item_selector = f".pf-v6-c-menu li[data-device-name='{device}'] button{disabled_selector}" if available: self.browser.wait_visible(item_selector) else: self.browser.wait_not_present(f".pf-v6-c-menu li[data-device-name='{device}']") self.toggle_mountpoint_row_device(row) def select_mountpoint_row_reformat(self, row, selected=True): self.browser.set_checked(f"{self.table_row(row)} td[data-label='Reformat'] input", selected) def remove_mountpoint_row(self, row, initial_count): self.browser.wait_js_cond(f"ph_count('#{CUSTOM_MOUNT_POINT}-table tbody tr') == {initial_count}") self.browser.click(f"{self.table_row(row)} button[aria-label='Remove']") self.browser.wait_js_cond(f"ph_count('#{CUSTOM_MOUNT_POINT}-table tbody tr') == {initial_count - 1}") def check_mountpoint_row_reformat(self, row, checked): checked_selector = "input:checked" if checked else "input:not(:checked)" self.browser.wait_visible(f"{self.table_row(row)} td[data-label='Reformat'] {checked_selector}") def add_mountpoint_row(self): rows = self.browser.call_js_func("ph_count", f'#{CUSTOM_MOUNT_POINT}-table tbody tr') self.browser.click("button:contains('Add mount')") self.browser.wait_js_cond(f"ph_count('#{CUSTOM_MOUNT_POINT}-table tbody tr') == {rows + 1}") def wait_mountpoint_table_column_helper(self, row, column, text=None, present=True): if present: self.browser.wait_in_text(f"#{CUSTOM_MOUNT_POINT}-table-row-{row}-{column} .pf-v6-c-helper-text__item.pf-m-error", text) else: self.browser.wait_not_present(f"#{CUSTOM_MOUNT_POINT}-table-row-{row}-{column} .pf-v6-c-helper-text__item.pf-m-error") class Storage(StorageEncryption, StorageMountPointMapping, StorageScenario, StorageReclaimDialog, StorageUtils): def __init__(self, browser, machine): StorageEncryption.__init__(self, browser, machine) StorageMountPointMapping.__init__(self, browser, machine) StorageScenario.__init__(self, browser, machine) StorageReclaimDialog.__init__(self, browser) StorageUtils.__init__(self, browser, machine)