mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-05-06 10:47:12 +08:00
6b76ea4707
The YAML-to-env-var bridge in load_gateway_config() mapped every Discord and Telegram config key (require_mention, auto_thread, reactions, etc.) except reply_to_mode. Users setting discord.reply_to_mode or telegram.reply_to_mode in ~/.hermes/config.yaml got no effect — the adapter only read the env var, which nothing populated from YAML. Add the missing bridge for both platforms, following the existing pattern. Top-level <platform>.reply_to_mode preferred, falls back to <platform>.extra.reply_to_mode, env var never overwritten. Handles YAML 1.1 bare `off` → Python False coercion. This is a re-submission of the work from #9837 and #13930, which both implemented the same fix but neither landed (see co-authors below). Co-authored-by: Matteo De Agazio <hypnosis.mda@gmail.com> Co-authored-by: ishardo <239075732+ishardo@users.noreply.github.com>
307 lines
13 KiB
Python
307 lines
13 KiB
Python
"""Tests for Telegram reply_to_mode functionality.
|
|
|
|
Covers the threading behavior control for multi-chunk replies:
|
|
- "off": Never thread replies to original message
|
|
- "first": Only first chunk threads (default)
|
|
- "all": All chunks thread to original message
|
|
"""
|
|
import os
|
|
import sys
|
|
from unittest.mock import MagicMock, AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from gateway.config import PlatformConfig, GatewayConfig, Platform, _apply_env_overrides, load_gateway_config
|
|
|
|
|
|
def _ensure_telegram_mock():
|
|
"""Mock the telegram package if it's not installed."""
|
|
if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"):
|
|
return
|
|
mod = MagicMock()
|
|
mod.ext.ContextTypes.DEFAULT_TYPE = type(None)
|
|
mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2"
|
|
mod.constants.ChatType.GROUP = "group"
|
|
mod.constants.ChatType.SUPERGROUP = "supergroup"
|
|
mod.constants.ChatType.CHANNEL = "channel"
|
|
mod.constants.ChatType.PRIVATE = "private"
|
|
for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"):
|
|
sys.modules.setdefault(name, mod)
|
|
|
|
|
|
_ensure_telegram_mock()
|
|
|
|
from gateway.platforms.telegram import TelegramAdapter # noqa: E402
|
|
|
|
|
|
@pytest.fixture()
|
|
def adapter_factory():
|
|
"""Factory to create TelegramAdapter with custom reply_to_mode."""
|
|
def create(reply_to_mode: str = "first"):
|
|
config = PlatformConfig(enabled=True, token="test-token", reply_to_mode=reply_to_mode)
|
|
return TelegramAdapter(config)
|
|
return create
|
|
|
|
|
|
class TestReplyToModeConfig:
|
|
"""Tests for reply_to_mode configuration loading."""
|
|
|
|
def test_default_mode_is_first(self, adapter_factory):
|
|
adapter = adapter_factory()
|
|
assert adapter._reply_to_mode == "first"
|
|
|
|
def test_off_mode(self, adapter_factory):
|
|
adapter = adapter_factory(reply_to_mode="off")
|
|
assert adapter._reply_to_mode == "off"
|
|
|
|
def test_first_mode(self, adapter_factory):
|
|
adapter = adapter_factory(reply_to_mode="first")
|
|
assert adapter._reply_to_mode == "first"
|
|
|
|
def test_all_mode(self, adapter_factory):
|
|
adapter = adapter_factory(reply_to_mode="all")
|
|
assert adapter._reply_to_mode == "all"
|
|
|
|
def test_invalid_mode_stored_as_is(self, adapter_factory):
|
|
"""Invalid modes are stored but _should_thread_reply handles them."""
|
|
adapter = adapter_factory(reply_to_mode="invalid")
|
|
assert adapter._reply_to_mode == "invalid"
|
|
|
|
def test_none_mode_defaults_to_first(self):
|
|
config = PlatformConfig(enabled=True, token="test-token")
|
|
adapter = TelegramAdapter(config)
|
|
assert adapter._reply_to_mode == "first"
|
|
|
|
def test_empty_string_mode_defaults_to_first(self):
|
|
config = PlatformConfig(enabled=True, token="test-token", reply_to_mode="")
|
|
adapter = TelegramAdapter(config)
|
|
assert adapter._reply_to_mode == "first"
|
|
|
|
|
|
class TestShouldThreadReply:
|
|
"""Tests for _should_thread_reply method."""
|
|
|
|
def test_no_reply_to_returns_false(self, adapter_factory):
|
|
adapter = adapter_factory(reply_to_mode="first")
|
|
assert adapter._should_thread_reply(None, 0) is False
|
|
assert adapter._should_thread_reply("", 0) is False
|
|
|
|
def test_off_mode_never_threads(self, adapter_factory):
|
|
adapter = adapter_factory(reply_to_mode="off")
|
|
assert adapter._should_thread_reply("msg-123", 0) is False
|
|
assert adapter._should_thread_reply("msg-123", 1) is False
|
|
assert adapter._should_thread_reply("msg-123", 5) is False
|
|
|
|
def test_first_mode_only_first_chunk(self, adapter_factory):
|
|
adapter = adapter_factory(reply_to_mode="first")
|
|
assert adapter._should_thread_reply("msg-123", 0) is True
|
|
assert adapter._should_thread_reply("msg-123", 1) is False
|
|
assert adapter._should_thread_reply("msg-123", 2) is False
|
|
assert adapter._should_thread_reply("msg-123", 10) is False
|
|
|
|
def test_all_mode_all_chunks(self, adapter_factory):
|
|
adapter = adapter_factory(reply_to_mode="all")
|
|
assert adapter._should_thread_reply("msg-123", 0) is True
|
|
assert adapter._should_thread_reply("msg-123", 1) is True
|
|
assert adapter._should_thread_reply("msg-123", 2) is True
|
|
assert adapter._should_thread_reply("msg-123", 10) is True
|
|
|
|
def test_invalid_mode_falls_back_to_first(self, adapter_factory):
|
|
"""Invalid mode behaves like 'first' - only first chunk threads."""
|
|
adapter = adapter_factory(reply_to_mode="invalid")
|
|
assert adapter._should_thread_reply("msg-123", 0) is True
|
|
assert adapter._should_thread_reply("msg-123", 1) is False
|
|
|
|
|
|
class TestSendWithReplyToMode:
|
|
"""Tests for send() method respecting reply_to_mode."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_off_mode_no_reply_threading(self, adapter_factory):
|
|
adapter = adapter_factory(reply_to_mode="off")
|
|
adapter._bot = MagicMock()
|
|
adapter._bot.send_message = AsyncMock(return_value=MagicMock(message_id=1))
|
|
adapter.truncate_message = lambda content, max_len, **kw: ["chunk1", "chunk2", "chunk3"]
|
|
|
|
await adapter.send("12345", "test content", reply_to="999")
|
|
|
|
for call in adapter._bot.send_message.call_args_list:
|
|
assert call.kwargs.get("reply_to_message_id") is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_first_mode_only_first_chunk_threads(self, adapter_factory):
|
|
adapter = adapter_factory(reply_to_mode="first")
|
|
adapter._bot = MagicMock()
|
|
adapter._bot.send_message = AsyncMock(return_value=MagicMock(message_id=1))
|
|
adapter.truncate_message = lambda content, max_len, **kw: ["chunk1", "chunk2", "chunk3"]
|
|
|
|
await adapter.send("12345", "test content", reply_to="999")
|
|
|
|
calls = adapter._bot.send_message.call_args_list
|
|
assert len(calls) == 3
|
|
assert calls[0].kwargs.get("reply_to_message_id") == 999
|
|
assert calls[1].kwargs.get("reply_to_message_id") is None
|
|
assert calls[2].kwargs.get("reply_to_message_id") is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_all_mode_all_chunks_thread(self, adapter_factory):
|
|
adapter = adapter_factory(reply_to_mode="all")
|
|
adapter._bot = MagicMock()
|
|
adapter._bot.send_message = AsyncMock(return_value=MagicMock(message_id=1))
|
|
adapter.truncate_message = lambda content, max_len, **kw: ["chunk1", "chunk2", "chunk3"]
|
|
|
|
await adapter.send("12345", "test content", reply_to="999")
|
|
|
|
calls = adapter._bot.send_message.call_args_list
|
|
assert len(calls) == 3
|
|
for call in calls:
|
|
assert call.kwargs.get("reply_to_message_id") == 999
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_reply_to_param_no_threading(self, adapter_factory):
|
|
adapter = adapter_factory(reply_to_mode="all")
|
|
adapter._bot = MagicMock()
|
|
adapter._bot.send_message = AsyncMock(return_value=MagicMock(message_id=1))
|
|
adapter.truncate_message = lambda content, max_len, **kw: ["chunk1", "chunk2"]
|
|
|
|
await adapter.send("12345", "test content", reply_to=None)
|
|
|
|
calls = adapter._bot.send_message.call_args_list
|
|
for call in calls:
|
|
assert call.kwargs.get("reply_to_message_id") is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_single_chunk_respects_mode(self, adapter_factory):
|
|
adapter = adapter_factory(reply_to_mode="first")
|
|
adapter._bot = MagicMock()
|
|
adapter._bot.send_message = AsyncMock(return_value=MagicMock(message_id=1))
|
|
adapter.truncate_message = lambda content, max_len, **kw: ["single chunk"]
|
|
|
|
await adapter.send("12345", "test", reply_to="999")
|
|
|
|
calls = adapter._bot.send_message.call_args_list
|
|
assert len(calls) == 1
|
|
assert calls[0].kwargs.get("reply_to_message_id") == 999
|
|
|
|
|
|
class TestConfigSerialization:
|
|
"""Tests for reply_to_mode serialization."""
|
|
|
|
def test_to_dict_includes_reply_to_mode(self):
|
|
config = PlatformConfig(enabled=True, token="test", reply_to_mode="all")
|
|
result = config.to_dict()
|
|
assert result["reply_to_mode"] == "all"
|
|
|
|
def test_from_dict_loads_reply_to_mode(self):
|
|
data = {"enabled": True, "token": "test", "reply_to_mode": "off"}
|
|
config = PlatformConfig.from_dict(data)
|
|
assert config.reply_to_mode == "off"
|
|
|
|
def test_from_dict_defaults_to_first(self):
|
|
data = {"enabled": True, "token": "test"}
|
|
config = PlatformConfig.from_dict(data)
|
|
assert config.reply_to_mode == "first"
|
|
|
|
|
|
class TestEnvVarOverride:
|
|
"""Tests for TELEGRAM_REPLY_TO_MODE environment variable override."""
|
|
|
|
def _make_config(self):
|
|
config = GatewayConfig()
|
|
config.platforms[Platform.TELEGRAM] = PlatformConfig(enabled=True, token="test")
|
|
return config
|
|
|
|
def test_env_var_sets_off_mode(self):
|
|
config = self._make_config()
|
|
with patch.dict(os.environ, {"TELEGRAM_REPLY_TO_MODE": "off"}, clear=False):
|
|
_apply_env_overrides(config)
|
|
assert config.platforms[Platform.TELEGRAM].reply_to_mode == "off"
|
|
|
|
def test_env_var_sets_all_mode(self):
|
|
config = self._make_config()
|
|
with patch.dict(os.environ, {"TELEGRAM_REPLY_TO_MODE": "all"}, clear=False):
|
|
_apply_env_overrides(config)
|
|
assert config.platforms[Platform.TELEGRAM].reply_to_mode == "all"
|
|
|
|
def test_env_var_case_insensitive(self):
|
|
config = self._make_config()
|
|
with patch.dict(os.environ, {"TELEGRAM_REPLY_TO_MODE": "ALL"}, clear=False):
|
|
_apply_env_overrides(config)
|
|
assert config.platforms[Platform.TELEGRAM].reply_to_mode == "all"
|
|
|
|
def test_env_var_invalid_value_ignored(self):
|
|
config = self._make_config()
|
|
with patch.dict(os.environ, {"TELEGRAM_REPLY_TO_MODE": "banana"}, clear=False):
|
|
_apply_env_overrides(config)
|
|
assert config.platforms[Platform.TELEGRAM].reply_to_mode == "first"
|
|
|
|
def test_env_var_empty_value_ignored(self):
|
|
config = self._make_config()
|
|
with patch.dict(os.environ, {"TELEGRAM_REPLY_TO_MODE": ""}, clear=False):
|
|
_apply_env_overrides(config)
|
|
assert config.platforms[Platform.TELEGRAM].reply_to_mode == "first"
|
|
|
|
|
|
class TestTelegramYamlConfigLoading:
|
|
"""Tests for reply_to_mode loaded from config.yaml telegram section."""
|
|
|
|
def _write_config(self, tmp_path, content: str):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(content, encoding="utf-8")
|
|
return hermes_home
|
|
|
|
def test_top_level_reply_to_mode_off(self, tmp_path, monkeypatch):
|
|
"""YAML 1.1 parses bare 'off' as boolean False — must map back to 'off'."""
|
|
hermes_home = self._write_config(tmp_path, "telegram:\n reply_to_mode: off\n")
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("TELEGRAM_REPLY_TO_MODE", raising=False)
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.environ.get("TELEGRAM_REPLY_TO_MODE") == "off"
|
|
|
|
def test_top_level_reply_to_mode_all(self, tmp_path, monkeypatch):
|
|
hermes_home = self._write_config(tmp_path, "telegram:\n reply_to_mode: all\n")
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("TELEGRAM_REPLY_TO_MODE", raising=False)
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.environ.get("TELEGRAM_REPLY_TO_MODE") == "all"
|
|
|
|
def test_extra_reply_to_mode_off(self, tmp_path, monkeypatch):
|
|
"""telegram.extra.reply_to_mode is also honoured."""
|
|
hermes_home = self._write_config(
|
|
tmp_path, "telegram:\n extra:\n reply_to_mode: \"off\"\n"
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("TELEGRAM_REPLY_TO_MODE", raising=False)
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.environ.get("TELEGRAM_REPLY_TO_MODE") == "off"
|
|
|
|
def test_env_var_takes_precedence_over_yaml(self, tmp_path, monkeypatch):
|
|
"""Existing TELEGRAM_REPLY_TO_MODE env var is not overwritten by YAML."""
|
|
hermes_home = self._write_config(tmp_path, "telegram:\n reply_to_mode: all\n")
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.setenv("TELEGRAM_REPLY_TO_MODE", "first")
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.environ.get("TELEGRAM_REPLY_TO_MODE") == "first"
|
|
|
|
def test_top_level_takes_precedence_over_extra(self, tmp_path, monkeypatch):
|
|
"""telegram.reply_to_mode wins over telegram.extra.reply_to_mode."""
|
|
hermes_home = self._write_config(
|
|
tmp_path,
|
|
"telegram:\n reply_to_mode: all\n extra:\n reply_to_mode: \"off\"\n",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("TELEGRAM_REPLY_TO_MODE", raising=False)
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.environ.get("TELEGRAM_REPLY_TO_MODE") == "all"
|