Reset master to testing content

tests/native_notifications.py

@@ -0,0 +1,175 @@
+#!/usr/bin/env python3
+"""Private Xvfb/AT-SPI proof for the Go notification-list helper."""
+import json
+import os
+from pathlib import Path
+import select
+import subprocess
+import sys
+
+sys.dont_write_bytecode = True
+from native_x11 import wait_for
+from native_test_session import run_private_session
+
+
+def inside():
+    import pyatspi
+
+    binary = os.environ.get("CTHULHU_BINARY", str(Path(__file__).resolve().parents[1] / "cthulhu"))
+    process = subprocess.Popen(
+        [binary, "--notification-list-ui"], stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE, text=True,
+    )
+
+    def send(value):
+        process.stdin.write(json.dumps(value) + "\n")
+        process.stdin.flush()
+
+    def request():
+        assert select.select([process.stdout], [], [], 5)[0], "no notification-list request"
+        return json.loads(process.stdout.readline())
+
+    def walk(node):
+        yield node
+        for child in node:
+            if child is not None:
+                yield from walk(child)
+
+    def find_window():
+        for application in pyatspi.Registry.getDesktop(0):
+            for node in application:
+                if node and node.name == "2 notifications":
+                    return node
+        return None
+
+    def press(node):
+        assert node.queryComponent().grabFocus(), ("cannot focus", node.name)
+        subprocess.run(["xdotool", "key", "--clearmodifiers", "space"], check=True)
+
+    def refresh_controls():
+        controls.clear()
+        for node in walk(window):
+            if node.name:
+                controls.setdefault(node.name, node)
+
+    def entry(identifier, message, received, can_control=False, actions=None):
+        return {"id": identifier, "message": message, "received": received,
+                "canControl": can_control, "actions": actions or []}
+
+    initial = {
+        "title": "2 notifications",
+        "selected": "1",
+        "entries": [
+            entry("1", "Live message", "0 seconds ago", True, [
+                {"key": "default", "label": "View"},
+                {"key": "reply", "label": "Reply"},
+            ]),
+            entry("0", "Older message", "1 minute ago"),
+        ],
+    }
+    try:
+        send(initial)
+        window = wait_for(find_window, "notification accessible window missing")
+        nodes = list(walk(window))
+        controls = {}
+        for node in nodes:
+            if node.name:
+                controls.setdefault(node.name, node)
+
+        table = next(node for node in nodes if node.getRole() == pyatspi.ROLE_TABLE)
+        assert table.name == "2 notifications", table.name
+        names = {node.name for node in nodes}
+        for name in ("Live message", "Older message", "0 seconds ago", "1 minute ago",
+                      "Copy", "Dismiss Notification", "Clear", "Close", "View", "Reply"):
+            assert name in names, ("missing accessible name", name)
+        interactive_roles = {
+            pyatspi.ROLE_PUSH_BUTTON, pyatspi.ROLE_TABLE, pyatspi.ROLE_TABLE_CELL,
+        }
+        unnamed = sorted(node.getRoleName() for node in nodes
+                         if node.getRole() in interactive_roles and not node.name)
+        assert not unnamed, ("unnamed interactive controls", unnamed)
+        assert "Dismiss Notification" in controls
+        assert controls["Dismiss Notification"].getState().contains(pyatspi.STATE_SENSITIVE)
+
+        xid = subprocess.check_output(
+            ["xdotool", "search", "--onlyvisible", "--name", "^2 notifications$"], text=True,
+        ).splitlines()[0]
+        subprocess.run(["xdotool", "windowfocus", xid], check=True)
+        assert table.queryComponent().grabFocus(), "cannot focus notification table"
+        subprocess.run(["xdotool", "key", "--clearmodifiers", "Down"], check=True)
+        selected = request()
+        assert selected == {"selection": "0", "row": 1}, selected
+        send(dict(initial, selected="0"))
+
+        wait_for(lambda: "Notification is no longer available" in {
+            node.name for node in walk(window)
+        }, "unavailable action status missing")
+        assert not controls["Dismiss Notification"].getState().contains(pyatspi.STATE_SENSITIVE)
+        assert "View" not in {node.name for node in walk(window)}
+        assert "Reply" not in {node.name for node in walk(window)}
+
+        send(dict(initial, selected="1"))
+        wait_for(lambda: controls["Dismiss Notification"].getState().contains(pyatspi.STATE_SENSITIVE),
+                 "live selection did not restore controls")
+        refresh_controls()
+        press(controls["Reply"])
+        invoked = request()
+        assert invoked == {"action": "reply", "selection": "1", "row": 0}, invoked
+        send(dict(initial, selected="1"))
+        refresh_controls()
+
+        press(controls["Copy"])
+        copied = request()
+        assert copied == {"copy": True}, copied
+
+        press(controls["Dismiss Notification"])
+        dismissed = request()
+        assert dismissed == {"dismiss": True, "selection": "1", "row": 0}, dismissed
+        send({"title": "1 notification", "selected": "0",
+              "entries": [entry("0", "Older message", "1 minute ago")]})
+        refresh_controls()
+
+        press(controls["Clear"])
+        cleared = request()
+        assert cleared == {"clear": True}, cleared
+        send({"close": True})
+        assert process.wait(timeout=5) == 0
+        assert process.stdout.read() == "", "clear submitted extra requests"
+        print("PASS: notification list accessibility, selection refresh, actions, copy, dismiss, clear")
+    finally:
+        if process.poll() is None:
+            process.kill()
+        process.wait(timeout=5)
+
+
+def main():
+    if "--inside" in sys.argv:
+        inside()
+        return
+    read_fd, write_fd = os.pipe()
+    server = subprocess.Popen(
+        ["Xvfb", "-displayfd", str(write_fd), "-screen", "0", "1024x768x24", "-nolisten", "tcp"],
+        pass_fds=(write_fd,), stderr=subprocess.DEVNULL,
+    )
+    os.close(write_fd)
+    try:
+        with os.fdopen(read_fd) as reader:
+            display = reader.readline().strip()
+        assert display, "Xvfb failed"
+        environment = dict(
+            os.environ, DISPLAY=":" + display, GDK_BACKEND="x11",
+            GTK_MODULES="gail:atk-bridge", GIO_USE_VFS="local",
+        )
+        for key in ("DBUS_SESSION_BUS_ADDRESS", "AT_SPI_BUS_ADDRESS", "WAYLAND_DISPLAY"):
+            environment.pop(key, None)
+        run_private_session(
+            ["dbus-run-session", "--", sys.executable, str(Path(__file__).resolve()), "--inside"],
+            env=environment, check=True, timeout=60,
+        )
+    finally:
+        server.terminate()
+        server.wait(timeout=5)
+
+
+if __name__ == "__main__":
+    main()