cthulhu/tests/native_notifications.py

#!/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


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)
        subprocess.run(
            ["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()