#!/usr/bin/env python3
"""Private Xvfb/AT-SPI proof for the speech history and monitor 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"))

    def walk(node):
        yield node
        for child in node:
            if child is not None:
                yield from walk(child)

    def window_named(title):
        for application in pyatspi.Registry.getDesktop(0):
            for node in application:
                if node and node.name == title:
                    return node
        return None

    def open_helper(mode, items=None):
        process = subprocess.Popen(
            [binary, "--speech-history-ui"], stdin=subprocess.PIPE,
            stdout=subprocess.PIPE, text=True,
        )
        send(process, {"mode": mode, "items": items or []})
        return process

    def send(process, value):
        process.stdin.write(json.dumps(value) + "\n")
        process.stdin.flush()

    def request(process):
        assert select.select([process.stdout], [], [], 5)[0], "no speech history response"
        return json.loads(process.stdout.readline())

    def click(node):
        assert node.queryComponent().grabFocus(), ("cannot focus", node.name)
        subprocess.run(["xdotool", "key", "--clearmodifiers", "space"], check=True)

    history = open_helper("history", ["Newest speech", "Older speech", "Another line"])
    try:
        window = wait_for(lambda: window_named("Speech History - Cthulhu"), "history window missing")
        controls = {node.name: node for node in walk(window) if node.name}
        for name in ("Filter:", "Newest speech", "Older speech", "Another line",
                     "Copy to clipboard", "Remove from history", "Close"):
            assert name in controls, ("missing history control", name, list(controls))
        table = next(node for node in walk(window) if node.getRole() == pyatspi.ROLE_TABLE)
        assert table.queryComponent().grabFocus(), "cannot focus history table"
        subprocess.run(["xdotool", "key", "--clearmodifiers", "Down"], check=True)
        wait_for(lambda: table.querySelection().getSelectedChild(0).name == "Older speech",
                 "Down did not select the next history entry")
        subprocess.run(["xdotool", "key", "--clearmodifiers", "Up"], check=True)
        wait_for(lambda: table.querySelection().getSelectedChild(0).name == "Newest speech",
                 "Up did not restore the first history entry")
        subprocess.run(["xdotool", "key", "--clearmodifiers", "o"], check=True)
        wait_for(lambda: table.queryTable().nRows == 1, "typing did not filter by prefix")
        subprocess.run(["xdotool", "key", "--clearmodifiers", "BackSpace"], check=True)
        wait_for(lambda: table.queryTable().nRows == 3, "backspace did not clear the filter")
        assert table.queryComponent().grabFocus(), "cannot refocus history table"
        click(controls["Copy to clipboard"])
        assert request(history) == {"action": "copy"}
        copied = subprocess.check_output(["xclip", "-selection", "clipboard", "-o"], text=True)
        assert copied == "Newest speech", copied
        click(controls["Remove from history"])
        removed = request(history)
        assert removed == {"action": "remove", "selection": "Newest speech"}, removed
        send(history, {"items": ["Older speech", "Another line"]})
        wait_for(lambda: table.queryTable().nRows == 2, "removal did not refresh the list")
        click(controls["Remove from history"])
        assert request(history)["selection"] == "Older speech"
        send(history, {"items": ["Another line"]})
        wait_for(lambda: table.queryTable().nRows == 1, "second removal did not refresh the list")
        click(controls["Remove from history"])
        assert request(history)["selection"] == "Another line"
        send(history, {"items": []})
        wait_for(lambda: table.queryTable().nRows == 0, "removing the final item left a row")
        click(controls["Close"])
        assert history.wait(timeout=5) == 0
    finally:
        if history.poll() is None:
            history.kill()
        history.wait(timeout=5)

    monitor = open_helper("monitor")
    try:
        window = wait_for(lambda: window_named("Speech Monitor - Cthulhu"), "monitor window missing")
        def monitor_text():
            for node in walk(window):
                if node.name == "Speech monitor output":
                    return node.queryText().getText(0, -1)
            return ""
        send(monitor, {"append": "Live utterance"})
        wait_for(lambda: "Live utterance" in monitor_text(), "monitor text missing")
        controls = {node.name: node for node in walk(window) if node.name}
        click(controls["Clear"])
        wait_for(lambda: "Live utterance" not in monitor_text(), "monitor text was not cleared")
        click(controls["Close"])
        assert monitor.wait(timeout=5) == 0
    finally:
        if monitor.poll() is None:
            monitor.kill()
        monitor.wait(timeout=5)
    print("PASS: speech history accessibility, copy, remove, and live monitor")


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()