from collections import UserDict
import steps
from step_logger import log_step
from storage import StorageEncryption
from users import create_user
class InstallerSteps(UserDict):
ACCOUNTS = steps.ACCOUNTS
DATE_TIME = steps.DATE_TIME
CUSTOM_MOUNT_POINT = steps.CUSTOM_MOUNT_POINT
INSTALLATION_METHOD = steps.INSTALLATION_METHOD
NETWORK = steps.NETWORK
SOFTWARE_SELECTION = steps.SOFTWARE_SELECTION
LANGUAGE = steps.LANGUAGE
PROGRESS = steps.PROGRESS
REVIEW = steps.REVIEW
STORAGE_CONFIGURATION = steps.STORAGE_CONFIGURATION
def __init__(self, hidden_steps=None, scenario=None, machine=None):
super().__init__()
if (scenario == "mount-point-mapping"):
self.STORAGE_CONFIGURATION = "anaconda-screen-storage-configuration-manual"
ACCOUNTS = self.ACCOUNTS
CUSTOM_MOUNT_POINT = self.CUSTOM_MOUNT_POINT
DATE_TIME = self.DATE_TIME
INSTALLATION_METHOD = self.INSTALLATION_METHOD
NETWORK = self.NETWORK
SOFTWARE_SELECTION = self.SOFTWARE_SELECTION
LANGUAGE = self.LANGUAGE
PROGRESS = self.PROGRESS
REVIEW = self.REVIEW
STORAGE_CONFIGURATION = self.STORAGE_CONFIGURATION
_steps_jump = {
LANGUAGE: [NETWORK],
NETWORK: [DATE_TIME],
DATE_TIME: [SOFTWARE_SELECTION],
SOFTWARE_SELECTION: [INSTALLATION_METHOD],
STORAGE_CONFIGURATION: [ACCOUNTS],
CUSTOM_MOUNT_POINT: [ACCOUNTS],
ACCOUNTS: [REVIEW],
REVIEW: [PROGRESS],
PROGRESS: [],
}
_hidden_steps = hidden_steps or []
payload_type = getattr(machine, "payload_type", "liveimg") if machine is not None else "liveimg"
if payload_type != "dnf":
_steps_jump[DATE_TIME] = INSTALLATION_METHOD
_hidden_steps.append(SOFTWARE_SELECTION)
if scenario in ['use-configured-storage', 'use-configured-storage-kickstart', 'home-reuse']:
_steps_jump[INSTALLATION_METHOD] = [ACCOUNTS]
_hidden_steps.extend([CUSTOM_MOUNT_POINT, STORAGE_CONFIGURATION])
if scenario != 'use-configured-storage':
del _steps_jump[STORAGE_CONFIGURATION]
del _steps_jump[CUSTOM_MOUNT_POINT]
else:
_steps_jump[INSTALLATION_METHOD] = [STORAGE_CONFIGURATION, CUSTOM_MOUNT_POINT]
self._steps_jump = _steps_jump
self.hidden_steps = _hidden_steps
_parent_steps = {CUSTOM_MOUNT_POINT: STORAGE_CONFIGURATION}
self._parent_steps = _parent_steps
_steps_callbacks = {
ACCOUNTS: create_user,
STORAGE_CONFIGURATION: lambda browser, machine: StorageEncryption(
browser, machine
).set_encryption_selected(False),
}
self._steps_callbacks = _steps_callbacks
class Installer():
def __init__(self, browser, machine, hidden_steps=None, scenario=None):
self.browser = browser
self.machine = machine
self.steps = InstallerSteps(hidden_steps, scenario, machine)
@log_step(snapshot_before=True)
def begin_installation(self, should_fail=False, needs_confirmation=True, button_text='Erase data and install'):
current_page = self.get_current_page()
if needs_confirmation:
self.browser.wait_visible("#installation-next-btn[aria-disabled=true]")
self.browser.click(f"#{self.steps.REVIEW}-next-confirmation-checkbox")
self.browser.wait_visible("#installation-next-btn:not([aria-disabled=true])")
self.browser.wait_text("#installation-next-btn", button_text)
self.browser.click("#installation-next-btn")
if should_fail:
self.wait_current_page(current_page)
else:
self.wait_current_page(self.steps._steps_jump[current_page][0])
def _previous_pages(self, page):
return [k for k, v in self.steps._steps_jump.items() if page in v]
def reach(self, target_page):
path = []
prev_pages = [target_page]
current_page = self.get_current_page()
while current_page not in prev_pages:
page = prev_pages[0]
path.append(page)
prev_pages = self._previous_pages(page)
while self.get_current_page() != target_page:
next_page = path.pop()
if next_page not in self.steps.hidden_steps:
self.next(next_page=next_page)
if next_page in self.steps._steps_callbacks and next_page != target_page:
self.steps._steps_callbacks[next_page](self.browser, self.machine)
@log_step()
def next(self, should_fail=False, next_page=""):
current_page = self.get_current_page()
if not next_page:
next_page = self.steps._steps_jump[current_page][0]
while next_page in self.steps.hidden_steps:
next_page = self.steps._steps_jump[next_page][0]
self.browser.click("#installation-next-btn")
expected_page = current_page if should_fail else next_page
self.wait_current_page(expected_page)
if should_fail:
self.browser.wait_visible(f"#{expected_page}-step-notification")
return expected_page
@log_step()
def check_next_disabled(self, disabled=True):
"""Check if the Next button is disabled.
:param disabled: True if Next button should be disabled, False if not
:type disabled: bool, optional
"""
if disabled:
self.browser.wait_visible("#installation-next-btn[aria-disabled=true]")
else:
self.browser.wait_visible("#installation-next-btn:not([aria-disabled=true])")
def check_sidebar_step_disabled(self, step, disabled=True):
"""Check if a sidebar step is disabled.
:param disabled: True if the sidebar step should be disabled, False if not
:type disabled: bool, optional
"""
value = "true" if disabled else "false"
self.browser.wait_visible(f"#{step}[aria-disabled={value}]")
@log_step(snapshot_before=True)
def back(self, should_fail=False, previous_page=""):
current_page = self.get_current_page()
self.browser.click("button:contains(Back)")
if should_fail:
self.wait_current_page(current_page)
else:
if not previous_page:
previous_page = self._previous_pages(current_page)[0]
while previous_page in self.steps.hidden_steps:
previous_page = self._previous_pages(previous_page)[0]
self.wait_current_page(previous_page)
@log_step()
def open(self, step=None):
step = step or self.steps.LANGUAGE
while step in self.steps.hidden_steps:
step = self.steps._steps_jump[step][0]
self.browser.open(f"/cockpit/@localhost/anaconda-webui/index.html#/{step}")
with self.browser.wait_timeout(60):
self.wait_current_page(step)
def click_step_on_sidebar(self, step=None):
step = step or self.get_current_page()
self.browser.wait_visible(f"#{step}:not([disabled])")
self.browser.click(f"#{step}")
@log_step()
def reach_on_sidebar(self, target_page):
if target_page in self.steps._parent_steps:
self.click_step_on_sidebar(self.steps._parent_steps[target_page])
self.click_step_on_sidebar(target_page)
self.wait_current_page(target_page)
def get_current_page(self):
return self.browser.eval_js('window.location.hash;').replace('#/', '') or self.steps[0]
@log_step(snapshot_after=True)
def wait_current_page(self, page):
self.browser.wait_not_present("#disk-encryption-next-spinner")
self.browser.wait_js_cond(f'window.location.hash === "#/{page}"')
if page == self.steps.PROGRESS:
self.browser.wait_visible(".pf-v6-c-progress-stepper")
else:
self.browser.wait_visible(f"#{page}.pf-m-current")
@log_step(snapshot_after=True)
def check_prerelease_info(self, is_expected=None):
""" Checks whether the pre-release information is visible or not.
If is_expected is not set, the expected state is deduced from .buildstamp file.
:param is_expected: Is it expected that the info is visible or not, defaults to None
:type is_expected: bool, optional
"""
if is_expected is not None:
value = str(is_expected)
else:
value = self.machine.execute("grep IsFinal= /.buildstamp").split("=", 1)[1]
if value.lower() == "false":
self.browser.wait_visible("#betanag-icon")
else:
self.browser.wait_not_present("#betang-icon")
