Harden XTerm handoff and web event routing

tests/test_input_event_manager_x11_focus_regressions.py

@@ -530,19 +530,200 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
         activeScript = mock.Mock()
         scriptManager = mock.Mock()
         scriptManager.get_active_script.return_value = activeScript
-        screen = mock.Mock()
-        activeWindow = object()
-        screen.get_active_window.return_value = activeWindow
         manager._scriptWithSuspendedGrabsForXterm = activeScript
 
         with (
             mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager),
-            mock.patch.object(manager, "_x11_window_xterm_match", return_value=False) as matcher,
+            mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=False) as matcher,
             mock.patch.object(input_event_manager.debug, "print_tokens"),
         ):
-            manager._on_active_x11_window_changed(screen, None)
+            manager._on_active_x11_window_changed(mock.Mock(), None)
 
+        matcher.assert_called_once_with()
+        activeScript.addKeyGrabs.assert_called_once_with()
+        self.assertIsNone(manager._scriptWithSuspendedGrabsForXterm)
+
+    def test_active_x11_window_uses_ewmh_id_instead_of_stale_wnck_active_window(self):
+        manager = input_event_manager.InputEventManager()
+        wnck = mock.Mock()
+        screen = mock.Mock()
+        ewmhWindow = object()
+        wnck.Screen.get_default.return_value = screen
+        wnck.Window.get.return_value = ewmhWindow
+
+        with (
+            mock.patch.object(manager, "_get_wnck", return_value=wnck),
+            mock.patch.object(manager, "_ensure_xterm_focus_monitor"),
+            mock.patch.object(manager, "_get_ewmh_active_x11_window_id", return_value=0x1400010),
+        ):
+            result = manager._get_active_x11_window()
+
+        self.assertIs(result, ewmhWindow)
+        wnck.Window.get.assert_called_once_with(0x1400010)
+        screen.get_active_window.assert_not_called()
+
+    def test_active_x11_window_does_not_fall_back_to_stale_wnck_window_for_known_ewmh_id(self):
+        manager = input_event_manager.InputEventManager()
+        wnck = mock.Mock()
+        screen = mock.Mock()
+        wnck.Screen.get_default.return_value = screen
+        wnck.Window.get.return_value = None
+
+        with (
+            mock.patch.object(manager, "_get_wnck", return_value=wnck),
+            mock.patch.object(manager, "_ensure_xterm_focus_monitor"),
+            mock.patch.object(manager, "_get_ewmh_active_x11_window_id", return_value=0x1400010),
+            mock.patch.object(input_event_manager.debug, "print_tokens"),
+        ):
+            result = manager._get_active_x11_window()
+
+        self.assertIsNone(result)
+        screen.get_active_window.assert_not_called()
+
+    def test_active_x11_window_falls_back_to_wnck_when_ewmh_is_unavailable(self):
+        manager = input_event_manager.InputEventManager()
+        wnck = mock.Mock()
+        screen = mock.Mock()
+        fallbackWindow = object()
+        wnck.Screen.get_default.return_value = screen
+        screen.get_active_window.return_value = fallbackWindow
+
+        with (
+            mock.patch.object(manager, "_get_wnck", return_value=wnck),
+            mock.patch.object(manager, "_ensure_xterm_focus_monitor"),
+            mock.patch.object(manager, "_get_ewmh_active_x11_window_id", return_value=None),
+        ):
+            result = manager._get_active_x11_window()
+
+        self.assertIs(result, fallbackWindow)
+
+    def test_xterm_match_reuses_cached_metadata_while_ewmh_window_id_is_unchanged(self):
+        manager = input_event_manager.InputEventManager()
+        activeWindow = object()
+
+        with (
+            mock.patch.object(
+                manager,
+                "_get_ewmh_active_x11_window_id",
+                return_value=0x1400010,
+            ),
+            mock.patch.object(
+                manager,
+                "_get_active_x11_window",
+                return_value=activeWindow,
+            ) as getActiveWindow,
+            mock.patch.object(
+                manager,
+                "_x11_window_xterm_match",
+                return_value=False,
+            ) as matcher,
+        ):
+            firstResult = manager._active_x11_window_xterm_match()
+            secondResult = manager._active_x11_window_xterm_match()
+
+        self.assertFalse(firstResult)
+        self.assertFalse(secondResult)
+        getActiveWindow.assert_called_once_with(ewmhWindowId=0x1400010)
         matcher.assert_called_once_with(activeWindow)
