Add support for specialized & network disks and advanced partitioning message bar

anabot/runtime/installation/hub/partitioning/__init__.py

@@ -18,7 +18,7 @@
 from anabot.runtime.installation.common import done_handler
 
 # submodules
-from . import advanced, luks_dialog
+from . import advanced, luks_dialog, specialized_disks
 
 _local_path = '/installation/hub/partitioning'
 def handle_act(path, *args, **kwargs):
@@ -115,6 +115,29 @@ def disk_handler(element, app_node, local_node):
 def disk_check(element, app_node, local_node):
     return disk_manipulate(element, app_node, local_node, True)
 
+def add_specialized_disk_manipulate(element, app_node, local_node, dryrun):
+    try:
+        add_button = getnode(local_node, "push button", tr("Add Specialized Disk"))
+    except TimeoutError:
+        return NotFound("clickable button for adding specialized / network disks")
+    if not dryrun:
+        add_button.click()
+        try:
+            tab = getnode(app_node, "page tab", tr("Searc_h", context="GUI|Installation Destination|Filter"))
+            spec_disk_panel = getparent(tab, "panel")
+        except TimeoutError:
+            return NotFound("specialized disk page tab or panel")
+        default_handler(element, app_node, spec_disk_panel)
+    return Pass()
+
+@handle_act('/add_specialized_disk')
+def add_specialized_disk_handler(element, app_node, local_node):
+    add_specialized_disk_manipulate(element, app_node, local_node, False)
+
+@handle_chck('/add_specialized_disk')
+def add_specialized_disk_check(element, app_node, local_node):
+    return add_specialized_disk_manipulate(element, app_node, local_node, True)
+
 # RHEL-7
 def mode_manipulate(element, app_node, local_node, dryrun):
     mode = get_attr(element, "mode")

anabot/runtime/installation/hub/partitioning/advanced/__init__.py

@@ -447,3 +447,65 @@ def encrypt_data_handler(element, app_node, local_node):
 @handle_chck('/encrypt_data')
 def encrypt_data_check(element, app_node, local_node):
     return encrypt_data_manipulate(element, app_node, local_node, True)
+
+def infobar_manipulate(element, app_node, local_node, dryrun):
+    try:
+        infobar = getnode(app_node, "info bar")
+        infobar_label = getnode(infobar, "label")
+        infobar_message = infobar_label.text
+    except TimeoutError:
+        return NotFound("Info bar or info bar message")
+
+    if dryrun:
+        expected_message = get_attr(element, "message", "*")
+        translated_message = tr(expected_message)
+        if fnmatchcase(infobar_message, translated_message):
+            return Pass()
+        else:
+            return Fail("Message in info bar '%s' doesn't match the expected message '%s'." %
+                (infobar_message, translated_message))
+    default_handler(element, app_node, infobar_label)
+    # no action makes sense here (when dryrun == False), only a check
+    return Pass()
+
+@handle_act('/infobar')
+def infobar_handler(element, app_node, local_node):
+    return infobar_manipulate(element, app_node, local_node, False)
+
+@handle_chck('/infobar')
+def infobar_check(element, app_node, local_node):
+    return infobar_manipulate(element, app_node, local_node, True)
+
+def infobar_alert_manipulate(element, app_node, local_node, dryrun):
+    if not dryrun:
+        local_node.click()
+    try:
+        dialog = getnode(app_node, "alert")
+    except TimeoutError:
+        return Pass() if dryrun else NotFound("alert dialog")
+    if dryrun:
+        return Fail("Alert dialog is still visible.")
+    try:
+        alert_message = getnode(dialog, "label").text
+        close_button = getnode(dialog, "push button", tr("_Close"))
+    except TimeoutError:
+        return NotFound("alert label or close button")
+
+    expected_message = get_attr(element, "message", "*")
+    expected_message = tr(expected_message)
+    if fnmatchcase(alert_message, expected_message):
+        result = Pass()
+    else:
+        result = Fail("Alert message '%s' doesn't match expected message '%s'.")
+
+    close_button.click()
+    return result
+
+@handle_act('/infobar/alert')
+def infobar_alert_handler(element, app_node, local_node):
+    return infobar_alert_manipulate(element, app_node, local_node, False)
+
+@handle_chck('/infobar/alert')
+@check_action_result
+def infobar_alert_check(element, app_node, local_node):
+    return infobar_alert_manipulate(element, app_node, local_node, True)

anabot/runtime/installation/hub/partitioning/specialized_disks/__init__.py

@@ -0,0 +1,72 @@
+# -*- coding: utf-8 -*-
+
+import logging
+logger = logging.getLogger('anabot')
+
+from fnmatch import fnmatchcase
+
+from anabot.runtime.decorators import handle_action, handle_check, check_action_result
+from anabot.runtime.functions import get_attr, getnode, getnodes, getsibling
+from anabot.runtime.errors import TimeoutError
+from anabot.runtime.translate import tr
+from anabot.runtime.actionresult import NotFoundResult as NotFound, ActionResultPass as Pass, ActionResultFail as Fail
+
+_local_path = '/installation/hub/partitioning/add_specialized_disk'
+def handle_act(path, *args, **kwargs):
+    return handle_action(_local_path + path, *args, **kwargs)
+
+def handle_chck(path, *args, **kwargs):
+    return handle_check(_local_path + path, *args, **kwargs)
+
+def select_manipulate(element, app_node, local_node, dryrun):
+    name = get_attr(element, "names", "*")
+    action = get_attr(element, "action", "select")
+
+    try:
+        results_label = getnode(app_node, "label", tr("Search Res_ults:", context="GUI|Installation Destination|Filter|Search"))
+        table_pane = getsibling(results_label, 1, "scroll pane")
+        disks_table = getnode(table_pane, "table")
+    except TimeoutError:
+        return NotFound("results label, table pane or disks table")
+    cells = getnodes(disks_table, "table cell")
+    column_count = len(getnodes(disks_table, "table column header"))
+    # every table row has checkbox on position 0 and device name on 1
+    name_cells = [
+        c
+        for c in cells[1::column_count]
+        if fnmatchcase(c.name, name)
+    ]
+    checkbox_cells = [
+        getsibling(c, -1, "table cell")
+        for c in name_cells
+    ]
+    logger.info("Devices that match name '%s' found: '%s'" % (name, [n.name for n in name_cells]))
+    for c in checkbox_cells:
+        if (action == "select") != (c.checked):
+            if dryrun:
+                return Fail("Checkbox status for device '%s' doesn't match required action '%s'" % (c.name, action))
+            logger.info("Clicking on checkbox for device %s" % c.name)
+            c.click()
+    return Pass()
+
+@handle_act('/select')
+def select_handler(element, app_node, local_node):
+    return select_manipulate(element, app_node, local_node, False)
+
+@handle_chck('/select')
+def select_check(element, app_node, local_node):
+    return select_manipulate(element, app_node, local_node, True)
+
+@handle_act('/done')
+def done_handler(element, app_node, local_node):
+    try:
+        done_button = getnode(app_node, "push button", tr("_Done", drop_underscore=False, context="GUI|Spoke Navigation"))
+    except TimeoutError:
+        return NotFound("clickable Done button")
+    done_button.click()
+    return Pass()
+
+@handle_chck('/done')
+@check_action_result
+def done_check(element, app_node, local_node):
+    return Pass()