RomanNum3ral
/
hermes-agent
mirror of https://github.com/NousResearch/hermes-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

test(send_message): move plugin fallback regressions to target_parse suite 

Keep opaque-plugin routing coverage in the lightweight suite so it still
runs when optional telegram deps are absent. Address PR review feedback.
This commit is contained in:
Mark Mennell 2026-08-01 13:43:14 +10:00 committed by Teknium
parent a009b71766
commit b58c7aaf9d
2 changed files with 3 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
tests/tools/test_send_message_target_parse.py
View File

@ -66,7 +66,7 @@ def test_send_message_routes_whatsapp_group_jid_without_home_fallback() -> None:
)
def test_resolved_opaque_plugin_target_uses_directory_id():
def test_resolved_opaque_plugin_target_uses_directory_id() -> None:
from gateway.platform_registry import PlatformEntry, platform_registry
platform_name = "opaque-resolved-test"
@ -121,7 +121,7 @@ def test_resolved_opaque_plugin_target_uses_directory_id():
)
def test_unresolved_opaque_plugin_target_passes_through_verbatim():
def test_unresolved_opaque_plugin_target_passes_through_verbatim() -> None:
from gateway.platform_registry import PlatformEntry, platform_registry
platform_name = "opaque-verbatim-test"
@ -185,7 +185,7 @@ def test_unresolved_opaque_plugin_target_passes_through_verbatim():
)
def test_unresolved_builtin_target_keeps_directory_error():
def test_unresolved_builtin_target_keeps_directory_error() -> None:
telegram_cfg = SimpleNamespace(enabled=True, token="***", extra={})
config = SimpleNamespace(
platforms={Platform.TELEGRAM: telegram_cfg},

140
tests/tools/test_send_message_tool.py
View File

@ -275,146 +275,6 @@ def _ensure_slack_mock(monkeypatch):
class TestSendMessageTool:
def test_resolved_opaque_plugin_target_uses_directory_id(self):
from gateway.platform_registry import PlatformEntry, platform_registry
platform_name = "opaque-resolved-test"
entry = PlatformEntry(
name=platform_name,
label="Opaque resolved test",
adapter_factory=lambda cfg: None,
check_fn=lambda: True,
)
platform_registry.register(entry)
platform = Platform(platform_name)
pconfig = SimpleNamespace(enabled=True, token=None, extra={})
config = SimpleNamespace(
platforms={platform: pconfig},
get_home_channel=lambda _platform: None,
)
try:
with patch("gateway.config.load_gateway_config", return_value=config), \
patch("tools.interrupt.is_interrupted", return_value=False), \
patch(
"gateway.channel_directory.resolve_channel_name",
return_value="opaque:directory-id",
), \
patch("model_tools._run_async", side_effect=_run_async_immediately), \
patch(
"tools.send_message_tool._send_to_platform",
new=AsyncMock(return_value={"success": True}),
) as send_mock, \
patch("gateway.mirror.mirror_to_session", return_value=True):
result = json.loads(
send_message_tool(
{
"action": "send",
"target": f"{platform_name}:Friendly name",
"message": "hello",
}
)
)
finally:
platform_registry.unregister(platform_name)
assert result["success"] is True
send_mock.assert_awaited_once_with(
platform,
pconfig,
"opaque:directory-id",
"hello",
thread_id=None,
media_files=[],
force_document=False,
)
def test_unresolved_opaque_plugin_target_passes_through_verbatim(self):
from gateway.platform_registry import PlatformEntry, platform_registry
platform_name = "opaque-verbatim-test"
entry = PlatformEntry(
name=platform_name,
label="Opaque verbatim test",
adapter_factory=lambda cfg: None,
check_fn=lambda: True,
)
platform_registry.register(entry)
platform = Platform(platform_name)
# Simulate a fresh `hermes send` process: the dynamic Platform member
# is known from config, but plugin discovery has not registered its
# adapter entry yet.
platform_registry.unregister(platform_name)
pconfig = SimpleNamespace(enabled=True, token=None, extra={})
config = SimpleNamespace(
platforms={platform: pconfig},
get_home_channel=lambda _platform: None,
)
try:
with patch("gateway.config.load_gateway_config", return_value=config), \
patch("tools.interrupt.is_interrupted", return_value=False), \
patch(
"gateway.channel_directory.resolve_channel_name",
return_value=None,
), \
patch(
"hermes_cli.plugins.discover_plugins",
side_effect=lambda: platform_registry.register(entry),
) as discover_mock, \
patch("model_tools._run_async", side_effect=_run_async_immediately), \
patch(
"tools.send_message_tool._send_to_platform",
new=AsyncMock(return_value={"success": True}),
) as send_mock, \
patch("gateway.mirror.mirror_to_session", return_value=True):
result = json.loads(
send_message_tool(
{
"action": "send",
"target": f"{platform_name}:dm:panyaozhen",
"message": "hello",
}
)
)
finally:
platform_registry.unregister(platform_name)
assert result["success"] is True
discover_mock.assert_called_once_with()
send_mock.assert_awaited_once_with(
platform,
pconfig,
"dm:panyaozhen",
"hello",
thread_id=None,
media_files=[],
force_document=False,
)
def test_unresolved_builtin_target_keeps_directory_error(self):
config, _telegram_cfg = _make_config()
with patch("gateway.config.load_gateway_config", return_value=config), \
patch("gateway.channel_directory.resolve_channel_name", return_value=None), \
patch(
"tools.send_message_tool._send_to_platform",
new=AsyncMock(return_value={"success": True}),
) as send_mock:
result = json.loads(
send_message_tool(
{
"action": "send",
"target": "telegram:missing-room",
"message": "hello",
}
)
)
assert result == {
"error": "Could not resolve 'missing-room' on telegram. "
"Use send_message(action='list') to see available targets."
}
send_mock.assert_not_awaited()
def test_ntfy_topic_target_is_explicit(self):
chat_id, thread_id, is_explicit = _parse_target_ref("ntfy", "alerts-channel")