cthulhu/tests/native_tentacles.py

# Copyright (c) 2026 Storm Dragon
# Authors: Storm Dragon, Angus Kola
# SPDX-License-Identifier: GPL-3.0-or-later
"""Tentacle management action routing with fixture responses, no downloads."""
import select

from native_x11 import wait_for


def check_page(pyatspi, process, window, nodes, named, walk, focus, key,
               request, send, initial, application):
    tab = next(node for node in nodes
               if node.name == "Tentacles" and node.getRole() == pyatspi.ROLE_PAGE_TAB)
    assert tab.parent.querySelection().selectChild(tab.getIndexInParent())
    origin = "https://example.org/tentacles"
    repository = {"URL": origin, "ID": "example.source", "Name": "Example",
                  "Tentacles": [{"ID": "echo", "Name": "Example package", "Version": "1.0",
                                 "Capabilities": ["presentation", "output-observe", "keyboard-inject"], "Releases": [{}]}]}
    snapshot = {"Entries": [], "Repositories": [repository]}
    table = next(node for node in walk(window)
                 if node.name == "Tentacles" and node.getRole() == pyatspi.ROLE_TABLE)

    def press(name):
        button = named(name)
        wait_for(lambda: button.getState().contains(pyatspi.STATE_SENSITIVE), name + " disabled")
        focus(button)
        key("space")

    status_node = named(initial["Message"])

    def accept(action, package_id="", url=""):
        value = request()["Tentacle"]
        assert value == {"Action": action, "ID": package_id, "URL": url, "Entry": None}, value
        message = "Accepted " + action
        if action == "settings":
            # Successful settings replies intentionally leave the status alone.
            # Observe a changed snapshot cell as the response barrier instead.
            snapshot["Entries"][0]["Message"] = message
        send(dict(initial, Tentacles=snapshot, Message=message))
        if action == "settings":
            wait_for(lambda: any(node.name == message for node in walk(table)),
                     "settings response was not rendered")
        else:
            wait_for(lambda: status_node.name == message, action + " response missing")
        wait_for(lambda: named("Refresh status").getState().contains(pyatspi.STATE_SENSITIVE),
                 action + " left preferences busy")

    url_entry = next(node for node in walk(window)
                     if node.name == "Repository URL" and
                     node.getState().contains(pyatspi.STATE_EDITABLE))
    url_entry.queryEditableText().setTextContents(origin)
    press("Add repository")
    accept("repository-add", url=origin)
    wait_for(lambda: table.queryTable().nRows == 1, "catalog row missing")
    focus(table)
    key("Home")

    def confirmation():
        for app in pyatspi.Registry.getDesktop(0):
            for node in app:
                if node and node.getRole() in (pyatspi.ROLE_DIALOG, pyatspi.ROLE_ALERT):
                    if any("Only install packages" in child.name for child in walk(node)):
                        return node
        return None

    def answer(label):
        dialog = wait_for(confirmation, "install confirmation missing")
        message = " ".join(node.name for node in walk(dialog))
        assert "Example package version 1.0" in message, message
        assert "send speech and braille messages" in message, message
        assert "share local reader output with the remote controller" in message, message
        assert "accept keyboard input from the remote controller" in message, message
        assert "Each session requires local approval" in message, message
        button = next(node for node in walk(dialog)
                      if node.name == label and node.getRole() == pyatspi.ROLE_PUSH_BUTTON)
        focus(button)
        key("space")
        wait_for(lambda: not confirmation(), "confirmation did not close")

    press("Install")
    answer("Cancel")
    assert not select.select([process.stdout], [], [], 0.2)[0], "Cancel submitted installation"
    press("Install")
    answer("Install")
    entry = {"ID": "installed-echo", "LocalID": "echo", "Name": "Example package",
             "Version": "1.0", "Repository": origin, "RepositoryID": "example.source",
             "Presentation": True, "Enabled": False}
    status = {"Entry": entry, "State": "disabled"}
    snapshot["Entries"] = [status]
    accept("install", "echo", origin)
    focus(table)
    key("Home", "space")
    entry["Enabled"], status["State"] = True, "active"
    accept("enable", entry["ID"])
    press("Retry")
    accept("retry", entry["ID"])
    press("Settings")
    accept("settings", entry["ID"])
    focus(table)
    key("space")
    entry["Enabled"], status["State"] = False, "disabled"
    accept("disable", entry["ID"])
    press("Remove tentacle")
    snapshot["Entries"] = []
    accept("remove", entry["ID"])
    press("Refresh status")
    accept("list")
    repositories = named("Tentacle repositories")
    assert repositories.querySelection().selectChild(1), "repository tab missing"
    press("Refresh repository")
    accept("repository-refresh", url=origin)
    press("Remove repository")
    snapshot["Repositories"] = []
    accept("repository-remove", url=origin)
    wait_for(lambda: table.queryTable().nRows == 0, "removed repository left rows")
    press("Cancel")
    assert process.wait(timeout=5) == 0
    print("PASS: repository actions, install confirmation/Cancel, enable/disable, retry, settings, remove, refresh")