diff --git a/inputremapper/focus/backends/sway.py b/inputremapper/focus/backends/sway.py index 8e53d71..fbff23a 100644 --- a/inputremapper/focus/backends/sway.py +++ b/inputremapper/focus/backends/sway.py @@ -171,10 +171,14 @@ class SwayFocusBackend(FocusBackend): return change = data.get("change") - if change not in ("focus", "title", "new"): + if change not in ("focus", "title"): return - event = _container_to_event(data.get("container") or {}) + if change == "title" and not (data.get("container") or {}).get("focused"): + event = self.get_current() + else: + event = _container_to_event(data.get("container") or {}) + if event is not None and self._on_focus is not None: self._on_focus(event) diff --git a/inputremapper/focus/backends/wlr_foreign_toplevel.py b/inputremapper/focus/backends/wlr_foreign_toplevel.py index 391f0ec..f95fc32 100644 --- a/inputremapper/focus/backends/wlr_foreign_toplevel.py +++ b/inputremapper/focus/backends/wlr_foreign_toplevel.py @@ -69,6 +69,7 @@ class WlrForeignToplevelBackend(FocusBackend): # handle -> {"app_id": str, "title": str, "activated": bool} self._toplevels: Dict[Any, Dict[str, Any]] = {} self._current: Optional[FocusEvent] = None + self._current_handle: Optional[Any] = None @staticmethod def is_available() -> bool: @@ -171,12 +172,20 @@ class WlrForeignToplevelBackend(FocusBackend): event = FocusEvent( app_id=info["app_id"], title=info["title"], backend=self.name ) + self._current_handle = handle self._current = event if self._on_focus is not None: self._on_focus(event) def _on_closed(self, handle) -> None: self._toplevels.pop(handle, None) + if handle != self._current_handle: + return + + self._current_handle = None + self._current = FocusEvent(app_id="", title="", backend=self.name) + if self._on_focus is not None: + self._on_focus(self._current) def _on_readable(self, _fd: int, condition: int) -> bool: if condition & (GLib.IO_HUP | GLib.IO_ERR): @@ -207,4 +216,5 @@ class WlrForeignToplevelBackend(FocusBackend): self._manager = None self._registry = None self._toplevels.clear() + self._current_handle = None self._current = None diff --git a/inputremapper/focus/focus_service.py b/inputremapper/focus/focus_service.py index 1f205ad..ac14891 100644 --- a/inputremapper/focus/focus_service.py +++ b/inputremapper/focus/focus_service.py @@ -164,12 +164,6 @@ class FocusService: def _on_focus(self, event: FocusEvent) -> None: """Called by the backend on every focus change (debounced here).""" - if not event.app_id: - # Transient focus (menus, popups, no focused window). Ignore it so - # we don't tear down injections during alt-tab. - logger.debug("Ignoring transient focus event: %s", event) - return - self._pending_event = event if self._debounce_id is not None: GLib.source_remove(self._debounce_id) diff --git a/tests/unit/test_focus_service.py b/tests/unit/test_focus_service.py index aca9a63..16bb9cb 100644 --- a/tests/unit/test_focus_service.py +++ b/tests/unit/test_focus_service.py @@ -233,11 +233,33 @@ class TestFocusService(unittest.TestCase): # -- debounce / transient focus --------------------------------------- - def test_transient_empty_focus_is_ignored(self): + def test_empty_focus_stops_app_controlled(self): service = self._make_service([self._binding("firefox", [("devX", "p1")])]) + service._reconcile(_event("firefox")) + self.daemon.calls["start_injecting"].clear() + service._on_focus(_event(app_id="")) - self.assertIsNone(service._pending_event) - self.assertIsNone(service._debounce_id) + service._flush() + self.assertEqual(self.daemon.calls["stop_injecting"], ["devX"]) + self.assertEqual(service._app_controlled, {}) + self.assertEqual(self.daemon.calls["start_injecting"], []) + + def test_focus_event_for_bound_app_starts_injecting(self): + service = self._make_service([self._binding("firefox", [("devX", "p1")])]) + service._on_focus(_event("firefox")) + service._flush() + self.assertEqual(self.daemon.calls["start_injecting"], [("devX", "p1")]) + self.assertEqual(service._app_controlled, {"devX": "p1"}) + + def test_focus_event_for_unbound_app_stops_app_controlled(self): + service = self._make_service([self._binding("firefox", [("devX", "p1")])]) + service._reconcile(_event("firefox")) + self.daemon.calls["start_injecting"].clear() + + service._on_focus(_event("code")) + service._flush() + self.assertEqual(self.daemon.calls["stop_injecting"], ["devX"]) + self.assertEqual(service._app_controlled, {}) self.assertEqual(self.daemon.calls["start_injecting"], []) def test_debounce_collapses_rapid_focus(self): diff --git a/tests/unit/test_focus_watcher.py b/tests/unit/test_focus_watcher.py index a30df6e..7744a67 100644 --- a/tests/unit/test_focus_watcher.py +++ b/tests/unit/test_focus_watcher.py @@ -33,6 +33,7 @@ from inputremapper.focus.backends.wlr_foreign_toplevel import ( WlrForeignToplevelBackend, ) from inputremapper.focus.backends.xorg import XorgFocusBackend +from inputremapper.focus.focus_backend import FocusEvent from inputremapper.focus.focus_watcher import focus_backend_classes, select_backend from tests.lib.test_setup import test_setup @@ -173,6 +174,52 @@ class TestSwayParsing(unittest.TestCase): self.assertEqual(len(received), 1) self.assertEqual(received[0].app_id, "kitty") + def test_handle_event_ignores_new_window(self): + backend = SwayFocusBackend() + received = [] + backend._on_focus = received.append + payload = json.dumps( + {"change": "new", "container": {"app_id": "firefox", "name": "Mozilla"}} + ).encode() + backend._handle_event(payload) + self.assertEqual(received, []) + + def test_handle_event_title_nonfocused_requeries_current_focus(self): + backend = SwayFocusBackend() + received = [] + backend._on_focus = received.append + current = FocusEvent(app_id="code", title="editor", backend="sway") + + with patch.object(backend, "get_current", return_value=current) as get_current: + payload = json.dumps( + { + "change": "title", + "container": { + "focused": False, + "app_id": "firefox", + "name": "Background Mozilla", + }, + } + ).encode() + backend._handle_event(payload) + + get_current.assert_called_once_with() + self.assertEqual(received, [current]) + + def test_handle_event_title_focused_uses_payload(self): + backend = SwayFocusBackend() + received = [] + backend._on_focus = received.append + payload = json.dumps( + { + "change": "title", + "container": {"focused": True, "app_id": "kitty", "name": "term"}, + } + ).encode() + backend._handle_event(payload) + self.assertEqual(len(received), 1) + self.assertEqual(received[0].app_id, "kitty") + def test_handle_event_ignores_unrelated_change(self): backend = SwayFocusBackend() received = [] @@ -294,6 +341,7 @@ class TestWlrBackend(unittest.TestCase): self.assertEqual(received[0].app_id, "firefox") self.assertEqual(received[0].title, "Mozilla") self.assertEqual(backend.get_current().app_id, "firefox") + self.assertIs(backend._current_handle, handle) def test_done_without_activation_is_ignored(self): backend = WlrForeignToplevelBackend() @@ -315,6 +363,56 @@ class TestWlrBackend(unittest.TestCase): backend._on_closed(handle) self.assertNotIn(handle, backend._toplevels) + def test_closed_current_toplevel_emits_unfocus(self): + backend = WlrForeignToplevelBackend() + received = [] + backend._on_focus = received.append + handle = object() + backend._toplevels[handle] = { + "app_id": "firefox", + "title": "Mozilla", + "activated": True, + } + backend._on_done(handle) + received.clear() + + backend._on_closed(handle) + + self.assertNotIn(handle, backend._toplevels) + self.assertIsNone(backend._current_handle) + self.assertEqual( + backend.get_current(), + FocusEvent(app_id="", title="", backend="wlr-foreign-toplevel"), + ) + self.assertEqual(received, [backend.get_current()]) + + def test_closed_non_current_toplevel_does_not_emit(self): + backend = WlrForeignToplevelBackend() + received = [] + backend._on_focus = received.append + current_handle = object() + closed_handle = object() + backend._toplevels[current_handle] = { + "app_id": "code", + "title": "Editor", + "activated": True, + } + backend._toplevels[closed_handle] = { + "app_id": "firefox", + "title": "Mozilla", + "activated": False, + } + backend._on_done(current_handle) + current = backend.get_current() + received.clear() + + backend._on_closed(closed_handle) + + self.assertNotIn(closed_handle, backend._toplevels) + self.assertIs(backend._current_handle, current_handle) + self.assertEqual(backend.get_current(), current) + self.assertEqual(received, []) + def test_state_unknown_handle_ignored(self): backend = WlrForeignToplevelBackend() # must not raise for a handle we never registered