+
+    def test_xterm_focus_change_suspends_grabs_without_keyboard_event(self):
+        manager = input_event_manager.InputEventManager()
+        activeScript = mock.Mock()
+        scriptManager = mock.Mock()
+        scriptManager.get_active_script.return_value = activeScript
+
+        with (
+            mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager),
+            mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=True) as matcher,
+            mock.patch.object(input_event_manager.debug, "print_tokens"),
+        ):
+            manager._on_active_x11_window_changed(mock.Mock(), None)
+
+        matcher.assert_called_once_with()
+        activeScript.removeKeyGrabs.assert_called_once_with()
+        self.assertIs(manager._scriptWithSuspendedGrabsForXterm, activeScript)
+
+    def test_initial_xterm_handoff_suspends_grabs(self):
+        manager = input_event_manager.InputEventManager()
+        manager._device = object()
+
+        with (
+            mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=True),
+            mock.patch.object(manager, "_suspend_key_grabs_for_xterm") as suspend,
+            mock.patch.object(manager, "_ensure_xterm_recovery_timer") as ensureTimer,
+            mock.patch.object(input_event_manager.debug, "print_tokens"),
+        ):
+            manager._initialize_xterm_handoff()
+
+        suspend.assert_called_once_with()
+        ensureTimer.assert_called_once_with()
+
+    def test_xterm_focus_poll_suspends_grabs_after_entering_xterm(self):
+        manager = input_event_manager.InputEventManager()
+        manager._device = object()
+
+        with (
+            mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=True),
+            mock.patch.object(manager, "_suspend_key_grabs_for_xterm") as suspend,
+            mock.patch.object(manager, "_restore_key_grabs_after_xterm") as restore,
+            mock.patch.object(input_event_manager.debug, "print_tokens"),
+        ):
+            result = manager._poll_xterm_grab_recovery()
+
+        self.assertTrue(result)
+        suspend.assert_called_once_with()
+        restore.assert_not_called()
+
+    def test_xterm_focus_poll_restores_grabs_after_leaving_xterm(self):
+        manager = input_event_manager.InputEventManager()
+        manager._device = object()
+
+        with (
+            mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=False),
+            mock.patch.object(manager, "_suspend_key_grabs_for_xterm") as suspend,
+            mock.patch.object(manager, "_restore_key_grabs_after_xterm") as restore,
+            mock.patch.object(input_event_manager.debug, "print_tokens"),
+        ):
+            result = manager._poll_xterm_grab_recovery()
+
+        self.assertTrue(result)
+        restore.assert_called_once_with()
+        suspend.assert_not_called()
+
+    def test_xterm_suppression_does_not_check_x11_without_active_handoff(self):
+        manager = input_event_manager.InputEventManager()
+
+        with mock.patch.object(manager, "_active_x11_window_xterm_match") as matcher:
+            result = manager.should_suppress_atspi_events_for_xterm(refresh=True)
+
+        self.assertFalse(result)
+        matcher.assert_not_called()
+
+    def test_xterm_suppression_ignores_background_event_without_x11_refresh(self):
+        manager = input_event_manager.InputEventManager()
+        manager._scriptWithSuspendedGrabsForXterm = object()
+
+        with mock.patch.object(manager, "_active_x11_window_xterm_match") as matcher:
+            result = manager.should_suppress_atspi_events_for_xterm(refresh=False)
+
+        self.assertTrue(result)
+        matcher.assert_not_called()
+
+    def test_xterm_suppression_releases_handoff_for_real_focus_return(self):
+        manager = input_event_manager.InputEventManager()
+        activeScript = mock.Mock()
+        scriptManager = mock.Mock()
+        scriptManager.get_active_script.return_value = activeScript
+        manager._scriptWithSuspendedGrabsForXterm = activeScript
+
+        with (
+            mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager),
+            mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=False),
+            mock.patch.object(input_event_manager.debug, "print_tokens"),
+        ):
+            result = manager.should_suppress_atspi_events_for_xterm(refresh=True)
+
+        self.assertFalse(result)
         activeScript.addKeyGrabs.assert_called_once_with()
         self.assertIsNone(manager._scriptWithSuspendedGrabsForXterm)
 
@@ -550,14 +731,13 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
         manager = input_event_manager.InputEventManager()
         manager._device = object()
         manager._scriptWithSuspendedGrabsForXterm = object()
-        screen = mock.Mock()
-        screen.get_active_window.side_effect = RuntimeError("focus unavailable")
 
         with (
+            mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=None),
             mock.patch.object(input_event_manager.GLib, "timeout_add", return_value=23),
-            mock.patch.object(input_event_manager.debug, "print_message"),
+            mock.patch.object(input_event_manager.debug, "print_tokens"),
         ):
-            manager._on_active_x11_window_changed(screen, None)
+            manager._on_active_x11_window_changed(mock.Mock(), None)
 
         self.assertEqual(manager._xtermRecoverySourceId, 23)
 
@@ -619,6 +799,9 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
         manager._xtermFocusHandlerId = 17
         manager._xtermRecoverySourceId = 23
         manager._scriptWithSuspendedGrabsForXterm = object()
+        x11Display = mock.Mock()
+        manager._x11Display = x11Display
+        manager._didAttemptX11Display = True
 
         with (
             mock.patch.object(input_event_manager.GLib, "source_remove") as sourceRemove,
@@ -632,6 +815,8 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
         self.assertEqual(manager._xtermFocusHandlerId, 0)
         self.assertEqual(manager._xtermRecoverySourceId, 0)
         self.assertIsNone(manager._scriptWithSuspendedGrabsForXterm)
+        x11Display.close.assert_called_once_with()
+        self.assertFalse(manager._didAttemptX11Display)
 
     def test_identifier_is_xterm_matches_exact_xterm_only(self):
         self.assertTrue(input_event_manager.InputEventManager._identifier_is_xterm("xterm"))