From 379e7dd014275ca066dc7ee3d7f0a1dfd61c2bdd Mon Sep 17 00:00:00 2001 From: Dominikh Date: Sun, 10 May 2026 18:37:11 +0000 Subject: [PATCH] test(send_message): cover _check_send_message gating paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a TestCheckSendMessage class with 7 focused tests pinning the four passing conditions and the failure modes: - HERMES_KANBAN_TASK grants access (the new branch) - HERMES_KANBAN_TASK short-circuits before consulting session_context or gateway.status (so workers don't depend on those import paths being healthy) - HERMES_SESSION_PLATFORM=telegram grants access - HERMES_SESSION_PLATFORM=local falls through to gateway check - is_gateway_running()=True grants access - All signals absent → False - gateway.status ImportError is swallowed → False Pinning the short-circuit (test #2) is the load-bearing one — it documents the contract that worker-side availability cannot regress to depending on gateway-side state lookups. --- tests/tools/test_send_message_tool.py | 104 ++++++++++++++++++++++++++ 1 file changed, 104 insertions(+) diff --git a/tests/tools/test_send_message_tool.py b/tests/tools/test_send_message_tool.py index 024cf43f948..7077b6be88b 100644 --- a/tests/tools/test_send_message_tool.py +++ b/tests/tools/test_send_message_tool.py @@ -2229,3 +2229,107 @@ class TestSendViaAdapterStandaloneFallback: assert result["success"] is True assert result["message_id"] == "abc-123" assert result["extra_field"] == "preserved" + + +# --------------------------------------------------------------------------- +# _check_send_message — availability gating +# --------------------------------------------------------------------------- + +class TestCheckSendMessage: + """The tool's check_fn governs whether the model sees ``send_message`` as + callable for a given session. The four passing conditions are: + + 1. ``HERMES_KANBAN_TASK`` is set (worker spawned by the kanban dispatcher + — parent gateway is by definition running, but the worker's + ``HERMES_HOME`` may be a profile dir without a ``gateway.pid``). + 2. ``HERMES_SESSION_PLATFORM`` resolves to a non-empty, non-``local`` value + (the session is wired to a messaging platform like Telegram). + 3. ``is_gateway_running()`` returns True (CLI / orchestrator profile with + a live gateway colocated under the same ``HERMES_HOME``). + 4. None of the above → False, tool is hidden. + """ + + def test_kanban_task_env_grants_access(self, monkeypatch): + """Workers spawned by the dispatcher (HERMES_KANBAN_TASK set) must be + allowed regardless of session_platform / gateway-pid state.""" + from tools.send_message_tool import _check_send_message + + monkeypatch.setenv("HERMES_KANBAN_TASK", "t_abc12345") + monkeypatch.delenv("HERMES_SESSION_PLATFORM", raising=False) + + with patch("gateway.session_context.get_session_env", return_value=""), \ + patch("gateway.status.is_gateway_running", return_value=False): + assert _check_send_message() is True + + def test_kanban_task_env_short_circuits_before_gateway_check(self, monkeypatch): + """Honoring HERMES_KANBAN_TASK must not depend on importing or calling + gateway.status — the worker may run with a HERMES_HOME that has no + gateway.pid, and we don't want that import path to be load-bearing.""" + from tools.send_message_tool import _check_send_message + + monkeypatch.setenv("HERMES_KANBAN_TASK", "t_abc12345") + + sentinel = object() + with patch("gateway.session_context.get_session_env", + side_effect=AssertionError("session_context not consulted " + "when HERMES_KANBAN_TASK is set")), \ + patch("gateway.status.is_gateway_running", + side_effect=AssertionError("gateway.status not consulted " + "when HERMES_KANBAN_TASK is set")): + assert _check_send_message() is True + + def test_messaging_platform_session_grants_access(self, monkeypatch): + """Telegram/Discord/etc. sessions pass via the platform branch even + without HERMES_KANBAN_TASK.""" + from tools.send_message_tool import _check_send_message + + monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False) + + with patch("gateway.session_context.get_session_env", return_value="telegram"), \ + patch("gateway.status.is_gateway_running", return_value=False): + assert _check_send_message() is True + + def test_local_platform_falls_through_to_gateway_check(self, monkeypatch): + """``HERMES_SESSION_PLATFORM=local`` means CLI-style — must defer to + is_gateway_running() rather than auto-grant.""" + from tools.send_message_tool import _check_send_message + + monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False) + + with patch("gateway.session_context.get_session_env", return_value="local"), \ + patch("gateway.status.is_gateway_running", return_value=True) as gw_mock: + assert _check_send_message() is True + gw_mock.assert_called_once() + + def test_running_gateway_grants_access(self, monkeypatch): + """Plain CLI session (no kanban task, empty platform) with a live + gateway: tool is callable.""" + from tools.send_message_tool import _check_send_message + + monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False) + + with patch("gateway.session_context.get_session_env", return_value=""), \ + patch("gateway.status.is_gateway_running", return_value=True): + assert _check_send_message() is True + + def test_no_signals_means_unavailable(self, monkeypatch): + """No kanban task, no platform, no gateway: tool is hidden.""" + from tools.send_message_tool import _check_send_message + + monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False) + + with patch("gateway.session_context.get_session_env", return_value=""), \ + patch("gateway.status.is_gateway_running", return_value=False): + assert _check_send_message() is False + + def test_gateway_status_import_error_is_swallowed(self, monkeypatch): + """If gateway.status can't be imported (unusual deployment / partial + install), the check returns False rather than raising.""" + from tools.send_message_tool import _check_send_message + + monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False) + + with patch("gateway.session_context.get_session_env", return_value=""), \ + patch("gateway.status.is_gateway_running", + side_effect=ImportError("simulated")): + assert _check_send_message() is False