Storm Dragon
89686fa029
Bring the fragmented-line browser fixture into agreement with current browse navigation. Rich editors are read through their text descendants, so Down from the preceding heading must read "Review the code." directly. Remove the stale empty editor-control stop while retaining forward and reverse line navigation, character and word movement, Read All, and inline Braille-routing assertions. Fix Firefox's missing announcement when creating a second tab. Firefox can emit selection events before the new tab enters its accessible child list, causing the native selected-child lookup to return nothing. Use the subsequent child-added event only when its target is a live, freshly selected browser-chrome tab with the emitting tab list as its direct parent. Keep document-authored tabs and unselected background additions excluded, retain foreground filtering and tab-identity deduplication, and release native references on the GLib owner. Normalize child-added origin suffixes such as :system before event dispatch, preserving the original native type in diagnostics. Rename the shared mutation normalizer to reflect its expanded scope and update its callers, including the Pidgin insertion check. Extend the normalization regression to cover child additions and reject unrelated event names. Firefox keeps its normal animations and all existing tab-announcement and address-editing assertions. Make private-session teardown tolerate the document portal's asynchronous FUSE unmount after D-Bus exits. Retry only ENOTCONN and EBUSY cleanup failures for at most five seconds. Keep unrelated errors and persistent mount failures visible. Add regressions for transient teardown, preservation of a failing child's exit status, unrelated cleanup errors, and the cleanup deadline. Runtime directories and accessibility buses remain private, with no init-system dependency. Update browser verification notes and the parity ledger with the fixes, evidence, regression risks, and concrete live acceptance procedures. Validation: - All 33 Chromium browser proofs passed in the final combined run. - Firefox baseline and tab checks passed repeatedly with normal animations. - All eight private-session isolation and cleanup regressions passed. - AT-SPI, app, and review package race suites passed during development; focused race regressions passed again after the final event-normalization change. - Go vet passed for internal/atspi, internal/app, and internal/review. - A separate reader binary built successfully. - Parity-ledger validation passed for 75 rows; git diff --check passed.
173 lines
8.6 KiB
Python
173 lines
8.6 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright (c) 2026 Storm Dragon
|
|
# Authors: Storm Dragon, Angus Kola
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Runtime isolation and real AT-SPI socket collision regressions."""
|
|
import ast
|
|
import errno
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from native_test_session import run_private_session
|
|
|
|
|
|
class PrivateSessionTests(unittest.TestCase):
|
|
def test_child_runtime_and_bus_are_private(self):
|
|
environment = dict(os.environ, XDG_RUNTIME_DIR="/invalid/desktop-runtime",
|
|
DBUS_SESSION_BUS_ADDRESS="unix:path=/invalid/desktop-bus",
|
|
AT_SPI_BUS_ADDRESS="unix:path=/invalid/accessibility-bus",
|
|
DBUS_STARTER_ADDRESS="unix:path=/invalid/starter-bus",
|
|
DBUS_STARTER_BUS_TYPE="session", WAYLAND_DISPLAY="desktop-wayland")
|
|
originalEnv = environment.copy()
|
|
script = """import json, os, socket, stat
|
|
from pathlib import Path
|
|
runtime = Path(os.environ['XDG_RUNTIME_DIR'])
|
|
listener = socket.socket(socket.AF_UNIX)
|
|
listener.bind(str(runtime / 'probe.sock'))
|
|
print(json.dumps(dict(runtime=str(runtime), mode=stat.S_IMODE(runtime.stat().st_mode),
|
|
bus=os.environ['DBUS_SESSION_BUS_ADDRESS'],
|
|
inherited=[name for name in ('AT_SPI_BUS_ADDRESS', 'DBUS_STARTER_ADDRESS',
|
|
'DBUS_STARTER_BUS_TYPE', 'WAYLAND_DISPLAY') if name in os.environ])))
|
|
"""
|
|
result = run_private_session(
|
|
["dbus-run-session", "--", sys.executable, "-c", script], env=environment,
|
|
capture_output=True, text=True, check=True, timeout=15)
|
|
state = json.loads(result.stdout)
|
|
self.assertEqual(state["mode"], 0o700)
|
|
self.assertEqual(state["inherited"], [])
|
|
self.assertNotEqual(state["bus"], originalEnv["DBUS_SESSION_BUS_ADDRESS"])
|
|
self.assertFalse(Path(state["runtime"]).exists(), "child runtime was not cleaned up")
|
|
self.assertEqual(environment, originalEnv)
|
|
|
|
def test_cleanup_after_child_failure(self):
|
|
script = "import os, sys; print(os.environ['XDG_RUNTIME_DIR'], flush=True); sys.exit(7)"
|
|
with self.assertRaises(subprocess.CalledProcessError) as caught:
|
|
run_private_session(
|
|
["dbus-run-session", "--", sys.executable, "-c", script], env=os.environ,
|
|
capture_output=True, text=True, check=True, timeout=15)
|
|
self.assertEqual(caught.exception.returncode, 7)
|
|
self.assertFalse(Path(caught.exception.stdout.strip()).exists())
|
|
|
|
def test_portal_teardown_retry_preserves_child_failure(self):
|
|
originalCleanup = tempfile.TemporaryDirectory.cleanup
|
|
attempts = []
|
|
|
|
def cleanup(runtime):
|
|
attempts.append(runtime.name)
|
|
if len(attempts) <= 2:
|
|
errorCode = (errno.ENOTCONN, errno.EBUSY)[len(attempts) - 1]
|
|
raise OSError(errorCode, "Document portal is shutting down")
|
|
originalCleanup(runtime)
|
|
|
|
with patch("native_test_session.tempfile.TemporaryDirectory.cleanup", cleanup), \
|
|
patch("native_test_session.time.sleep") as pause:
|
|
with self.assertRaises(subprocess.CalledProcessError) as caught:
|
|
run_private_session(
|
|
["dbus-run-session", "--", sys.executable, "-c", "raise SystemExit(7)"],
|
|
env=os.environ, capture_output=True, check=True, timeout=15)
|
|
self.assertEqual(caught.exception.returncode, 7)
|
|
self.assertEqual(len(attempts), 3)
|
|
self.assertEqual(pause.call_count, 2)
|
|
self.assertFalse(Path(attempts[0]).exists())
|
|
|
|
def test_cleanup_does_not_hide_unrelated_errors(self):
|
|
originalCleanup = tempfile.TemporaryDirectory.cleanup
|
|
|
|
def cleanup(runtime):
|
|
originalCleanup(runtime)
|
|
raise PermissionError(errno.EACCES, "Cleanup denied")
|
|
|
|
with patch("native_test_session.tempfile.TemporaryDirectory.cleanup", cleanup), \
|
|
patch("native_test_session.time.sleep") as pause:
|
|
with self.assertRaises(PermissionError):
|
|
run_private_session(
|
|
["dbus-run-session", "--", sys.executable, "-c", "pass"],
|
|
env=os.environ, capture_output=True, check=True, timeout=15)
|
|
pause.assert_not_called()
|
|
|
|
def test_portal_teardown_wait_is_bounded(self):
|
|
originalCleanup = tempfile.TemporaryDirectory.cleanup
|
|
|
|
def cleanup(runtime):
|
|
originalCleanup(runtime)
|
|
raise OSError(errno.ENOTCONN, "Document portal did not unmount")
|
|
|
|
with patch("native_test_session.tempfile.TemporaryDirectory.cleanup", cleanup), \
|
|
patch("native_test_session.time.monotonic", side_effect=[0, 6]), \
|
|
patch("native_test_session.time.sleep") as pause:
|
|
with self.assertRaises(OSError) as caught:
|
|
run_private_session(
|
|
["dbus-run-session", "--", sys.executable, "-c", "pass"],
|
|
env=os.environ, capture_output=True, check=True, timeout=15)
|
|
self.assertEqual(caught.exception.errno, errno.ENOTCONN)
|
|
pause.assert_not_called()
|
|
|
|
def test_rejects_dbus_before_xvfb_without_launching(self):
|
|
with patch("native_test_session.subprocess.run") as launch:
|
|
with self.assertRaisesRegex(ValueError, "Start Xvfb before D-Bus"):
|
|
run_private_session(
|
|
["dbus-run-session", "--", "xvfb-run", "-a", "true"], env=os.environ)
|
|
launch.assert_not_called()
|
|
|
|
def test_cli_activates_atspi_on_virtual_display(self):
|
|
script = """import ast, json, os, subprocess
|
|
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, check=True, timeout=10)
|
|
print(json.dumps(dict(display=os.environ['DISPLAY'], runtime=os.environ['XDG_RUNTIME_DIR'],
|
|
address=ast.literal_eval(result.stdout)[0])))
|
|
"""
|
|
result = subprocess.run(
|
|
[sys.executable, "-B", str(Path(__file__).with_name("native_test_session.py")),
|
|
sys.executable, "-c", script],
|
|
env=dict(os.environ, DISPLAY=":7777"), capture_output=True, text=True,
|
|
check=True, timeout=20)
|
|
state = json.loads(result.stdout)
|
|
self.assertNotEqual(state["display"], ":7777")
|
|
expectedPath = Path(state["runtime"]) / "at-spi" / ("bus" + state["display"].replace(":", "_"))
|
|
self.assertEqual(state["address"].split(",", 1)[0], "unix:path=" + str(expectedPath))
|
|
self.assertFalse(Path(state["runtime"]).exists())
|
|
|
|
def test_atspi_activation_preserves_other_session_socket(self):
|
|
# No real display is contacted. Reproduce D-Bus inheriting one display
|
|
# while its client uses another, with a live canary in the parent runtime.
|
|
with tempfile.TemporaryDirectory(prefix="cth-parent-", dir="/tmp") as parentRuntime:
|
|
socketDir = Path(parentRuntime) / "at-spi"
|
|
socketDir.mkdir(mode=0o700)
|
|
socketPath = socketDir / "bus_7777"
|
|
with socket.socket(socket.AF_UNIX) as listener:
|
|
listener.bind(str(socketPath))
|
|
listener.listen(1)
|
|
originalInode = socketPath.stat().st_ino
|
|
environment = dict(os.environ, XDG_RUNTIME_DIR=parentRuntime, DISPLAY=":7777",
|
|
AT_SPI_BUS_ADDRESS="unix:path=/invalid/desktop-accessibility")
|
|
result = run_private_session(
|
|
["dbus-run-session", "--", "env", "DISPLAY=:7778", "gdbus", "call",
|
|
"--session", "--dest", "org.a11y.Bus", "--object-path", "/org/a11y/bus",
|
|
"--method", "org.a11y.Bus.GetAddress"], env=environment,
|
|
capture_output=True, text=True, check=True, timeout=15)
|
|
address = ast.literal_eval(result.stdout)[0]
|
|
busPath = Path(address.split(",", 1)[0].removeprefix("unix:path="))
|
|
self.assertEqual(busPath.name, "bus_7777", address)
|
|
self.assertNotEqual(busPath.parent.parent, Path(parentRuntime))
|
|
self.assertFalse(busPath.parent.parent.exists(), "private runtime survived cleanup")
|
|
self.assertEqual(socketPath.stat().st_ino, originalInode)
|
|
self.assertTrue(stat.S_ISSOCK(socketPath.stat().st_mode))
|
|
with socket.socket(socket.AF_UNIX) as client:
|
|
client.settimeout(2)
|
|
client.connect(str(socketPath))
|
|
connection, _ = listener.accept()
|
|
connection.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|