From a1d2e30b42d30b570bd0082aece99fd9936e4892 Mon Sep 17 00:00:00 2001 From: Ivan Date: Fri, 17 Apr 2026 23:50:29 -0500 Subject: [PATCH] feat(translator): refactor Argos CLI detection and add integration tests for forwarding functionality --- meshchatx/src/backend/translator_handler.py | 25 +++- .../test_lxmf_forwarding_integration.py | 125 ++++++++++++++++++ .../test_translator_argos_integration.py | 50 +++++++ .../test_translator_handler_extended.py | 1 + 4 files changed, 195 insertions(+), 6 deletions(-) create mode 100644 tests/backend/test_lxmf_forwarding_integration.py create mode 100644 tests/backend/test_translator_argos_integration.py diff --git a/meshchatx/src/backend/translator_handler.py b/meshchatx/src/backend/translator_handler.py index fe07a6a..2be09dc 100644 --- a/meshchatx/src/backend/translator_handler.py +++ b/meshchatx/src/backend/translator_handler.py @@ -22,8 +22,17 @@ try: except ImportError: HAS_ARGOS_LIB = False -HAS_ARGOS_CLI = shutil.which("argos-translate") is not None -HAS_ARGOS = HAS_ARGOS_LIB or HAS_ARGOS_CLI +ARGOS_CLI_EXECUTABLE_NAMES = ("argos-translate", "argostranslate") + + +def _find_argos_cli_executable() -> str | None: + """Resolve the Argos CLI on PATH (checked at call time, not import time).""" + for name in ARGOS_CLI_EXECUTABLE_NAMES: + path = shutil.which(name) + if path: + return path + return None + LANGUAGE_CODE_TO_NAME = { "en": "English", @@ -83,9 +92,10 @@ class TranslatorHandler: "LIBRETRANSLATE_URL", "http://localhost:5000", ) - self.has_argos = HAS_ARGOS self.has_argos_lib = HAS_ARGOS_LIB - self.has_argos_cli = HAS_ARGOS_CLI + self._argos_cli_executable = _find_argos_cli_executable() + self.has_argos_cli = self._argos_cli_executable is not None + self.has_argos = self.has_argos_lib or self.has_argos_cli self.has_requests = HAS_AIOHTTP async def _fetch_languages_async(self, url: str): @@ -322,9 +332,12 @@ class TranslatorHandler: msg = f"Invalid language codes: {source_lang} -> {target_lang}" raise ValueError(msg) - executable = shutil.which("argos-translate") + executable = self._argos_cli_executable or _find_argos_cli_executable() if not executable: - msg = "argos-translate executable not found in PATH" + msg = ( + "Argos Translate CLI not found in PATH " + f"(tried: {', '.join(ARGOS_CLI_EXECUTABLE_NAMES)})" + ) raise RuntimeError(msg) try: diff --git a/tests/backend/test_lxmf_forwarding_integration.py b/tests/backend/test_lxmf_forwarding_integration.py new file mode 100644 index 0000000..4b0aa17 --- /dev/null +++ b/tests/backend/test_lxmf_forwarding_integration.py @@ -0,0 +1,125 @@ +# SPDX-License-Identifier: 0BSD + +"""Integration-style tests for LXMF handle_forwarding (rule path and reply path).""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from meshchatx.meshchat import ReticulumMeshChat +from meshchatx.src.backend.async_utils import AsyncUtils + + +def _run_async_immediate(coro): + asyncio.run(coro) + + +@pytest.fixture +def forwarding_app(): + app = MagicMock(spec=ReticulumMeshChat) + app.current_context = MagicMock() + app.current_context.database = MagicMock() + app.current_context.database.messages = MagicMock() + app.current_context.database.misc = MagicMock() + app.current_context.forwarding_manager = MagicMock() + + app.send_message = AsyncMock() + + app.handle_forwarding = ReticulumMeshChat.handle_forwarding.__get__( + app, + ReticulumMeshChat, + ) + return app + + +def _make_lxmf_message(source_hex: str, dest_hex: str): + msg = MagicMock() + msg.source_hash = bytes.fromhex(source_hex) + msg.destination_hash = bytes.fromhex(dest_hex) + msg.content = "body" + msg.title = "t" + msg.get_fields = MagicMock(return_value={}) + return msg + + +@pytest.mark.integration +def test_handle_forwarding_forward_path_sends_via_alias(forwarding_app): + app = forwarding_app + ctx = app.current_context + + src = "aa" * 16 + dest_identity = "bb" * 16 + forward_to = "cc" * 16 + alias = "dd" * 16 + + ctx.database.messages.get_forwarding_mapping.return_value = None + ctx.database.misc.get_forwarding_rules.return_value = [ + { + "source_filter_hash": None, + "forward_to_hash": forward_to, + }, + ] + ctx.forwarding_manager.get_or_create_mapping.return_value = { + "alias_hash": alias, + } + + msg = _make_lxmf_message(src, dest_identity) + + with patch.object(AsyncUtils, "run_async", side_effect=_run_async_immediate): + app.handle_forwarding(msg, context=ctx) + + app.send_message.assert_called_once() + kwargs = app.send_message.call_args[1] + assert kwargs["destination_hash"] == forward_to + assert kwargs["sender_identity_hash"] == alias + assert kwargs["content"] == "body" + + +@pytest.mark.integration +def test_handle_forwarding_reply_path_sends_to_original_sender(forwarding_app): + app = forwarding_app + ctx = app.current_context + + original = "ee" * 16 + alias = "ff" * 16 + + ctx.database.messages.get_forwarding_mapping.return_value = { + "original_sender_hash": original, + } + msg = _make_lxmf_message("aa" * 16, alias) + + with patch.object(AsyncUtils, "run_async", side_effect=_run_async_immediate): + app.handle_forwarding(msg, context=ctx) + + ctx.database.misc.get_forwarding_rules.assert_not_called() + app.send_message.assert_called_once() + kwargs = app.send_message.call_args[1] + assert kwargs["destination_hash"] == original + assert "sender_identity_hash" not in kwargs + + +@pytest.mark.integration +def test_handle_forwarding_skips_rule_when_source_filter_mismatches(forwarding_app): + app = forwarding_app + ctx = app.current_context + + src = "aa" * 16 + dest_identity = "bb" * 16 + other = "99" * 16 + + ctx.database.messages.get_forwarding_mapping.return_value = None + ctx.database.misc.get_forwarding_rules.return_value = [ + { + "source_filter_hash": other, + "forward_to_hash": "cc" * 16, + }, + ] + + msg = _make_lxmf_message(src, dest_identity) + + with patch.object(AsyncUtils, "run_async", side_effect=_run_async_immediate): + app.handle_forwarding(msg, context=ctx) + + app.send_message.assert_not_called() + ctx.forwarding_manager.get_or_create_mapping.assert_not_called() diff --git a/tests/backend/test_translator_argos_integration.py b/tests/backend/test_translator_argos_integration.py new file mode 100644 index 0000000..5f2daab --- /dev/null +++ b/tests/backend/test_translator_argos_integration.py @@ -0,0 +1,50 @@ +# SPDX-License-Identifier: 0BSD + +"""Optional integration checks for Argos CLI (skipped when not installed).""" + +import shutil + +import pytest + +from meshchatx.src.backend.translator_handler import ( + ARGOS_CLI_EXECUTABLE_NAMES, + TranslatorHandler, + _find_argos_cli_executable, +) + + +def _argos_cli_on_path() -> bool: + return _find_argos_cli_executable() is not None + + +@pytest.mark.integration +def test_find_argos_cli_matches_shutil_which(): + expected = None + for name in ARGOS_CLI_EXECUTABLE_NAMES: + expected = shutil.which(name) + if expected: + break + assert _find_argos_cli_executable() == expected + + +@pytest.mark.integration +@pytest.mark.skipif(not _argos_cli_on_path(), reason="Argos CLI not on PATH") +def test_translate_en_es_via_cli_round_trip(): + handler = TranslatorHandler(enabled=True) + assert handler.has_argos_cli + assert not handler.has_argos_lib + + result = handler.translate_text("Hello", "en", "es", use_argos=True) + assert result["source"] == "argos" + assert result["source_lang"] == "en" + assert result["target_lang"] == "es" + assert len(result["translated_text"].strip()) > 0 + + +@pytest.mark.integration +@pytest.mark.skipif(not _argos_cli_on_path(), reason="Argos CLI not on PATH") +def test_get_supported_languages_includes_argos_when_libretranslate_down(): + handler = TranslatorHandler(enabled=True) + langs = handler.get_supported_languages() + argos = [x for x in langs if x.get("source") == "argos"] + assert len(argos) >= 1 diff --git a/tests/backend/test_translator_handler_extended.py b/tests/backend/test_translator_handler_extended.py index c0178bc..e1557f2 100644 --- a/tests/backend/test_translator_handler_extended.py +++ b/tests/backend/test_translator_handler_extended.py @@ -73,6 +73,7 @@ def test_translate_argos_cli(mock_run): handler.has_argos_cli = True handler.has_argos = True handler.has_requests = False # Force CLI + handler._argos_cli_executable = "/usr/bin/argos-translate" with patch("shutil.which", return_value="/usr/bin/argos-translate"): result = handler.translate_text(