From 679ff9161bc4bd3d984981a056b3bec9a91e2ca0 Mon Sep 17 00:00:00 2001 From: Scrates1 <49557842+Scrates1@users.noreply.github.com> Date: Mon, 3 Aug 2026 10:09:38 +0800 Subject: [PATCH] fix(ibus): require a restorable engine for scoped injection (#623) * fix(ibus): require a restorable engine for scoped injection * fix(ibus): preserve registered XKB restore targets Resolve GNOME XKB sources against registered IBus engine IDs so bare-XKB Wayland sessions keep scoped injection without fabricated restore names. Keep unknown states on the safe fallback path. * test(ibus): cover GNOME restore target failures --- src/vocalinux/text_injection/ibus_engine.py | 190 ++++++----- tests/test_ibus_engine.py | 330 ++++++++++++++------ 2 files changed, 338 insertions(+), 182 deletions(-) diff --git a/src/vocalinux/text_injection/ibus_engine.py b/src/vocalinux/text_injection/ibus_engine.py index b01087a..6054d61 100644 --- a/src/vocalinux/text_injection/ibus_engine.py +++ b/src/vocalinux/text_injection/ibus_engine.py @@ -338,87 +338,113 @@ def _is_gnome_session() -> bool: return "GNOME" in desktop.upper() -def get_current_engine_gnome_fallback() -> Optional[str]: - """ - Get the current keyboard layout as an IBus engine name via GNOME settings. - - On GNOME/Wayland, IBus may have no global engine (input sources are managed - by Mutter, not ibus-daemon). In that case ``ibus engine`` returns nothing - or a misleading ``xkb:us::eng`` default. This function reads the actual - keyboard layout from ``org.gnome.desktop.input-sources`` via gsettings. - - Returns: - IBus-format engine name (e.g. ``xkb:br::``) or None if not GNOME - or if gsettings fails. - """ +def _get_gnome_current_source() -> Optional[tuple[str, str]]: + """Return GNOME's current input source from the MRU list.""" if not _is_gnome_session(): return None try: - sources_result = subprocess.run( - ["gsettings", "get", "org.gnome.desktop.input-sources", "sources"], + result = subprocess.run( + ["gsettings", "get", "org.gnome.desktop.input-sources", "mru-sources"], capture_output=True, text=True, timeout=5, ) - if sources_result.returncode != 0: - logger.debug("gsettings get sources failed; not using GNOME fallback") + if result.returncode != 0: + logger.debug("gsettings get mru-sources failed; not using GNOME fallback") return None import ast - sources = ast.literal_eval(sources_result.stdout.strip()) - if not sources: - logger.debug("GNOME input-sources list is empty") + sources_text = result.stdout.strip() + if sources_text.startswith("@a(ss) "): + sources_text = sources_text[len("@a(ss) ") :] + sources = ast.literal_eval(sources_text) + if not isinstance(sources, (list, tuple)) or not sources: + logger.debug("GNOME mru-sources list is empty or invalid") return None - # Get the current index - current_result = subprocess.run( - ["gsettings", "get", "org.gnome.desktop.input-sources", "current"], + source = sources[0] + if not isinstance(source, (list, tuple)) or len(source) != 2: + logger.debug("GNOME current input source has an invalid shape") + return None + + source_type, source_id = source + if not isinstance(source_type, str) or not isinstance(source_id, str) or not source_id: + logger.debug("GNOME current input source has invalid values") + return None + return source_type, source_id + except (subprocess.SubprocessError, OSError, TypeError, ValueError, SyntaxError) as e: + logger.debug(f"GNOME fallback via gsettings failed: {e}") + return None + + +def _resolve_registered_xkb_engine(source_id: str) -> Optional[str]: + """Resolve a GNOME XKB source to a registered IBus engine ID.""" + layout, _, variant = source_id.partition("+") + if not layout: + return None + + try: + result = subprocess.run( + ["ibus", "list-engine", "--name-only"], capture_output=True, text=True, timeout=5, ) - if current_result.returncode != 0: - current_idx = 0 - else: - # ponytail: gsettings prints "uint32 N" or just "N" — split()[-1] handles both - current_idx = int(current_result.stdout.strip().split()[-1]) + if result.returncode != 0: + logger.debug("ibus list-engine failed; cannot verify the GNOME XKB source") + return None - if current_idx < 0 or current_idx >= len(sources): - current_idx = 0 + matches = [] + for engine_name in result.stdout.splitlines(): + engine_name = engine_name.strip() + parts = engine_name.split(":") + if len(parts) == 4 and parts[0] == "xkb" and parts[1] == layout and parts[2] == variant: + matches.append(engine_name) - source_type, source_id = sources[current_idx] + if matches: + # Multiple engine IDs can share an identical XKB layout/variant and + # differ only in language metadata. The first registered match keeps + # the same keyboard map without fabricating an engine name. + logger.debug(f"GNOME XKB source '{source_id}' maps to registered engine '{matches[0]}'") + return matches[0] - # GNOME uses ('xkb', 'br+dyn') or ('ibus', 'anthy') tuples. - # For xkb sources, build an IBus engine name like xkb::: - if source_type == "xkb": - layout, _, variant = source_id.partition("+") - if variant: - engine_name = f"xkb:{layout}::{variant}" - else: - engine_name = f"xkb:{layout}::" - logger.debug( - f"GNOME fallback: current source index {current_idx}, " - f"engine name: {engine_name}" - ) - return engine_name - elif source_type == "ibus": - # IBus IM engine — return the engine ID directly - logger.debug(f"GNOME fallback: current source is IBus engine '{source_id}'") - return source_id + logger.debug(f"GNOME XKB source '{source_id}' has no registered IBus engine") + except (subprocess.SubprocessError, OSError) as e: + logger.debug(f"Could not list registered IBus engines: {e}") + return None - logger.debug(f"GNOME fallback: unknown source type '{source_type}'") + +def get_current_engine_gnome_fallback() -> Optional[str]: + """ + Get a verified IBus engine for GNOME's current input source. + + On GNOME/Wayland, IBus may have no global engine (input sources are managed + by Mutter, not ibus-daemon). GNOME's ``current`` key is deprecated and + ignored, so the first entry in ``mru-sources`` is used as the current source. + + Explicit IBus source IDs are returned directly. XKB source IDs are matched + against registered IBus engines; names such as ``xkb:cn::`` are never + fabricated because a failed restoration can leave Vocalinux active. + + Returns: + A verified IBus engine ID, or None if the current source cannot be + mapped reliably to one. + """ + source = _get_gnome_current_source() + if not source: return None - except ( - subprocess.SubprocessError, - FileNotFoundError, - ValueError, - SyntaxError, - ) as e: - logger.debug(f"GNOME fallback via gsettings failed: {e}") - return None + source_type, source_id = source + if source_type == "ibus": + logger.debug(f"GNOME fallback: current source is IBus engine '{source_id}'") + return source_id + if source_type == "xkb": + return _resolve_registered_xkb_engine(source_id) + + logger.debug(f"GNOME fallback: unknown source type '{source_type}'") + return None def get_current_engine() -> Optional[str]: @@ -427,9 +453,8 @@ def get_current_engine() -> Optional[str]: On GNOME/Wayland, IBus may have no global engine set (Mutter manages input sources). In that case ``ibus engine`` can return a misleading - ``xkb:us::eng`` default that doesn't match the user's actual keyboard - layout. We detect this and fall back to reading the layout from GNOME's - gsettings so the real layout is preserved during engine restore. + ``xkb:us::eng`` default. GNOME sources are accepted only when they are + explicit IBus IDs or map to a registered XKB engine. Returns: The IBus engine name, or None if it cannot be determined. @@ -448,7 +473,8 @@ def get_current_engine() -> Optional[str]: stderr_lower = result.stderr.lower() if result.stderr else "" if "no global engine" in stderr_lower: logger.debug( - "ibus engine reports 'No global engine'; " "attempting GNOME gsettings fallback" + "ibus engine reports 'No global engine'; " + "checking GNOME's most recent IBus source" ) return get_current_engine_gnome_fallback() @@ -468,13 +494,19 @@ def get_current_engine() -> Optional[str]: "attempting GNOME gsettings fallback (see #497)" ) gnome_engine = get_current_engine_gnome_fallback() - if gnome_engine and gnome_engine != engine: + if gnome_engine: logger.debug( f"GNOME gsettings reports different engine: " f"{gnome_engine} (was {engine})" ) return gnome_engine + logger.debug( + "GNOME did not report a restorable IBus source; " + "ignoring the suspicious xkb:us::eng default" + ) + return None + return engine except (subprocess.SubprocessError, FileNotFoundError): pass @@ -1175,13 +1207,6 @@ class IBusTextInjector: the user's normal IBus/XKB engine active preserves dead-key composition and layout-specific behavior during ordinary typing. """ - # Capture the current engine before starting the Vocalinux process - # so inject_text() can restore it after committing text - if not is_engine_active(): - self._previous_engine = get_current_engine() - if self._previous_engine: - logger.debug(f"Captured current engine for later restore: {self._previous_engine}") - if not start_engine_process(): raise IBusSetupError("Failed to start IBus engine process. Check logs for details.") @@ -1318,18 +1343,18 @@ class IBusTextInjector: restore_engine: Optional[str] = None if not is_engine_active(): - current_engine = get_current_engine() or self._previous_engine - if current_engine: - restore_engine = current_engine - logger.debug( - "Temporarily activating Vocalinux IBus engine " - f"(will restore {current_engine})" - ) - else: - logger.debug( - "Could not determine current IBus engine; " - "activating Vocalinux engine without restore" + current_engine = get_current_engine() + if not current_engine: + logger.warning( + "Could not determine a restorable IBus engine; " + "using the non-IBus text injection fallback" ) + return False + + restore_engine = current_engine + logger.debug( + f"Temporarily activating Vocalinux IBus engine (will restore {current_engine})" + ) if not switch_engine(ENGINE_NAME): logger.error("Failed to activate Vocalinux IBus engine for injection") return False @@ -1434,7 +1459,10 @@ class IBusTextInjector: finally: if restore_engine: logger.debug(f"Restoring IBus engine after injection: {restore_engine}") - switch_engine(restore_engine) + if not switch_engine(restore_engine): + logger.error(f"Failed to restore IBus engine after injection: {restore_engine}") + else: + logger.debug(f"Restored IBus engine after injection: {restore_engine}") else: logger.debug( "Skipping engine restoration after injection — " diff --git a/tests/test_ibus_engine.py b/tests/test_ibus_engine.py index 44c8820..8d04e64 100644 --- a/tests/test_ibus_engine.py +++ b/tests/test_ibus_engine.py @@ -239,12 +239,12 @@ class TestGetCurrentEngine(unittest.TestCase): ): """Test that 'No global engine' in stderr triggers GNOME fallback.""" mock_run.return_value = MagicMock(stdout="", stderr="No global engine\n", returncode=1) - mock_gnome_fallback.return_value = "xkb:br::" + mock_gnome_fallback.return_value = "libpinyin" from vocalinux.text_injection.ibus_engine import get_current_engine result = get_current_engine() - self.assertEqual(result, "xkb:br::") + self.assertEqual(result, "libpinyin") mock_gnome_fallback.assert_called_once() @patch("vocalinux.text_injection.ibus_engine.get_current_engine_gnome_fallback") @@ -256,29 +256,29 @@ class TestGetCurrentEngine(unittest.TestCase): ): """Test that xkb:us::eng on GNOME/Wayland triggers gsettings fallback (#497).""" mock_run.return_value = MagicMock(stdout="xkb:us::eng\n", stderr="", returncode=0) - mock_gnome_fallback.return_value = "xkb:br::" + mock_gnome_fallback.return_value = "libpinyin" from vocalinux.text_injection.ibus_engine import get_current_engine result = get_current_engine() - self.assertEqual(result, "xkb:br::") + self.assertEqual(result, "libpinyin") mock_gnome_fallback.assert_called_once() @patch("vocalinux.text_injection.ibus_engine.get_current_engine_gnome_fallback") @patch("vocalinux.text_injection.ibus_engine._is_gnome_session", return_value=True) @patch("vocalinux.text_injection.ibus_engine._is_wayland_session", return_value=True) @patch("subprocess.run") - def test_get_current_engine_keeps_us_default_when_gnome_fails( + def test_get_current_engine_rejects_suspicious_us_when_gnome_fails( self, mock_run, mock_wayland, mock_gnome, mock_gnome_fallback ): - """Test that we keep xkb:us::eng when GNOME fallback returns None.""" + """Test that an unverified xkb:us::eng default is not used for restoration.""" mock_run.return_value = MagicMock(stdout="xkb:us::eng\n", stderr="", returncode=0) mock_gnome_fallback.return_value = None from vocalinux.text_injection.ibus_engine import get_current_engine result = get_current_engine() - self.assertEqual(result, "xkb:us::eng") + self.assertIsNone(result) @patch("vocalinux.text_injection.ibus_engine.get_current_engine_gnome_fallback") @patch("vocalinux.text_injection.ibus_engine._is_gnome_session", return_value=False) @@ -309,7 +309,7 @@ class TestGetCurrentEngineGnomeFallback(unittest.TestCase): @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) @patch("subprocess.run") - def test_gsettings_sources_failure_returns_none(self, mock_run): + def test_gsettings_mru_sources_failure_returns_none(self, mock_run): """Test that gsettings failure returns None.""" mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="") @@ -320,8 +320,8 @@ class TestGetCurrentEngineGnomeFallback(unittest.TestCase): @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) @patch("subprocess.run") - def test_empty_sources_returns_none(self, mock_run): - """Test that empty input-sources list returns None.""" + def test_empty_mru_sources_returns_none(self, mock_run): + """Test that an empty mru-sources list returns None.""" mock_run.return_value = MagicMock(returncode=0, stdout="[]", stderr="") from vocalinux.text_injection.ibus_engine import get_current_engine_gnome_fallback @@ -331,82 +331,202 @@ class TestGetCurrentEngineGnomeFallback(unittest.TestCase): @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) @patch("subprocess.run") - def test_xkb_source_returns_ibus_engine_name(self, mock_run): - """Test that an xkb source is converted to IBus engine format.""" - sources_output = "[('xkb', 'br'), ('xkb', 'us+altgr-intl')]" - current_output = "uint32 0" - mock_run.side_effect = [ - MagicMock(returncode=0, stdout=sources_output, stderr=""), - MagicMock(returncode=0, stdout=current_output, stderr=""), - ] + def test_invalid_current_source_shape_returns_none(self, mock_run): + """Test that a malformed current MRU tuple is rejected.""" + mock_run.return_value = MagicMock( + returncode=0, + stdout="[('ibus', 'libpinyin', 'extra')]", + stderr="", + ) from vocalinux.text_injection.ibus_engine import get_current_engine_gnome_fallback result = get_current_engine_gnome_fallback() - self.assertEqual(result, "xkb:br::") + self.assertIsNone(result) @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) @patch("subprocess.run") - def test_xkb_source_with_variant(self, mock_run): - """Test that an xkb source with variant is parsed correctly.""" - sources_output = "[('xkb', 'us+altgr-intl')]" - current_output = "uint32 0" + def test_invalid_current_source_values_return_none(self, mock_run): + """Test that non-string and empty current source values are rejected.""" + from vocalinux.text_injection.ibus_engine import get_current_engine_gnome_fallback + + for sources_output in ("[(1, 'libpinyin')]", "[('ibus', 1)]", "[('ibus', '')]"): + with self.subTest(sources_output=sources_output): + mock_run.return_value = MagicMock( + returncode=0, + stdout=sources_output, + stderr="", + ) + result = get_current_engine_gnome_fallback() + self.assertIsNone(result) + + @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) + @patch("subprocess.run") + def test_xkb_source_returns_registered_engine(self, mock_run): + """Test that an XKB source resolves to an exact registered IBus engine.""" + sources_output = "[('xkb', 'br'), ('xkb', 'us+altgr-intl')]" + engines_output = "xkb:us::eng\nxkb:br::por\nxkb:br:nodeadkeys:por\n" mock_run.side_effect = [ MagicMock(returncode=0, stdout=sources_output, stderr=""), - MagicMock(returncode=0, stdout=current_output, stderr=""), + MagicMock(returncode=0, stdout=engines_output, stderr=""), ] from vocalinux.text_injection.ibus_engine import get_current_engine_gnome_fallback result = get_current_engine_gnome_fallback() - self.assertEqual(result, "xkb:us::altgr-intl") + self.assertEqual(result, "xkb:br::por") + self.assertEqual(mock_run.call_args_list[0].args[0][-1], "mru-sources") + self.assertEqual(mock_run.call_args_list[1].args[0], ["ibus", "list-engine", "--name-only"]) + + @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) + @patch("subprocess.run") + def test_xkb_source_with_variant_returns_registered_engine(self, mock_run): + """Test that an XKB variant resolves only to a registered engine name.""" + sources_output = "[('xkb', 'us+altgr-intl')]" + engines_output = "xkb:us::eng\nxkb:us:altgr-intl:eng\n" + mock_run.side_effect = [ + MagicMock(returncode=0, stdout=sources_output, stderr=""), + MagicMock(returncode=0, stdout=engines_output, stderr=""), + ] + + from vocalinux.text_injection.ibus_engine import get_current_engine_gnome_fallback + + result = get_current_engine_gnome_fallback() + self.assertEqual(result, "xkb:us:altgr-intl:eng") @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) @patch("subprocess.run") def test_ibus_source_returns_engine_id(self, mock_run): """Test that an ibus source returns the engine ID directly.""" - sources_output = "[('ibus', 'anthy')]" - current_output = "uint32 0" - mock_run.side_effect = [ - MagicMock(returncode=0, stdout=sources_output, stderr=""), - MagicMock(returncode=0, stdout=current_output, stderr=""), - ] + sources_output = "[('ibus', 'libpinyin')]" + mock_run.return_value = MagicMock(returncode=0, stdout=sources_output, stderr="") from vocalinux.text_injection.ibus_engine import get_current_engine_gnome_fallback result = get_current_engine_gnome_fallback() - self.assertEqual(result, "anthy") + self.assertEqual(result, "libpinyin") @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) @patch("subprocess.run") - def test_current_index_selects_correct_source(self, mock_run): - """Test that the current index from gsettings selects the right source.""" - sources_output = "[('xkb', 'us'), ('xkb', 'br')]" - current_output = "uint32 1" - mock_run.side_effect = [ - MagicMock(returncode=0, stdout=sources_output, stderr=""), - MagicMock(returncode=0, stdout=current_output, stderr=""), - ] + def test_first_mru_source_is_current_gnome_source(self, mock_run): + """Test that the first MRU entry is treated as GNOME's current source.""" + sources_output = "[('ibus', 'libpinyin'), ('ibus', 'anthy')]" + mock_run.return_value = MagicMock(returncode=0, stdout=sources_output, stderr="") from vocalinux.text_injection.ibus_engine import get_current_engine_gnome_fallback result = get_current_engine_gnome_fallback() - self.assertEqual(result, "xkb:br::") + self.assertEqual(result, "libpinyin") + mock_run.assert_called_once() @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) @patch("subprocess.run") - def test_current_failure_defaults_to_index_zero(self, mock_run): - """Test that if gsettings get current fails, index 0 is used.""" - sources_output = "[('xkb', 'br'), ('xkb', 'us')]" + def test_current_xkb_source_does_not_scan_later_ibus_entry(self, mock_run): + """Test that a later IBus MRU entry does not replace the current XKB source.""" + sources_output = "[('xkb', 'br'), ('ibus', 'libpinyin')]" + engines_output = "xkb:br::por\n" mock_run.side_effect = [ MagicMock(returncode=0, stdout=sources_output, stderr=""), - MagicMock(returncode=1, stdout="", stderr=""), + MagicMock(returncode=0, stdout=engines_output, stderr=""), ] from vocalinux.text_injection.ibus_engine import get_current_engine_gnome_fallback result = get_current_engine_gnome_fallback() - self.assertEqual(result, "xkb:br::") + self.assertEqual(result, "xkb:br::por") + + @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) + @patch("subprocess.run") + def test_unregistered_xkb_source_returns_none(self, mock_run): + """Test that an XKB source without an exact registered engine is rejected.""" + sources_output = "[('xkb', 'cn')]" + engines_output = "xkb:us::eng\nxkb:br::por\n" + mock_run.side_effect = [ + MagicMock(returncode=0, stdout=sources_output, stderr=""), + MagicMock(returncode=0, stdout=engines_output, stderr=""), + ] + + from vocalinux.text_injection.ibus_engine import get_current_engine_gnome_fallback + + result = get_current_engine_gnome_fallback() + self.assertIsNone(result) + + def test_empty_xkb_source_id_returns_none(self): + """Test that an empty XKB source ID is rejected before listing engines.""" + from vocalinux.text_injection.ibus_engine import _resolve_registered_xkb_engine + + result = _resolve_registered_xkb_engine("") + self.assertIsNone(result) + + @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) + @patch("subprocess.run") + def test_xkb_engine_listing_failure_returns_none(self, mock_run): + """Test that a failed IBus engine listing keeps the source unknown.""" + mock_run.side_effect = [ + MagicMock(returncode=0, stdout="[('xkb', 'br')]", stderr=""), + MagicMock(returncode=1, stdout="", stderr="failed"), + ] + + from vocalinux.text_injection.ibus_engine import get_current_engine_gnome_fallback + + result = get_current_engine_gnome_fallback() + self.assertIsNone(result) + + @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) + @patch("subprocess.run") + def test_xkb_engine_listing_exception_returns_none(self, mock_run): + """Test that a missing IBus CLI keeps the XKB source unknown.""" + mock_run.side_effect = [ + MagicMock(returncode=0, stdout="[('xkb', 'br')]", stderr=""), + FileNotFoundError("ibus not found"), + ] + + from vocalinux.text_injection.ibus_engine import get_current_engine_gnome_fallback + + result = get_current_engine_gnome_fallback() + self.assertIsNone(result) + + @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) + @patch("subprocess.run") + def test_multiple_registered_xkb_engines_use_first_exact_match(self, mock_run): + """Test that language-only duplicates keep the first exact XKB match.""" + sources_output = "[('xkb', 'us+altgr-intl')]" + engines_output = "xkb:us:altgr-intl:eng\nxkb:us:altgr-intl:deu\n" + mock_run.side_effect = [ + MagicMock(returncode=0, stdout=sources_output, stderr=""), + MagicMock(returncode=0, stdout=engines_output, stderr=""), + ] + + from vocalinux.text_injection.ibus_engine import get_current_engine_gnome_fallback + + result = get_current_engine_gnome_fallback() + self.assertEqual(result, "xkb:us:altgr-intl:eng") + + @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) + @patch("subprocess.run") + def test_unknown_current_source_type_returns_none(self, mock_run): + """Test that a non-IBus, non-XKB GNOME source is not used for restoration.""" + mock_run.return_value = MagicMock( + returncode=0, + stdout="[('fcitx', 'mozc')]", + stderr="", + ) + + from vocalinux.text_injection.ibus_engine import get_current_engine_gnome_fallback + + result = get_current_engine_gnome_fallback() + self.assertIsNone(result) + + @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) + @patch("subprocess.run") + def test_typed_empty_mru_sources_returns_none(self, mock_run): + """Test that gsettings' typed empty-list output is handled safely.""" + mock_run.return_value = MagicMock(returncode=0, stdout="@a(ss) []", stderr="") + + from vocalinux.text_injection.ibus_engine import get_current_engine_gnome_fallback + + result = get_current_engine_gnome_fallback() + self.assertIsNone(result) @patch.dict("os.environ", {"XDG_CURRENT_DESKTOP": "GNOME"}) @patch("subprocess.run") @@ -665,7 +785,7 @@ class TestIBusTextInjector(unittest.TestCase): @patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir") @patch("vocalinux.text_injection.ibus_engine.start_engine_process", return_value=True) @patch("vocalinux.text_injection.ibus_engine.IBusTextInjector._wait_for_engine_ready") - def test_prepare_engine_captures_current_engine( + def test_prepare_engine_does_not_cache_restore_engine( self, mock_wait_ready, mock_start_engine, @@ -673,15 +793,17 @@ class TestIBusTextInjector(unittest.TestCase): mock_is_active, mock_get_current, ): - """Test prepare_engine captures the current engine for later restore.""" + """Test warmup does not cache an engine that can become stale before injection.""" from vocalinux.text_injection.ibus_engine import IBusTextInjector injector = IBusTextInjector(auto_activate=False) injector.prepare_engine() - self.assertEqual(injector._previous_engine, "xkb:fr::fra") + self.assertIsNone(injector._previous_engine) + mock_get_current.assert_not_called() + mock_is_active.assert_not_called() - @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:fr::fra") + @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="libpinyin") @patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False) @patch("vocalinux.text_injection.ibus_engine.switch_engine", return_value=True) @patch("socket.socket") @@ -714,7 +836,47 @@ class TestIBusTextInjector(unittest.TestCase): mock_sock.sendall.assert_called_once_with("Bonjour".encode("utf-8")) self.assertEqual( [call.args[0] for call in mock_switch.call_args_list], - [ENGINE_NAME, "xkb:fr::fra"], + [ENGINE_NAME, "libpinyin"], + ) + + @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="libpinyin") + @patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False) + @patch("vocalinux.text_injection.ibus_engine.switch_engine", side_effect=[True, False]) + @patch("socket.socket") + @patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH") + @patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True) + @patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir") + def test_inject_text_logs_restore_failure_after_successful_commit( + self, + mock_ensure_dir, + mock_socket_path, + mock_socket_cls, + mock_switch, + mock_is_active, + mock_get_current, + ): + """Test a restore failure is logged without retrying committed text.""" + from vocalinux.text_injection.ibus_engine import ENGINE_NAME, IBusTextInjector + + mock_socket_path.exists.return_value = True + mock_sock = MagicMock() + mock_sock.__enter__.return_value = mock_sock + mock_sock.__exit__.return_value = None + mock_sock.recv.return_value = b"OK" + mock_socket_cls.return_value = mock_sock + + injector = IBusTextInjector(auto_activate=False) + with self.assertLogs("vocalinux.text_injection.ibus_engine", level="ERROR") as logs: + result = injector.inject_text("你好") + + self.assertTrue(result) + self.assertEqual( + [call.args[0] for call in mock_switch.call_args_list], + [ENGINE_NAME, "libpinyin"], + ) + self.assertIn( + "Failed to restore IBus engine after injection: libpinyin", + "\n".join(logs.output), ) @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value=None) @@ -724,7 +886,7 @@ class TestIBusTextInjector(unittest.TestCase): @patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH") @patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True) @patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir") - def test_inject_text_uses_captured_engine_when_get_current_fails( + def test_inject_text_ignores_stale_cached_engine_when_get_current_fails( self, mock_ensure_dir, mock_socket_path, @@ -733,8 +895,8 @@ class TestIBusTextInjector(unittest.TestCase): mock_is_active, mock_get_current, ): - """Test inject_text uses _previous_engine when get_current_engine returns None.""" - from vocalinux.text_injection.ibus_engine import ENGINE_NAME, IBusTextInjector + """Test a startup-time engine is not used when the current engine is unknown.""" + from vocalinux.text_injection.ibus_engine import IBusTextInjector mock_socket_path.exists.return_value = True mock_sock = MagicMock() @@ -747,11 +909,9 @@ class TestIBusTextInjector(unittest.TestCase): injector._previous_engine = "xkb:de::deu" result = injector.inject_text("Hallo") - self.assertTrue(result) - self.assertEqual( - [call.args[0] for call in mock_switch.call_args_list], - [ENGINE_NAME, "xkb:de::deu"], - ) + self.assertFalse(result) + mock_switch.assert_not_called() + mock_socket_cls.assert_not_called() @patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True) @patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir") @@ -781,7 +941,7 @@ class TestIBusTextInjector(unittest.TestCase): result = injector.inject_text("Hello") self.assertFalse(result) - @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:us::eng") + @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:br::por") @patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False) @patch("vocalinux.text_injection.ibus_engine.switch_engine", return_value=True) @patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True) @@ -789,8 +949,8 @@ class TestIBusTextInjector(unittest.TestCase): def test_inject_text_success( self, mock_ensure_dir, mock_switch, mock_is_active, mock_get_current ): - """Test successful text injection via socket.""" - from vocalinux.text_injection.ibus_engine import IBusTextInjector + """Test successful text injection restores a verified XKB engine.""" + from vocalinux.text_injection.ibus_engine import ENGINE_NAME, IBusTextInjector # Create a mock server socket server_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) @@ -813,6 +973,10 @@ class TestIBusTextInjector(unittest.TestCase): server_sock.close() self.assertTrue(result) + self.assertEqual( + [call.args[0] for call in mock_switch.call_args_list], + [ENGINE_NAME, "xkb:br::por"], + ) @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:fr::fra") @patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False) @@ -912,7 +1076,7 @@ class TestIBusTextInjector(unittest.TestCase): @patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True) @patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir") @patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH") - def test_inject_text_activates_without_restore_when_engine_unknown( + def test_inject_text_does_not_activate_when_engine_unknown( self, mock_socket_path, mock_ensure_dir, @@ -920,7 +1084,7 @@ class TestIBusTextInjector(unittest.TestCase): mock_is_active, mock_get_current, ): - """Test inject_text activates engine but doesn't restore when current engine is unknown.""" + """Test injection falls back without changing IBus when restoration is unknown.""" from vocalinux.text_injection.ibus_engine import IBusTextInjector mock_socket_path.exists.return_value = True @@ -933,44 +1097,8 @@ class TestIBusTextInjector(unittest.TestCase): injector = IBusTextInjector(auto_activate=False) result = injector.inject_text("Hello") - self.assertTrue(result) - # Should activate engine once, no restore (previous engine unknown) - mock_switch.assert_called_once_with("vocalinux") - - @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value=None) - @patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False) - @patch("vocalinux.text_injection.ibus_engine.switch_engine", return_value=True) - @patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True) - @patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir") - def test_inject_text_no_engine_retries_when_no_previous_engine( - self, mock_ensure_dir, mock_switch, mock_is_active, mock_get_current - ): - """Test NO_ENGINE retries even when previous engine could not be captured.""" - from vocalinux.text_injection.ibus_engine import IBusTextInjector - - server_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - server_sock.bind(str(self.socket_path)) - server_sock.listen(3) - - def handle_connection(): - for _ in range(3): - conn, _ = server_sock.accept() - with conn: - conn.recv(65536) - conn.sendall(b"NO_ENGINE") - - server_thread = threading.Thread(target=handle_connection, daemon=True) - server_thread.start() - - with patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH", self.socket_path): - with patch("vocalinux.text_injection.ibus_engine.time"): - injector = IBusTextInjector(auto_activate=False) - result = injector.inject_text("Hello") - - server_sock.close() self.assertFalse(result) - # Should have tried to activate engine once at start, then re-activated on each retry - self.assertEqual(mock_switch.call_count, 3) + mock_switch.assert_not_called() @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:fr::fra") @patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False)