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.
52 lines
2.5 KiB
Python
52 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Check private browser daemon startup without contacting a desktop bus."""
|
|
import subprocess
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
import native_browser
|
|
|
|
|
|
class AtspiStartupTests(unittest.TestCase):
|
|
def test_readiness_queries_bus_without_activating_service(self):
|
|
process = Mock()
|
|
process.poll.return_value = None
|
|
replies = [Mock(returncode=0, stdout="(false,)"),
|
|
Mock(returncode=0, stdout="(true,)")]
|
|
with patch("native_browser.subprocess.run", side_effect=replies) as run, \
|
|
patch("native_browser.time.sleep"):
|
|
native_browser.wait_for_bus_name(process, ["--session"], "org.a11y.Bus")
|
|
self.assertEqual(run.call_count, 2)
|
|
for call in run.call_args_list:
|
|
args = call.args[0]
|
|
self.assertEqual(args[args.index("--dest") + 1], "org.freedesktop.DBus")
|
|
self.assertEqual(args[-2:], ["org.freedesktop.DBus.NameHasOwner", "org.a11y.Bus"])
|
|
self.assertEqual(call.kwargs["timeout"], 2)
|
|
|
|
def test_registry_failure_joins_both_daemons(self):
|
|
launcher, registry = Mock(), Mock()
|
|
with patch("native_browser.shutil.which", side_effect=lambda name: "/private/" + name), \
|
|
patch("native_browser.subprocess.Popen", side_effect=[launcher, registry]) as popen, \
|
|
patch("native_browser.wait_for_bus_name", side_effect=[None, RuntimeError("registry failed")]), \
|
|
patch("native_browser.subprocess.run", return_value=Mock(stdout="('unix:path=/private/bus',)")):
|
|
with self.assertRaisesRegex(RuntimeError, "registry failed"):
|
|
native_browser.start_atspi_bus()
|
|
self.assertEqual(popen.call_args.kwargs["env"]["AT_SPI_BUS_ADDRESS"], "unix:path=/private/bus")
|
|
for process in (launcher, registry):
|
|
process.terminate.assert_called_once_with()
|
|
process.wait.assert_called_once_with(timeout=5)
|
|
|
|
def test_probe_timeout_joins_launcher(self):
|
|
launcher = Mock()
|
|
with patch("native_browser.shutil.which", return_value="/private/daemon"), \
|
|
patch("native_browser.subprocess.Popen", return_value=launcher), \
|
|
patch("native_browser.wait_for_bus_name", side_effect=subprocess.TimeoutExpired("gdbus", 2)):
|
|
with self.assertRaises(subprocess.TimeoutExpired):
|
|
native_browser.start_atspi_bus()
|
|
launcher.terminate.assert_called_once_with()
|
|
launcher.wait.assert_called_once_with(timeout=5)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|