test(send_message): cover _check_send_message gating paths

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.
This commit is contained in:
Dominikh 2026-05-10 18:37:11 +00:00 committed by kshitij
parent 8ac998cb0c
commit 379e7dd014

View file

@ -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