common/hermes
2
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Merge branch 'main' of github.com:NousResearch/hermes-agent into feat/ink-refactor

Browse Source
This commit is contained in:
Brooklyn Nicholson
2026-04-15 10:21:00 -05:00
parent 33c615504d 722331a57d
commit 371166fe26
35 changed files with 2388 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
tests/cron/test_scheduler.py
View File

@@ -233,9 +233,10 @@ class TestDeliverResultWrapping:
send_mock.assert_called_once()
sent_content = send_mock.call_args.kwargs.get("content") or send_mock.call_args[0][-1]
assert "Cronjob Response: daily-report" in sent_content
assert "(job_id: test-job)" in sent_content
assert "-------------" in sent_content
assert "Here is today's summary." in sent_content
assert "The agent cannot see this message" in sent_content
assert "To stop or manage this job" in sent_content
def test_delivery_uses_job_id_when_no_name(self):
"""When a job has no name, the wrapper should fall back to job id."""

293
tests/gateway/test_busy_session_ack.py Normal file
View File

@@ -0,0 +1,293 @@
"""Tests for busy-session acknowledgment when user sends messages during active agent runs.
Verifies that users get an immediate status response instead of total silence
when the agent is working on a task. See PR fix for the @Lonely__MH report.
"""
import asyncio
import time
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
# ---------------------------------------------------------------------------
# Minimal stubs so we can import gateway code without heavy deps
# ---------------------------------------------------------------------------
import sys, types
_tg = types.ModuleType("telegram")
_tg.constants = types.ModuleType("telegram.constants")
_ct = MagicMock()
_ct.SUPERGROUP = "supergroup"
_ct.GROUP = "group"
_ct.PRIVATE = "private"
_tg.constants.ChatType = _ct
sys.modules.setdefault("telegram", _tg)
sys.modules.setdefault("telegram.constants", _tg.constants)
sys.modules.setdefault("telegram.ext", types.ModuleType("telegram.ext"))
from gateway.platforms.base import (
BasePlatformAdapter,
MessageEvent,
MessageType,
SessionSource,
build_session_key,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_event(text="hello", chat_id="123", platform_val="telegram"):
"""Build a minimal MessageEvent."""
source = SessionSource(
platform=MagicMock(value=platform_val),
chat_id=chat_id,
chat_type="private",
user_id="user1",
)
evt = MessageEvent(
text=text,
message_type=MessageType.TEXT,
source=source,
message_id="msg1",
)
return evt
def _make_runner():
"""Build a minimal GatewayRunner-like object for testing."""
from gateway.run import GatewayRunner, _AGENT_PENDING_SENTINEL
runner = object.__new__(GatewayRunner)
runner._running_agents = {}
runner._running_agents_ts = {}
runner._pending_messages = {}
runner._busy_ack_ts = {}
runner._draining = False
runner.adapters = {}
runner.config = MagicMock()
runner.session_store = None
runner.hooks = MagicMock()
runner.hooks.emit = AsyncMock()
return runner, _AGENT_PENDING_SENTINEL
def _make_adapter(platform_val="telegram"):
"""Build a minimal adapter mock."""
adapter = MagicMock()
adapter._pending_messages = {}
adapter._send_with_retry = AsyncMock()
adapter.config = MagicMock()
adapter.config.extra = {}
adapter.platform = MagicMock(value=platform_val)
return adapter
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
class TestBusySessionAck:
"""User sends a message while agent is running — should get acknowledgment."""
@pytest.mark.asyncio
async def test_sends_ack_when_agent_running(self):
"""First message during busy session should get a status ack."""
runner, sentinel = _make_runner()
adapter = _make_adapter()
event = _make_event(text="Are you working?")
sk = build_session_key(event.source)
# Simulate running agent
agent = MagicMock()
agent.get_activity_summary.return_value = {
"api_call_count": 21,
"max_iterations": 60,
"current_tool": "terminal",
"last_activity_ts": time.time(),
"last_activity_desc": "terminal",
"seconds_since_activity": 1.0,
}
runner._running_agents[sk] = agent
runner._running_agents_ts[sk] = time.time() - 600 # 10 min ago
runner.adapters[event.source.platform] = adapter
result = await runner._handle_active_session_busy_message(event, sk)
assert result is True # handled
# Verify ack was sent
adapter._send_with_retry.assert_called_once()
call_kwargs = adapter._send_with_retry.call_args
content = call_kwargs.kwargs.get("content") or call_kwargs[1].get("content", "")
if not content and call_kwargs.args:
# positional args
content = str(call_kwargs)
assert "Interrupting" in content or "respond" in content
assert "/stop" not in content # no need — we ARE interrupting
# Verify message was queued in adapter pending
assert sk in adapter._pending_messages
# Verify agent interrupt was called
agent.interrupt.assert_called_once_with("Are you working?")
@pytest.mark.asyncio
async def test_debounce_suppresses_rapid_acks(self):
"""Second message within 30s should NOT send another ack."""
runner, sentinel = _make_runner()
adapter = _make_adapter()
event1 = _make_event(text="hello?")
# Reuse the same source so platform mock matches
event2 = MessageEvent(
text="still there?",
message_type=MessageType.TEXT,
source=event1.source,
message_id="msg2",
)
sk = build_session_key(event1.source)
agent = MagicMock()
agent.get_activity_summary.return_value = {
"api_call_count": 5,
"max_iterations": 60,
"current_tool": None,
"last_activity_ts": time.time(),
"last_activity_desc": "api_call",
"seconds_since_activity": 0.5,
}
runner._running_agents[sk] = agent
runner._running_agents_ts[sk] = time.time() - 60
runner.adapters[event1.source.platform] = adapter
# First message — should get ack
result1 = await runner._handle_active_session_busy_message(event1, sk)
assert result1 is True
assert adapter._send_with_retry.call_count == 1
# Second message within cooldown — should be queued but no ack
result2 = await runner._handle_active_session_busy_message(event2, sk)
assert result2 is True
assert adapter._send_with_retry.call_count == 1 # still 1, no new ack
# But interrupt should still be called for both
assert agent.interrupt.call_count == 2
@pytest.mark.asyncio
async def test_ack_after_cooldown_expires(self):
"""After 30s cooldown, a new message should send a fresh ack."""
runner, sentinel = _make_runner()
adapter = _make_adapter()
event = _make_event(text="hello?")
sk = build_session_key(event.source)
agent = MagicMock()
agent.get_activity_summary.return_value = {
"api_call_count": 10,
"max_iterations": 60,
"current_tool": "web_search",
"last_activity_ts": time.time(),
"last_activity_desc": "tool",
"seconds_since_activity": 0.5,
}
runner._running_agents[sk] = agent
runner._running_agents_ts[sk] = time.time() - 120
runner.adapters[event.source.platform] = adapter
# First ack
await runner._handle_active_session_busy_message(event, sk)
assert adapter._send_with_retry.call_count == 1
# Fake that cooldown expired
runner._busy_ack_ts[sk] = time.time() - 31
# Second ack should go through
await runner._handle_active_session_busy_message(event, sk)
assert adapter._send_with_retry.call_count == 2
@pytest.mark.asyncio
async def test_includes_status_detail(self):
"""Ack message should include iteration and tool info when available."""
runner, sentinel = _make_runner()
adapter = _make_adapter()
event = _make_event(text="yo")
sk = build_session_key(event.source)
agent = MagicMock()
agent.get_activity_summary.return_value = {
"api_call_count": 21,
"max_iterations": 60,
"current_tool": "terminal",
"last_activity_ts": time.time(),
"last_activity_desc": "terminal",
"seconds_since_activity": 0.5,
}
runner._running_agents[sk] = agent
runner._running_agents_ts[sk] = time.time() - 600 # 10 min
runner.adapters[event.source.platform] = adapter
await runner._handle_active_session_busy_message(event, sk)
call_kwargs = adapter._send_with_retry.call_args
content = call_kwargs.kwargs.get("content", "")
assert "21/60" in content # iteration
assert "terminal" in content # current tool
assert "10 min" in content # elapsed
@pytest.mark.asyncio
async def test_draining_still_works(self):
"""Draining case should still produce the drain-specific message."""
runner, sentinel = _make_runner()
runner._draining = True
adapter = _make_adapter()
event = _make_event(text="hello")
sk = build_session_key(event.source)
runner.adapters[event.source.platform] = adapter
# Mock the drain-specific methods
runner._queue_during_drain_enabled = lambda: False
runner._status_action_gerund = lambda: "restarting"
result = await runner._handle_active_session_busy_message(event, sk)
assert result is True
call_kwargs = adapter._send_with_retry.call_args
content = call_kwargs.kwargs.get("content", "")
assert "restarting" in content
@pytest.mark.asyncio
async def test_pending_sentinel_no_interrupt(self):
"""When agent is PENDING_SENTINEL, don't call interrupt (it has no method)."""
runner, sentinel = _make_runner()
adapter = _make_adapter()
event = _make_event(text="hey")
sk = build_session_key(event.source)
runner._running_agents[sk] = sentinel
runner._running_agents_ts[sk] = time.time()
runner.adapters[event.source.platform] = adapter
result = await runner._handle_active_session_busy_message(event, sk)
assert result is True
# Should still send ack
adapter._send_with_retry.assert_called_once()
@pytest.mark.asyncio
async def test_no_adapter_falls_through(self):
"""If adapter is missing, return False so default path handles it."""
runner, sentinel = _make_runner()
event = _make_event(text="hello")
sk = build_session_key(event.source)
# No adapter registered
runner._running_agents[sk] = MagicMock()
result = await runner._handle_active_session_busy_message(event, sk)
assert result is False # not handled, let default path try

291
tests/gateway/test_duplicate_reply_suppression.py Normal file
View File

@@ -0,0 +1,291 @@
"""Tests for duplicate reply suppression across the gateway stack.
Covers three fix paths:
1. base.py: stale response suppressed when interrupt_event is set and a
pending message exists (#8221 / #2483)
2. run.py return path: already_sent propagated from stream consumer's
already_sent flag without requiring response_previewed (#8375)
3. run.py queued-message path: first response correctly detected as
already-streamed when already_sent is True without response_previewed
"""
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import (
BasePlatformAdapter,
MessageEvent,
MessageType,
ProcessingOutcome,
SendResult,
)
from gateway.session import SessionSource, build_session_key
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
class StubAdapter(BasePlatformAdapter):
"""Minimal concrete adapter for testing."""
def __init__(self):
super().__init__(PlatformConfig(enabled=True, token="fake"), Platform.DISCORD)
self.sent = []
async def connect(self):
return True
async def disconnect(self):
pass
async def send(self, chat_id, content, reply_to=None, metadata=None):
self.sent.append({"chat_id": chat_id, "content": content})
return SendResult(success=True, message_id="msg1")
async def send_typing(self, chat_id, metadata=None):
pass
async def get_chat_info(self, chat_id):
return {"id": chat_id}
def _make_event(text="hello", chat_id="c1", user_id="u1"):
return MessageEvent(
text=text,
source=SessionSource(
platform=Platform.DISCORD,
chat_id=chat_id,
chat_type="dm",
user_id=user_id,
),
message_id="m1",
)
# ===================================================================
# Test 1: base.py — stale response suppressed on interrupt (#8221)
# ===================================================================
class TestBaseInterruptSuppression:
@pytest.mark.asyncio
async def test_stale_response_suppressed_when_interrupted(self):
"""When interrupt_event is set AND a pending message exists,
base.py should suppress the stale response instead of sending it."""
adapter = StubAdapter()
stale_response = "This is the stale answer to the first question."
pending_response = "This is the answer to the second question."
call_count = 0
async def fake_handler(event):
nonlocal call_count
call_count += 1
if call_count == 1:
return stale_response
return pending_response
adapter.set_message_handler(fake_handler)
event_a = _make_event(text="first question")
session_key = build_session_key(event_a.source)
# Simulate: message A is being processed, message B arrives
# The interrupt event is set and B is in pending_messages
interrupt_event = asyncio.Event()
interrupt_event.set()
adapter._active_sessions[session_key] = interrupt_event
event_b = _make_event(text="second question")
adapter._pending_messages[session_key] = event_b
await adapter._process_message_background(event_a, session_key)
# The stale response should NOT have been sent.
stale_sends = [s for s in adapter.sent if s["content"] == stale_response]
assert len(stale_sends) == 0, (
f"Stale response was sent {len(stale_sends)} time(s) — should be suppressed"
)
# The pending message's response SHOULD have been sent.
pending_sends = [s for s in adapter.sent if s["content"] == pending_response]
assert len(pending_sends) == 1, "Pending message response should be sent"
@pytest.mark.asyncio
async def test_response_not_suppressed_without_interrupt(self):
"""Normal case: no interrupt, response should be sent."""
adapter = StubAdapter()
async def fake_handler(event):
return "Normal response"
adapter.set_message_handler(fake_handler)
event = _make_event()
session_key = build_session_key(event.source)
await adapter._process_message_background(event, session_key)
assert any(s["content"] == "Normal response" for s in adapter.sent)
@pytest.mark.asyncio
async def test_response_not_suppressed_with_interrupt_but_no_pending(self):
"""Interrupt event set but no pending message (race already resolved) —
response should still be sent."""
adapter = StubAdapter()
async def fake_handler(event):
return "Valid response"
adapter.set_message_handler(fake_handler)
event = _make_event()
session_key = build_session_key(event.source)
# Set interrupt but no pending message
interrupt_event = asyncio.Event()
interrupt_event.set()
adapter._active_sessions[session_key] = interrupt_event
await adapter._process_message_background(event, session_key)
assert any(s["content"] == "Valid response" for s in adapter.sent)
# ===================================================================
# Test 2: run.py — already_sent without response_previewed (#8375)
# ===================================================================
class TestAlreadySentWithoutResponsePreviewed:
"""The already_sent flag on the response dict should be set when the
stream consumer's already_sent is True, even if response_previewed is
False. This prevents duplicate sends when streaming was interrupted
by flood control."""
def _make_mock_stream_consumer(self, already_sent=False, final_response_sent=False):
sc = SimpleNamespace(
already_sent=already_sent,
final_response_sent=final_response_sent,
)
return sc
def test_already_sent_set_without_response_previewed(self):
"""Stream consumer already_sent=True should propagate to response
dict even when response_previewed is False."""
sc = self._make_mock_stream_consumer(already_sent=True, final_response_sent=False)
response = {"final_response": "text", "response_previewed": False}
# Reproduce the logic from run.py return path (post-fix)
if sc and isinstance(response, dict) and not response.get("failed"):
if (
getattr(sc, "final_response_sent", False)
or getattr(sc, "already_sent", False)
):
response["already_sent"] = True
assert response.get("already_sent") is True
def test_already_sent_not_set_when_nothing_sent(self):
"""When stream consumer hasn't sent anything, already_sent should
not be set on the response."""
sc = self._make_mock_stream_consumer(already_sent=False, final_response_sent=False)
response = {"final_response": "text", "response_previewed": False}
if sc and isinstance(response, dict) and not response.get("failed"):
if (
getattr(sc, "final_response_sent", False)
or getattr(sc, "already_sent", False)
):
response["already_sent"] = True
assert "already_sent" not in response
def test_already_sent_set_on_final_response_sent(self):
"""final_response_sent=True should still work as before."""
sc = self._make_mock_stream_consumer(already_sent=False, final_response_sent=True)
response = {"final_response": "text"}
if sc and isinstance(response, dict) and not response.get("failed"):
if (
getattr(sc, "final_response_sent", False)
or getattr(sc, "already_sent", False)
):
response["already_sent"] = True
assert response.get("already_sent") is True
def test_already_sent_not_set_on_failed_response(self):
"""Failed responses should never be suppressed — user needs to see
the error message even if streaming sent earlier partial output."""
sc = self._make_mock_stream_consumer(already_sent=True, final_response_sent=False)
response = {"final_response": "Error: something broke", "failed": True}
if sc and isinstance(response, dict) and not response.get("failed"):
if (
getattr(sc, "final_response_sent", False)
or getattr(sc, "already_sent", False)
):
response["already_sent"] = True
assert "already_sent" not in response
# ===================================================================
# Test 3: run.py queued-message path — _already_streamed detection
# ===================================================================
class TestQueuedMessageAlreadyStreamed:
"""The queued-message path should detect that the first response was
already streamed (already_sent=True) even without response_previewed."""
def _make_mock_sc(self, already_sent=False, final_response_sent=False):
return SimpleNamespace(
already_sent=already_sent,
final_response_sent=final_response_sent,
)
def test_queued_path_detects_already_streamed(self):
"""already_sent=True on stream consumer means first response was
streamed — skip re-sending before processing queued message."""
_sc = self._make_mock_sc(already_sent=True)
# Reproduce the queued-message logic from run.py (post-fix)
_already_streamed = bool(
_sc
and (
getattr(_sc, "final_response_sent", False)
or getattr(_sc, "already_sent", False)
)
)
assert _already_streamed is True
def test_queued_path_sends_when_not_streamed(self):
"""Nothing was streamed — first response should be sent before
processing the queued message."""
_sc = self._make_mock_sc(already_sent=False)
_already_streamed = bool(
_sc
and (
getattr(_sc, "final_response_sent", False)
or getattr(_sc, "already_sent", False)
)
)
assert _already_streamed is False
def test_queued_path_with_no_stream_consumer(self):
"""No stream consumer at all (streaming disabled) — not streamed."""
_sc = None
_already_streamed = bool(
_sc
and (
getattr(_sc, "final_response_sent", False)
or getattr(_sc, "already_sent", False)
)
)
assert _already_streamed is False

24
tests/gateway/test_matrix.py
View File

@@ -335,6 +335,29 @@ def _make_adapter():
return adapter
# ---------------------------------------------------------------------------
# Typing indicator
# ---------------------------------------------------------------------------
class TestMatrixTypingIndicator:
def setup_method(self):
self.adapter = _make_adapter()
self.adapter._client = MagicMock()
self.adapter._client.set_typing = AsyncMock()
@pytest.mark.asyncio
async def test_stop_typing_clears_matrix_typing_state(self):
"""stop_typing() should send typing=false instead of waiting for timeout expiry."""
from gateway.platforms.matrix import RoomID
await self.adapter.stop_typing("!room:example.org")
self.adapter._client.set_typing.assert_awaited_once_with(
RoomID("!room:example.org"),
timeout=0,
)
# ---------------------------------------------------------------------------
# mxc:// URL conversion
# ---------------------------------------------------------------------------
@@ -1831,4 +1854,3 @@ class TestMatrixPresence:
assert result is False

275
tests/hermes_cli/test_doctor_command_install.py Normal file
View File

@@ -0,0 +1,275 @@
"""Tests for the Command Installation check in hermes doctor."""
import os
import sys
import types
from argparse import Namespace
from pathlib import Path
import pytest
import hermes_cli.doctor as doctor_mod
def _setup_doctor_env(monkeypatch, tmp_path, venv_name="venv"):
"""Create a minimal HERMES_HOME + PROJECT_ROOT for doctor tests."""
home = tmp_path / ".hermes"
home.mkdir(parents=True, exist_ok=True)
(home / "config.yaml").write_text("memory: {}\n", encoding="utf-8")
project = tmp_path / "project"
project.mkdir(exist_ok=True)
# Create a fake venv entry point
venv_bin_dir = project / venv_name / "bin"
venv_bin_dir.mkdir(parents=True, exist_ok=True)
hermes_bin = venv_bin_dir / "hermes"
hermes_bin.write_text("#!/usr/bin/env python\n# entry point\n")
hermes_bin.chmod(0o755)
monkeypatch.setattr(doctor_mod, "HERMES_HOME", home)
monkeypatch.setattr(doctor_mod, "PROJECT_ROOT", project)
monkeypatch.setattr(doctor_mod, "_DHH", str(home))
# Stub model_tools so doctor doesn't fail on import
fake_model_tools = types.SimpleNamespace(
check_tool_availability=lambda *a, **kw: ([], []),
TOOLSET_REQUIREMENTS={},
)
monkeypatch.setitem(sys.modules, "model_tools", fake_model_tools)
# Stub auth checks
try:
from hermes_cli import auth as _auth_mod
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
except Exception:
pass
# Stub httpx.get to avoid network calls
try:
import httpx
monkeypatch.setattr(httpx, "get", lambda *a, **kw: types.SimpleNamespace(status_code=200))
except Exception:
pass
return home, project, hermes_bin
def _run_doctor(fix=False):
"""Run doctor and capture stdout."""
import io
import contextlib
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
doctor_mod.run_doctor(Namespace(fix=fix))
return buf.getvalue()
class TestDoctorCommandInstallation:
"""Tests for the ◆ Command Installation section."""
@pytest.mark.skipif(sys.platform == "win32", reason="Symlink check is Unix-only")
def test_correct_symlink_shows_ok(self, monkeypatch, tmp_path):
home, project, hermes_bin = _setup_doctor_env(monkeypatch, tmp_path)
# Create the command link dir with correct symlink
cmd_link_dir = tmp_path / ".local" / "bin"
cmd_link_dir.mkdir(parents=True)
cmd_link = cmd_link_dir / "hermes"
cmd_link.symlink_to(hermes_bin)
monkeypatch.setattr(Path, "home", lambda: tmp_path)
out = _run_doctor(fix=False)
assert "Command Installation" in out
assert "Venv entry point exists" in out
assert "correct target" in out
@pytest.mark.skipif(sys.platform == "win32", reason="Symlink check is Unix-only")
def test_missing_symlink_shows_fail(self, monkeypatch, tmp_path):
home, project, hermes_bin = _setup_doctor_env(monkeypatch, tmp_path)
monkeypatch.setattr(Path, "home", lambda: tmp_path)
# Don't create the symlink — it should be missing
out = _run_doctor(fix=False)
assert "Command Installation" in out
assert "Venv entry point exists" in out
assert "not found" in out
assert "hermes doctor --fix" in out
@pytest.mark.skipif(sys.platform == "win32", reason="Symlink check is Unix-only")
def test_fix_creates_missing_symlink(self, monkeypatch, tmp_path):
home, project, hermes_bin = _setup_doctor_env(monkeypatch, tmp_path)
monkeypatch.setattr(Path, "home", lambda: tmp_path)
out = _run_doctor(fix=True)
assert "Command Installation" in out
assert "Created symlink" in out
# Verify the symlink was actually created
cmd_link = tmp_path / ".local" / "bin" / "hermes"
assert cmd_link.is_symlink()
assert cmd_link.resolve() == hermes_bin.resolve()
@pytest.mark.skipif(sys.platform == "win32", reason="Symlink check is Unix-only")
def test_wrong_target_symlink_shows_warn(self, monkeypatch, tmp_path):
home, project, hermes_bin = _setup_doctor_env(monkeypatch, tmp_path)
# Create a symlink pointing to the wrong target
cmd_link_dir = tmp_path / ".local" / "bin"
cmd_link_dir.mkdir(parents=True)
cmd_link = cmd_link_dir / "hermes"
wrong_target = tmp_path / "wrong_hermes"
wrong_target.write_text("#!/usr/bin/env python\n")
cmd_link.symlink_to(wrong_target)
monkeypatch.setattr(Path, "home", lambda: tmp_path)
out = _run_doctor(fix=False)
assert "Command Installation" in out
assert "wrong target" in out
@pytest.mark.skipif(sys.platform == "win32", reason="Symlink check is Unix-only")
def test_fix_repairs_wrong_symlink(self, monkeypatch, tmp_path):
home, project, hermes_bin = _setup_doctor_env(monkeypatch, tmp_path)
# Create a symlink pointing to wrong target
cmd_link_dir = tmp_path / ".local" / "bin"
cmd_link_dir.mkdir(parents=True)
cmd_link = cmd_link_dir / "hermes"
wrong_target = tmp_path / "wrong_hermes"
wrong_target.write_text("#!/usr/bin/env python\n")
cmd_link.symlink_to(wrong_target)
monkeypatch.setattr(Path, "home", lambda: tmp_path)
out = _run_doctor(fix=True)
assert "Fixed symlink" in out
# Verify the symlink now points to the correct target
assert cmd_link.is_symlink()
assert cmd_link.resolve() == hermes_bin.resolve()
@pytest.mark.skipif(sys.platform == "win32", reason="Symlink check is Unix-only")
def test_missing_venv_entry_point_shows_warn(self, monkeypatch, tmp_path):
home = tmp_path / ".hermes"
home.mkdir(parents=True, exist_ok=True)
(home / "config.yaml").write_text("memory: {}\n", encoding="utf-8")
project = tmp_path / "project"
project.mkdir(exist_ok=True)
# Do NOT create any venv entry point
monkeypatch.setattr(doctor_mod, "HERMES_HOME", home)
monkeypatch.setattr(doctor_mod, "PROJECT_ROOT", project)
monkeypatch.setattr(doctor_mod, "_DHH", str(home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
fake_model_tools = types.SimpleNamespace(
check_tool_availability=lambda *a, **kw: ([], []),
TOOLSET_REQUIREMENTS={},
)
monkeypatch.setitem(sys.modules, "model_tools", fake_model_tools)
try:
from hermes_cli import auth as _auth_mod
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
except Exception:
pass
try:
import httpx
monkeypatch.setattr(httpx, "get", lambda *a, **kw: types.SimpleNamespace(status_code=200))
except Exception:
pass
out = _run_doctor(fix=False)
assert "Command Installation" in out
assert "Venv entry point not found" in out
@pytest.mark.skipif(sys.platform == "win32", reason="Symlink check is Unix-only")
def test_dot_venv_dir_is_found(self, monkeypatch, tmp_path):
"""The check finds entry points in .venv/ as well as venv/."""
home, project, _ = _setup_doctor_env(monkeypatch, tmp_path, venv_name=".venv")
# Create the command link with correct symlink
hermes_bin = project / ".venv" / "bin" / "hermes"
cmd_link_dir = tmp_path / ".local" / "bin"
cmd_link_dir.mkdir(parents=True)
cmd_link = cmd_link_dir / "hermes"
cmd_link.symlink_to(hermes_bin)
monkeypatch.setattr(Path, "home", lambda: tmp_path)
out = _run_doctor(fix=False)
assert "Venv entry point exists" in out
assert ".venv/bin/hermes" in out
@pytest.mark.skipif(sys.platform == "win32", reason="Symlink check is Unix-only")
def test_non_symlink_regular_file_shows_ok(self, monkeypatch, tmp_path):
"""If ~/.local/bin/hermes is a regular file (not symlink), accept it."""
home, project, hermes_bin = _setup_doctor_env(monkeypatch, tmp_path)
cmd_link_dir = tmp_path / ".local" / "bin"
cmd_link_dir.mkdir(parents=True)
cmd_link = cmd_link_dir / "hermes"
cmd_link.write_text("#!/bin/sh\nexec python -m hermes_cli.main \"$@\"\n")
monkeypatch.setattr(Path, "home", lambda: tmp_path)
out = _run_doctor(fix=False)
assert "non-symlink" in out
@pytest.mark.skipif(sys.platform == "win32", reason="Symlink check is Unix-only")
def test_termux_uses_prefix_bin(self, monkeypatch, tmp_path):
"""On Termux, the command link dir is $PREFIX/bin."""
prefix_dir = tmp_path / "termux_prefix"
prefix_bin = prefix_dir / "bin"
prefix_bin.mkdir(parents=True)
home, project, hermes_bin = _setup_doctor_env(monkeypatch, tmp_path)
monkeypatch.setenv("TERMUX_VERSION", "0.118.3")
monkeypatch.setenv("PREFIX", str(prefix_dir))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
out = _run_doctor(fix=False)
assert "Command Installation" in out
assert "$PREFIX/bin" in out
def test_windows_skips_check(self, monkeypatch, tmp_path):
"""On Windows, the Command Installation section is skipped."""
home = tmp_path / ".hermes"
home.mkdir(parents=True, exist_ok=True)
(home / "config.yaml").write_text("memory: {}\n", encoding="utf-8")
project = tmp_path / "project"
project.mkdir(exist_ok=True)
monkeypatch.setattr(doctor_mod, "HERMES_HOME", home)
monkeypatch.setattr(doctor_mod, "PROJECT_ROOT", project)
monkeypatch.setattr(doctor_mod, "_DHH", str(home))
monkeypatch.setattr(sys, "platform", "win32")
fake_model_tools = types.SimpleNamespace(
check_tool_availability=lambda *a, **kw: ([], []),
TOOLSET_REQUIREMENTS={},
)
monkeypatch.setitem(sys.modules, "model_tools", fake_model_tools)
try:
from hermes_cli import auth as _auth_mod
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
except Exception:
pass
try:
import httpx
monkeypatch.setattr(httpx, "get", lambda *a, **kw: types.SimpleNamespace(status_code=200))
except Exception:
pass
out = _run_doctor(fix=False)
assert "Command Installation" not in out

148
tests/hermes_cli/test_subparser_routing_fallback.py Normal file
View File

@@ -0,0 +1,148 @@
"""Tests for the defensive subparser routing workaround (bpo-9338).
The main() function in hermes_cli/main.py sets subparsers.required=True
when argv contains a known subcommand name. This forces deterministic
routing on Python versions where argparse fails to match subcommand tokens
when the parent parser has nargs='?' optional arguments (--continue).
If the subcommand token is consumed as a flag value (e.g. `hermes -c model`
to resume a session named 'model'), the required=True parse raises
SystemExit and the code falls back to the default required=False behaviour.
"""
import argparse
import io
import sys
import pytest
def _build_parser():
"""Build a minimal replica of the hermes top-level parser."""
parser = argparse.ArgumentParser(prog="hermes")
parser.add_argument("--version", "-V", action="store_true")
parser.add_argument("--resume", "-r", metavar="SESSION", default=None)
parser.add_argument(
"--continue", "-c",
dest="continue_last",
nargs="?",
const=True,
default=None,
metavar="SESSION_NAME",
)
parser.add_argument("--worktree", "-w", action="store_true", default=False)
parser.add_argument("--skills", "-s", action="append", default=None)
parser.add_argument("--yolo", action="store_true", default=False)
parser.add_argument("--pass-session-id", action="store_true", default=False)
subparsers = parser.add_subparsers(dest="command", help="Command to run")
chat_p = subparsers.add_parser("chat")
chat_p.add_argument("-q", "--query", default=None)
subparsers.add_parser("model")
subparsers.add_parser("gateway")
subparsers.add_parser("setup")
return parser, subparsers
def _safe_parse(parser, subparsers, argv):
"""Replica of the defensive parsing logic from main()."""
known_cmds = set(subparsers.choices.keys()) if hasattr(subparsers, "choices") else set()
has_cmd_token = any(t in known_cmds for t in argv if not t.startswith("-"))
if has_cmd_token:
subparsers.required = True
saved_stderr = sys.stderr
try:
sys.stderr = io.StringIO()
args = parser.parse_args(argv)
sys.stderr = saved_stderr
return args
except SystemExit:
sys.stderr = saved_stderr
subparsers.required = False
return parser.parse_args(argv)
else:
subparsers.required = False
return parser.parse_args(argv)
class TestSubparserRoutingFallback:
"""Verify the bpo-9338 defensive routing works for all key cases."""
def test_direct_subcommand(self):
parser, sub = _build_parser()
args = _safe_parse(parser, sub, ["model"])
assert args.command == "model"
def test_subcommand_with_flags(self):
parser, sub = _build_parser()
args = _safe_parse(parser, sub, ["--yolo", "model"])
assert args.command == "model"
assert args.yolo is True
def test_bare_hermes_defaults_to_none(self):
parser, sub = _build_parser()
args = _safe_parse(parser, sub, [])
assert args.command is None
def test_flags_only_defaults_to_none(self):
parser, sub = _build_parser()
args = _safe_parse(parser, sub, ["--yolo"])
assert args.command is None
assert args.yolo is True
def test_continue_flag_alone(self):
parser, sub = _build_parser()
args = _safe_parse(parser, sub, ["-c"])
assert args.command is None
assert args.continue_last is True
def test_continue_with_session_name(self):
parser, sub = _build_parser()
args = _safe_parse(parser, sub, ["-c", "myproject"])
assert args.command is None
assert args.continue_last == "myproject"
def test_continue_with_subcommand_name_as_session(self):
"""Edge case: session named 'model' — should be treated as session name, not subcommand."""
parser, sub = _build_parser()
args = _safe_parse(parser, sub, ["-c", "model"])
assert args.command is None
assert args.continue_last == "model"
def test_continue_with_session_then_subcommand(self):
parser, sub = _build_parser()
args = _safe_parse(parser, sub, ["-c", "myproject", "model"])
assert args.command == "model"
assert args.continue_last == "myproject"
def test_chat_with_query(self):
parser, sub = _build_parser()
args = _safe_parse(parser, sub, ["chat", "-q", "hello"])
assert args.command == "chat"
assert args.query == "hello"
def test_resume_flag(self):
parser, sub = _build_parser()
args = _safe_parse(parser, sub, ["-r", "abc123"])
assert args.command is None
assert args.resume == "abc123"
def test_resume_with_subcommand(self):
parser, sub = _build_parser()
args = _safe_parse(parser, sub, ["-r", "abc123", "chat"])
assert args.command == "chat"
assert args.resume == "abc123"
def test_skills_flag_with_subcommand(self):
parser, sub = _build_parser()
args = _safe_parse(parser, sub, ["-s", "myskill", "chat"])
assert args.command == "chat"
assert args.skills == ["myskill"]
def test_all_flags_with_subcommand(self):
parser, sub = _build_parser()
args = _safe_parse(parser, sub, ["--yolo", "-w", "-s", "myskill", "model"])
assert args.command == "model"
assert args.yolo is True
assert args.worktree is True
assert args.skills == ["myskill"]

10
tests/hermes_cli/test_tools_config.py
View File

@@ -8,6 +8,7 @@ from hermes_cli.tools_config import (
_platform_toolset_summary,
_save_platform_tools,
_toolset_has_keys,
CONFIGURABLE_TOOLSETS,
TOOL_CATEGORIES,
_visible_providers,
tools_command,
@@ -22,6 +23,15 @@ def test_get_platform_tools_uses_default_when_platform_not_configured():
assert enabled
def test_configurable_toolsets_include_messaging():
assert any(ts_key == "messaging" for ts_key, _, _ in CONFIGURABLE_TOOLSETS)
def test_get_platform_tools_default_telegram_includes_messaging():
enabled = _get_platform_tools({}, "telegram")
assert "messaging" in enabled
def test_get_platform_tools_preserves_explicit_empty_selection():
config = {"platform_toolsets": {"cli": []}}

111
tests/run_agent/test_invalid_context_length_warning.py Normal file
View File

@@ -0,0 +1,111 @@
"""Tests that invalid context_length values in config produce visible warnings."""
from unittest.mock import patch, MagicMock, call
def _build_agent(model_cfg, custom_providers=None, model="anthropic/claude-opus-4.6"):
"""Build an AIAgent with the given model config."""
cfg = {"model": model_cfg}
if custom_providers is not None:
cfg["custom_providers"] = custom_providers
with (
patch("hermes_cli.config.load_config", return_value=cfg),
patch("agent.model_metadata.get_model_context_length", return_value=128_000),
patch("run_agent.get_tool_definitions", return_value=[]),
patch("run_agent.check_toolset_requirements", return_value={}),
patch("run_agent.OpenAI"),
):
from run_agent import AIAgent
agent = AIAgent(
model=model,
api_key="test-key-1234567890",
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
)
return agent
def test_valid_integer_context_length_no_warning():
"""Plain integer context_length should work silently."""
with patch("run_agent.logger") as mock_logger:
agent = _build_agent({"default": "gpt5.4", "provider": "custom",
"base_url": "http://localhost:4000/v1",
"context_length": 256000})
assert agent._config_context_length == 256000
# No warning about invalid context_length
for c in mock_logger.warning.call_args_list:
assert "Invalid" not in str(c)
def test_string_k_suffix_context_length_warns():
"""context_length: '256K' should warn the user clearly."""
with patch("run_agent.logger") as mock_logger:
agent = _build_agent({"default": "gpt5.4", "provider": "custom",
"base_url": "http://localhost:4000/v1",
"context_length": "256K"})
assert agent._config_context_length is None
# Should have warned
warning_calls = [c for c in mock_logger.warning.call_args_list
if "Invalid" in str(c) and "256K" in str(c)]
assert len(warning_calls) == 1
assert "plain integer" in str(warning_calls[0])
def test_string_numeric_context_length_works():
"""context_length: '256000' (string) should parse fine via int()."""
with patch("run_agent.logger") as mock_logger:
agent = _build_agent({"default": "gpt5.4", "provider": "custom",
"base_url": "http://localhost:4000/v1",
"context_length": "256000"})
assert agent._config_context_length == 256000
for c in mock_logger.warning.call_args_list:
assert "Invalid" not in str(c)
def test_custom_providers_invalid_context_length_warns():
"""Invalid context_length in custom_providers should warn."""
custom_providers = [
{
"name": "LiteLLM",
"base_url": "http://localhost:4000/v1",
"models": {
"gpt5.4": {"context_length": "256K"}
},
}
]
with patch("run_agent.logger") as mock_logger:
agent = _build_agent(
{"default": "gpt5.4", "provider": "custom",
"base_url": "http://localhost:4000/v1"},
custom_providers=custom_providers,
model="gpt5.4",
)
warning_calls = [c for c in mock_logger.warning.call_args_list
if "Invalid" in str(c) and "256K" in str(c)]
assert len(warning_calls) == 1
assert "custom_providers" in str(warning_calls[0])
def test_custom_providers_valid_context_length():
"""Valid integer in custom_providers should work silently."""
custom_providers = [
{
"name": "LiteLLM",
"base_url": "http://localhost:4000/v1",
"models": {
"gpt5.4": {"context_length": 256000}
},
}
]
with patch("run_agent.logger") as mock_logger:
agent = _build_agent(
{"default": "gpt5.4", "provider": "custom",
"base_url": "http://localhost:4000/v1"},
custom_providers=custom_providers,
model="gpt5.4",
)
for c in mock_logger.warning.call_args_list:
assert "Invalid" not in str(c)

28
tests/run_agent/test_run_agent_codex_responses.py
View File

@@ -1249,13 +1249,17 @@ def test_chat_messages_to_responses_input_deduplicates_reasoning_ids(monkeypatch
]
items = agent._chat_messages_to_responses_input(messages)
reasoning_ids = [it["id"] for it in items if it.get("type") == "reasoning"]
# rs_aaa should appear only once (first occurrence kept)
assert reasoning_ids.count("rs_aaa") == 1
# rs_bbb and rs_ccc should each appear once
assert reasoning_ids.count("rs_bbb") == 1
assert reasoning_ids.count("rs_ccc") == 1
assert len(reasoning_ids) == 3
reasoning_items = [it for it in items if it.get("type") == "reasoning"]
# Dedup: rs_aaa appears in both turns but should only be emitted once.
# 3 unique items total: enc_1 (from rs_aaa), enc_2 (rs_bbb), enc_3 (rs_ccc).
assert len(reasoning_items) == 3
encrypted = [it["encrypted_content"] for it in reasoning_items]
assert encrypted.count("enc_1") == 1
assert "enc_2" in encrypted
assert "enc_3" in encrypted
# IDs must be stripped — with store=False the API 404s on id lookups.
for it in reasoning_items:
assert "id" not in it
def test_preflight_codex_input_deduplicates_reasoning_ids(monkeypatch):
@@ -1272,7 +1276,11 @@ def test_preflight_codex_input_deduplicates_reasoning_ids(monkeypatch):
normalized = agent._preflight_codex_input_items(raw_input)
reasoning_items = [it for it in normalized if it.get("type") == "reasoning"]
reasoning_ids = [it["id"] for it in reasoning_items]
assert reasoning_ids.count("rs_xyz") == 1
assert reasoning_ids.count("rs_zzz") == 1
# rs_xyz duplicate should be collapsed to one item; rs_zzz kept.
assert len(reasoning_items) == 2
encrypted = [it["encrypted_content"] for it in reasoning_items]
assert encrypted.count("enc_a") == 1
assert "enc_b" in encrypted
# IDs must be stripped — with store=False the API 404s on id lookups.
for it in reasoning_items:
assert "id" not in it

64
tests/run_agent/test_unicode_ascii_codec.py
View File

@@ -230,3 +230,67 @@ class TestSanitizeStructureNonAscii:
assert _sanitize_structure_non_ascii(payload) is True
assert payload["default_headers"]["X-Title"] == "Hermes Agent"
assert payload["default_headers"]["User-Agent"] == "Hermes/1.0 "
class TestApiKeyClientSync:
"""Verify that ASCII recovery updates the live OpenAI client's api_key.
The OpenAI SDK stores its own copy of api_key which auth_headers reads
dynamically. If only self.api_key is updated but self.client.api_key
is not, the next request still sends the corrupted key in the
Authorization header.
"""
def test_client_api_key_updated_on_sanitize(self):
"""Simulate the recovery path and verify client.api_key is synced."""
from unittest.mock import MagicMock
from run_agent import AIAgent
agent = AIAgent.__new__(AIAgent)
bad_key = "sk-proj-abc\u028bdef" # ʋ lookalike at position 11
agent.api_key = bad_key
agent._client_kwargs = {"api_key": bad_key}
agent.quiet_mode = True
# Mock client with its own api_key attribute (like the real OpenAI client)
mock_client = MagicMock()
mock_client.api_key = bad_key
agent.client = mock_client
# --- replicate the recovery logic from run_agent.py ---
_raw_key = agent.api_key
_clean_key = _strip_non_ascii(_raw_key)
assert _clean_key != _raw_key, "test precondition: key should have non-ASCII"
agent.api_key = _clean_key
agent._client_kwargs["api_key"] = _clean_key
if getattr(agent, "client", None) is not None and hasattr(agent.client, "api_key"):
agent.client.api_key = _clean_key
# All three locations should now hold the clean key
assert agent.api_key == "sk-proj-abcdef"
assert agent._client_kwargs["api_key"] == "sk-proj-abcdef"
assert agent.client.api_key == "sk-proj-abcdef"
# The bad char should be gone from all of them
assert "\u028b" not in agent.api_key
assert "\u028b" not in agent._client_kwargs["api_key"]
assert "\u028b" not in agent.client.api_key
def test_client_none_does_not_crash(self):
"""Recovery should not crash when client is None (pre-init)."""
from run_agent import AIAgent
agent = AIAgent.__new__(AIAgent)
bad_key = "sk-proj-\u028b"
agent.api_key = bad_key
agent._client_kwargs = {"api_key": bad_key}
agent.client = None
_clean_key = _strip_non_ascii(bad_key)
agent.api_key = _clean_key
agent._client_kwargs["api_key"] = _clean_key
if getattr(agent, "client", None) is not None and hasattr(agent.client, "api_key"):
agent.client.api_key = _clean_key
assert agent.api_key == "sk-proj-"
assert agent.client is None # should not have been touched

254
tests/tools/test_send_message_tool.py
View File

@@ -752,6 +752,38 @@ class TestParseTargetRefDiscord:
assert is_explicit is True
class TestParseTargetRefMatrix:
"""_parse_target_ref correctly handles Matrix room IDs and user MXIDs."""
def test_matrix_room_id_is_explicit(self):
"""Matrix room IDs (!) are recognized as explicit targets."""
chat_id, thread_id, is_explicit = _parse_target_ref("matrix", "!HLOQwxYGgFPMPJUSNR:matrix.org")
assert chat_id == "!HLOQwxYGgFPMPJUSNR:matrix.org"
assert thread_id is None
assert is_explicit is True
def test_matrix_user_mxid_is_explicit(self):
"""Matrix user MXIDs (@) are recognized as explicit targets."""
chat_id, thread_id, is_explicit = _parse_target_ref("matrix", "@hermes:matrix.org")
assert chat_id == "@hermes:matrix.org"
assert thread_id is None
assert is_explicit is True
def test_matrix_alias_is_not_explicit(self):
"""Matrix room aliases (#) are NOT explicit — they need resolution."""
chat_id, thread_id, is_explicit = _parse_target_ref("matrix", "#general:matrix.org")
assert chat_id is None
assert is_explicit is False
def test_matrix_prefix_only_matches_matrix_platform(self):
"""! and @ prefixes are only treated as explicit for the matrix platform."""
chat_id, _, is_explicit = _parse_target_ref("telegram", "!something")
assert is_explicit is False
chat_id, _, is_explicit = _parse_target_ref("discord", "@someone")
assert is_explicit is False
class TestSendDiscordThreadId:
"""_send_discord uses thread_id when provided."""
@@ -854,3 +886,225 @@ class TestSendToPlatformDiscordThread:
send_mock.assert_awaited_once()
_, call_kwargs = send_mock.await_args
assert call_kwargs["thread_id"] is None
# ---------------------------------------------------------------------------
# Discord media attachment support
# ---------------------------------------------------------------------------
class TestSendDiscordMedia:
"""_send_discord uploads media files via multipart/form-data."""
@staticmethod
def _build_mock(response_status, response_data=None, response_text="error body"):
"""Build a properly-structured aiohttp mock chain."""
mock_resp = MagicMock()
mock_resp.status = response_status
mock_resp.json = AsyncMock(return_value=response_data or {"id": "msg123"})
mock_resp.text = AsyncMock(return_value=response_text)
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=None)
mock_session = MagicMock()
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock(return_value=None)
mock_session.post = MagicMock(return_value=mock_resp)
return mock_session, mock_resp
def test_text_and_media_sends_both(self, tmp_path):
"""Text message is sent first, then each media file as multipart."""
img = tmp_path / "photo.png"
img.write_bytes(b"\x89PNG fake image data")
mock_session, _ = self._build_mock(200, {"id": "msg999"})
with patch("aiohttp.ClientSession", return_value=mock_session):
result = asyncio.run(
_send_discord("tok", "111", "hello", media_files=[(str(img), False)])
)
assert result["success"] is True
assert result["message_id"] == "msg999"
# Two POSTs: one text JSON, one multipart upload
assert mock_session.post.call_count == 2
def test_media_only_skips_text_post(self, tmp_path):
"""When message is empty and media is present, text POST is skipped."""
img = tmp_path / "photo.png"
img.write_bytes(b"\x89PNG fake image data")
mock_session, _ = self._build_mock(200, {"id": "media_only"})
with patch("aiohttp.ClientSession", return_value=mock_session):
result = asyncio.run(
_send_discord("tok", "222", " ", media_files=[(str(img), False)])
)
assert result["success"] is True
# Only one POST: the media upload (text was whitespace-only)
assert mock_session.post.call_count == 1
def test_missing_media_file_collected_as_warning(self):
"""Non-existent media paths produce warnings but don't fail."""
mock_session, _ = self._build_mock(200, {"id": "txt_ok"})
with patch("aiohttp.ClientSession", return_value=mock_session):
result = asyncio.run(
_send_discord("tok", "333", "hello", media_files=[("/nonexistent/file.png", False)])
)
assert result["success"] is True
assert "warnings" in result
assert any("not found" in w for w in result["warnings"])
# Only the text POST was made, media was skipped
assert mock_session.post.call_count == 1
def test_media_upload_failure_collected_as_warning(self, tmp_path):
"""Failed media upload becomes a warning, text still succeeds."""
img = tmp_path / "photo.png"
img.write_bytes(b"\x89PNG fake image data")
# First call (text) succeeds, second call (media) returns 413
text_resp = MagicMock()
text_resp.status = 200
text_resp.json = AsyncMock(return_value={"id": "txt_ok"})
text_resp.__aenter__ = AsyncMock(return_value=text_resp)
text_resp.__aexit__ = AsyncMock(return_value=None)
media_resp = MagicMock()
media_resp.status = 413
media_resp.text = AsyncMock(return_value="Request Entity Too Large")
media_resp.__aenter__ = AsyncMock(return_value=media_resp)
media_resp.__aexit__ = AsyncMock(return_value=None)
mock_session = MagicMock()
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock(return_value=None)
mock_session.post = MagicMock(side_effect=[text_resp, media_resp])
with patch("aiohttp.ClientSession", return_value=mock_session):
result = asyncio.run(
_send_discord("tok", "444", "hello", media_files=[(str(img), False)])
)
assert result["success"] is True
assert result["message_id"] == "txt_ok"
assert "warnings" in result
assert any("413" in w for w in result["warnings"])
def test_no_text_no_media_returns_error(self):
"""Empty text with no media returns error dict."""
mock_session, _ = self._build_mock(200)
with patch("aiohttp.ClientSession", return_value=mock_session):
result = asyncio.run(
_send_discord("tok", "555", "", media_files=[])
)
# Text is empty but media_files is empty, so text POST fires
# (the "skip text if media present" condition isn't met)
assert result["success"] is True
def test_multiple_media_files_uploaded_separately(self, tmp_path):
"""Each media file gets its own multipart POST."""
img1 = tmp_path / "a.png"
img1.write_bytes(b"img1")
img2 = tmp_path / "b.jpg"
img2.write_bytes(b"img2")
mock_session, _ = self._build_mock(200, {"id": "last"})
with patch("aiohttp.ClientSession", return_value=mock_session):
result = asyncio.run(
_send_discord("tok", "666", "hi", media_files=[
(str(img1), False), (str(img2), False)
])
)
assert result["success"] is True
# 1 text POST + 2 media POSTs = 3
assert mock_session.post.call_count == 3
class TestSendToPlatformDiscordMedia:
"""_send_to_platform routes Discord media correctly."""
def test_media_files_passed_on_last_chunk_only(self):
"""Discord media_files are only passed on the final chunk."""
call_log = []
async def mock_send_discord(token, chat_id, message, thread_id=None, media_files=None):
call_log.append({"message": message, "media_files": media_files or []})
return {"success": True, "platform": "discord", "chat_id": chat_id, "message_id": "1"}
# A message long enough to get chunked (Discord limit is 2000)
long_msg = "A" * 1900 + " " + "B" * 1900
with patch("tools.send_message_tool._send_discord", side_effect=mock_send_discord):
result = asyncio.run(
_send_to_platform(
Platform.DISCORD,
SimpleNamespace(enabled=True, token="tok", extra={}),
"999",
long_msg,
media_files=[("/fake/img.png", False)],
)
)
assert result["success"] is True
assert len(call_log) == 2 # Message was chunked
assert call_log[0]["media_files"] == [] # First chunk: no media
assert call_log[1]["media_files"] == [("/fake/img.png", False)] # Last chunk: media attached
def test_single_chunk_gets_media(self):
"""Short message (single chunk) gets media_files directly."""
send_mock = AsyncMock(return_value={"success": True, "message_id": "1"})
with patch("tools.send_message_tool._send_discord", send_mock):
result = asyncio.run(
_send_to_platform(
Platform.DISCORD,
SimpleNamespace(enabled=True, token="tok", extra={}),
"888",
"short message",
media_files=[("/fake/img.png", False)],
)
)
assert result["success"] is True
send_mock.assert_awaited_once()
call_kwargs = send_mock.await_args.kwargs
assert call_kwargs["media_files"] == [("/fake/img.png", False)]
class TestSendMatrixUrlEncoding:
"""_send_matrix URL-encodes Matrix room IDs in the API path."""
def test_room_id_is_percent_encoded_in_url(self):
"""Matrix room IDs with ! and : are percent-encoded in the PUT URL."""
import aiohttp
mock_resp = MagicMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(return_value={"event_id": "$evt123"})
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=None)
mock_session = MagicMock()
mock_session.put = MagicMock(return_value=mock_resp)
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock(return_value=None)
with patch("aiohttp.ClientSession", return_value=mock_session):
from tools.send_message_tool import _send_matrix
result = asyncio.get_event_loop().run_until_complete(
_send_matrix(
"test_token",
{"homeserver": "https://matrix.example.org"},
"!HLOQwxYGgFPMPJUSNR:matrix.org",
"hello",
)
)
assert result["success"] is True
# Verify the URL was called with percent-encoded room ID
put_url = mock_session.put.call_args[0][0]
assert "%21HLOQwxYGgFPMPJUSNR%3Amatrix.org" in put_url
assert "!HLOQwxYGgFPMPJUSNR:matrix.org" not in put_url