116 lines
5.1 KiB
Python
116 lines
5.1 KiB
Python
# 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"], "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
|
|
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")
|