Storm Dragon
4da61465b0
Keep failed cursor trials and native recapture anchored to the original document, and restore the requested cached root before LRU eviction. Base braille retry resets on established connection time rather than stalled startup. Poll private AT-SPI daemon ownership without triggering service activation and clean up partial startup failures. Add focused regressions and update browser testing notes.
231 lines
11 KiB
Python
231 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Chromium AT-SPI proof on a private X server, bus, and browser profile."""
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
from native_test_session import run_private_session
|
|
|
|
|
|
BROWSER_PROOFS = {
|
|
"article": ("./internal/atspi", "^TestNativeBrowserArticleContext$", 30, 35, "browser-article.html"),
|
|
"emoji": ("./internal/app", "^TestNativeBrowserEmoji$", 35, 40, "browser-emoji.html"),
|
|
"structural-focus": ("./internal/app", "^TestNativeBrowserStructuralFocus$", 45, 50),
|
|
"disabled-fields": ("./internal/app", "^TestNativeBrowserDisabledFieldsRemainReadable$", 40, 70, "browser-disabled-fields.html"),
|
|
"live-atomic": ("./internal/atspi", "^TestNativeBrowserAtomicLiveRegion$", 30, 35, "browser-live.html"),
|
|
"replaced-feed": ("./internal/atspi", "^TestNativeBrowserMissingFeedRefreshesWholeBranch$", 30, 35, "browser-replaced-feed.html"),
|
|
"nested-document": ("./internal/app", "^TestNativeBrowserNestedDocumentHeadings$", 40, 45),
|
|
"reveal": ("./internal/app", "^TestNativeBrowserRevealedHeadings$", 60, 65),
|
|
"flat-chrome": ("./internal/review", "^TestNativeBrowserChromeFlatReview$", 35, 40),
|
|
"generatedlinks": ("./internal/app", "^TestNativeBrowserGeneratedLinkNavigation$", 30, 35),
|
|
"replacement": ("./internal/app", "^TestNativeBrowserInlineReplacementNavigation$", 30, 35),
|
|
"tooltips": ("./internal/app", "^TestNativeBrowserTooltipNavigation$", 30, 35),
|
|
"menu": ("./internal/app", "^TestNativeBrowserMenuControls$", 30, 35),
|
|
"lists": ("./internal/app", "^TestNativeBrowserLists$", 35, 40, "browser-lists.html"),
|
|
"pointer": ("./internal/atspi", "^TestNativePointerInput$", 15, 20),
|
|
"progress": ("./internal/app", "^TestNativeBrowserProgress$", 30, 35, "browser-progress.html"),
|
|
"fragments": ("./internal/review", "^TestNativeBrowserFragmentLines$", 30, 35, "browser-fragments.html"),
|
|
"positions": ("./internal/atspi", "^TestNativeBrowserSetPositions$", 25, 30, "browser-positions.html"),
|
|
"tabs": ("./internal/app", "^TestNativeBrowserTabs$", 30, 35),
|
|
"namedlinks": ("./internal/app", "^TestNativeBrowserNamedLinks$", 25, 30, "browser-named-links.html"),
|
|
"linkedheadings": ("./internal/app", "^TestNativeLinkedHeadings$", 45, 50, "browser-linked-headings.html"),
|
|
"catalog": ("./internal/atspi", "^TestNativeBrowserCatalogCapture$", 30, 35, "browser-catalog.html"),
|
|
"lines": ("./internal/atspi", "^TestNativeBrowserLogicalLines$", 25, 30, "browser-lines.html"),
|
|
"tables": ("./internal/app", "^TestNativeBrowserTables$", 35, 40),
|
|
"dynamic": ("./internal/app", "^TestNativeBrowserDynamicNavigation$", 40, 45),
|
|
"ticking": ("./internal/app", "^TestNativeBrowserTickingNavigation$", 40, 45),
|
|
"browser": ("./internal/atspi", "^TestNativeBrowser$", 45, 50),
|
|
"live": ("./internal/atspi", "^TestNativeBrowserLiveRegion$", 30, 35),
|
|
"document": ("./internal/app", "^TestNativeBrowserDocumentCommands$", 45, 50),
|
|
"editing": ("./internal/app", "^TestNativeBrowserEditing$", 45, 50),
|
|
"editor": ("./internal/app", "^TestNativeBrowserEditorTyping$", 30, 35, "browser-editor.html"),
|
|
"selection": ("./internal/app", "^TestNativeBrowserSelection$", 35, 40),
|
|
"modes": ("./internal/app", "^TestNativeBrowserModes$", 45, 50),
|
|
"keys": ("./internal/platform/xorg", "^TestNativeBrowseKeyTransitions$", 20, 25),
|
|
}
|
|
|
|
|
|
def selected_proofs():
|
|
selected = {name.strip() for name in os.environ.get(
|
|
"CTHULHU_NATIVE_BROWSER_TESTS", ",".join(BROWSER_PROOFS)).split(",") if name.strip()}
|
|
unknown = selected - BROWSER_PROOFS.keys()
|
|
if not selected or unknown:
|
|
raise ValueError("Select browser proofs from: " + ",".join(BROWSER_PROOFS) +
|
|
("; unknown: " + ",".join(sorted(unknown)) if unknown else ""))
|
|
return selected
|
|
|
|
|
|
def stop_atspi_bus(processes):
|
|
for process in reversed(processes):
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait(timeout=5)
|
|
|
|
|
|
def wait_for_bus_name(process, busArgs, name):
|
|
deadline = time.monotonic() + 10
|
|
while time.monotonic() < deadline:
|
|
if process.poll() is not None:
|
|
raise RuntimeError(name + " daemon exited before the probe")
|
|
# Query the bus itself: contacting the service can activate another
|
|
# daemon (or a user-manager unit) before our child owns its name.
|
|
result = subprocess.run(
|
|
["gdbus", "call", *busArgs, "--dest", "org.freedesktop.DBus",
|
|
"--object-path", "/org/freedesktop/DBus", "--method",
|
|
"org.freedesktop.DBus.NameHasOwner", name],
|
|
capture_output=True, text=True, timeout=2)
|
|
if result.returncode == 0 and result.stdout.strip() == "(true,)":
|
|
return
|
|
time.sleep(0.1)
|
|
raise RuntimeError(name + " did not become ready")
|
|
|
|
|
|
def start_atspi_bus():
|
|
# D-Bus activation of org.a11y.Bus and of the AT-SPI registry may
|
|
# delegate to systemd user units, which cannot work in this private
|
|
# session (disposable runtime dir, no user manager tied to the private
|
|
# bus) and must not be required by the test suite on non-systemd systems
|
|
# either. Launch both daemons directly; later activation requests then
|
|
# find the names already owned.
|
|
def locate(name, candidates):
|
|
found = shutil.which(name)
|
|
if found:
|
|
return found
|
|
for candidate in candidates:
|
|
if os.access(candidate, os.X_OK):
|
|
return candidate
|
|
raise RuntimeError(name + " not found")
|
|
|
|
launcher = locate("at-spi-bus-launcher",
|
|
("/usr/lib/at-spi-bus-launcher", "/usr/libexec/at-spi-bus-launcher"))
|
|
registry = locate("at-spi2-registryd",
|
|
("/usr/lib/at-spi2-registryd", "/usr/libexec/at-spi2-registryd"))
|
|
processes = []
|
|
try:
|
|
process = subprocess.Popen([launcher, "--launch-immediately"],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
processes.append(process)
|
|
wait_for_bus_name(process, ["--session"], "org.a11y.Bus")
|
|
result = subprocess.run(
|
|
["gdbus", "call", "--session", "--dest", "org.a11y.Bus",
|
|
"--object-path", "/org/a11y/bus", "--method", "org.a11y.Bus.GetAddress"],
|
|
capture_output=True, text=True, timeout=2, check=True)
|
|
address = result.stdout.split("'", 2)[1]
|
|
registryEnv = os.environ.copy()
|
|
registryEnv["AT_SPI_BUS_ADDRESS"] = address
|
|
registryd = subprocess.Popen([registry], env=registryEnv,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
processes.append(registryd)
|
|
wait_for_bus_name(registryd, ["--address", address], "org.a11y.atspi.Registry")
|
|
return processes
|
|
except BaseException:
|
|
stop_atspi_bus(processes)
|
|
raise
|
|
|
|
|
|
def run_inside():
|
|
browser = os.environ.get("CTHULHU_BROWSER") or shutil.which("brave")
|
|
if not browser:
|
|
raise RuntimeError("Set CTHULHU_BROWSER to a Chromium-family executable")
|
|
selected = selected_proofs()
|
|
atspi = start_atspi_bus()
|
|
try:
|
|
for name, proof in BROWSER_PROOFS.items():
|
|
if name in selected:
|
|
print(f"Running browser proof: {name}", flush=True)
|
|
# F7 toggles a profile-wide setting. Each proof must start from
|
|
# the same browser state whether run alone or after editing.
|
|
run_browser_proof(browser, proof)
|
|
finally:
|
|
stop_atspi_bus(atspi)
|
|
|
|
|
|
def run_browser_proof(browser, proof):
|
|
with tempfile.TemporaryDirectory(prefix="cthulhu-browser-profile-") as profile:
|
|
with tempfile.TemporaryFile() as log:
|
|
process = subprocess.Popen([
|
|
browser, "--user-data-dir=" + profile, "--no-first-run",
|
|
"--no-default-browser-check", "--disable-background-networking",
|
|
"--disable-component-update", "--disable-sync", "--password-store=basic",
|
|
"--force-renderer-accessibility", "--ozone-platform=x11",
|
|
"--new-window", Path(__file__).with_name(proof[4] if len(proof) > 4 else "browser.html").resolve().as_uri(),
|
|
], stdout=log, stderr=log)
|
|
try:
|
|
deadline = time.monotonic() + 20
|
|
while time.monotonic() < deadline:
|
|
if process.poll() is not None:
|
|
raise RuntimeError("Browser exited before the probe")
|
|
window_name = "Cthulhu Linked Headings Probe" if proof[1] == "^TestNativeLinkedHeadings$" else "Cthulhu Browser Probe"
|
|
result = subprocess.run(
|
|
["xdotool", "search", "--onlyvisible", "--name", window_name],
|
|
capture_output=True, text=True)
|
|
if result.stdout.strip():
|
|
window = result.stdout.splitlines()[0]
|
|
subprocess.run(["xdotool", "windowfocus", window], check=True)
|
|
break
|
|
time.sleep(0.1)
|
|
else:
|
|
raise RuntimeError("Browser fixture window missing")
|
|
environment = dict(os.environ, CTHULHU_NATIVE_BROWSER="1")
|
|
if proof[1] == "^TestNativeLinkedHeadings$":
|
|
environment["CTHULHU_NATIVE_LINKED_HEADINGS"] = "1"
|
|
environment.setdefault("GOCACHE", "/tmp/cthulhu-cache")
|
|
package, expression, testTimeout, runTimeout = proof[:4]
|
|
subprocess.run(["go", "test", package, "-run", expression,
|
|
"-count=1", "-v", f"-timeout={testTimeout}s"],
|
|
env=environment, check=True, timeout=runTimeout,
|
|
cwd=Path(__file__).resolve().parents[1])
|
|
finally:
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait(timeout=5)
|
|
|
|
|
|
def main():
|
|
selected = selected_proofs()
|
|
if len(sys.argv) == 2 and sys.argv[1] == "--inside":
|
|
run_inside()
|
|
return
|
|
with tempfile.TemporaryDirectory(prefix="cthulhu-browser-"):
|
|
read_fd, write_fd = os.pipe()
|
|
server = subprocess.Popen(
|
|
["Xvfb", "-displayfd", str(write_fd), "-screen", "0", "800x600x24", "-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()
|
|
if not display:
|
|
raise RuntimeError("Xvfb failed to start")
|
|
environment = dict(os.environ, DISPLAY=":" + display, XDG_SESSION_TYPE="x11")
|
|
environment.pop("DBUS_SESSION_BUS_ADDRESS", None)
|
|
environment.pop("AT_SPI_BUS_ADDRESS", None)
|
|
environment.pop("WAYLAND_DISPLAY", None)
|
|
run_private_session(
|
|
["dbus-run-session", "--", sys.executable, str(Path(__file__).resolve()), "--inside"],
|
|
env=environment,
|
|
check=True,
|
|
# Budget each isolated browser startup, bounded Go run and
|
|
# cleanup instead of imposing one 120-second suite limit.
|
|
timeout=15 + sum(30 + BROWSER_PROOFS[name][3] for name in selected),
|
|
)
|
|
finally:
|
|
server.terminate()
|
|
server.wait(timeout=3)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|