mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-23 16:36:23 +00:00
The test installs a secret scope and asserts a scope miss does NOT fall back to the default profile's env — that isolation guarantee only holds under multiplexing, which the real gateway activates at startup via set_multiplex_active(). With the #67827 overlay fallthrough (scope miss → os.environ when multiplex is OFF), the test needs to model the multiplexed runtime it is actually testing.
2119 lines
80 KiB
Python
2119 lines
80 KiB
Python
"""Tests for gateway configuration management."""
|
|
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from agent.secret_scope import (
|
|
reset_secret_scope,
|
|
set_multiplex_active,
|
|
set_secret_scope,
|
|
)
|
|
from hermes_constants import reset_hermes_home_override, set_hermes_home_override
|
|
from gateway.config import (
|
|
ChannelOverride,
|
|
GatewayConfig,
|
|
HomeChannel,
|
|
Platform,
|
|
PlatformConfig,
|
|
SessionResetPolicy,
|
|
StreamingConfig,
|
|
_apply_env_overrides,
|
|
load_gateway_config,
|
|
)
|
|
|
|
|
|
class TestHomeChannelRoundtrip:
|
|
def test_to_dict_from_dict(self):
|
|
hc = HomeChannel(platform=Platform.DISCORD, chat_id="999", name="general")
|
|
d = hc.to_dict()
|
|
restored = HomeChannel.from_dict(d)
|
|
|
|
assert restored.platform == Platform.DISCORD
|
|
assert restored.chat_id == "999"
|
|
assert restored.name == "general"
|
|
|
|
|
|
class TestPlatformConfigRoundtrip:
|
|
def test_to_dict_from_dict(self):
|
|
pc = PlatformConfig(
|
|
enabled=True,
|
|
token="tok_123",
|
|
home_channel=HomeChannel(
|
|
platform=Platform.TELEGRAM,
|
|
chat_id="555",
|
|
name="Home",
|
|
),
|
|
extra={"foo": "bar"},
|
|
)
|
|
d = pc.to_dict()
|
|
restored = PlatformConfig.from_dict(d)
|
|
|
|
assert restored.enabled is True
|
|
assert restored.token == "tok_123"
|
|
assert restored.home_channel.chat_id == "555"
|
|
assert restored.extra == {"foo": "bar"}
|
|
|
|
def test_disabled_no_token(self):
|
|
pc = PlatformConfig()
|
|
d = pc.to_dict()
|
|
restored = PlatformConfig.from_dict(d)
|
|
assert restored.enabled is False
|
|
assert restored.token is None
|
|
|
|
def test_from_dict_coerces_quoted_false_enabled(self):
|
|
restored = PlatformConfig.from_dict({"enabled": "false"})
|
|
assert restored.enabled is False
|
|
|
|
def test_gateway_restart_notification_defaults_true(self):
|
|
assert PlatformConfig().gateway_restart_notification is True
|
|
assert PlatformConfig.from_dict({}).gateway_restart_notification is True
|
|
|
|
def test_gateway_restart_notification_roundtrip_false(self):
|
|
pc = PlatformConfig(enabled=True, gateway_restart_notification=False)
|
|
restored = PlatformConfig.from_dict(pc.to_dict())
|
|
assert restored.gateway_restart_notification is False
|
|
|
|
def test_gateway_restart_notification_coerces_quoted_false(self):
|
|
restored = PlatformConfig.from_dict({"gateway_restart_notification": "false"})
|
|
assert restored.gateway_restart_notification is False
|
|
|
|
def test_typing_indicator_defaults_true(self):
|
|
assert PlatformConfig().typing_indicator is True
|
|
assert PlatformConfig.from_dict({}).typing_indicator is True
|
|
|
|
def test_typing_indicator_roundtrip_false(self):
|
|
pc = PlatformConfig(enabled=True, typing_indicator=False)
|
|
restored = PlatformConfig.from_dict(pc.to_dict())
|
|
assert restored.typing_indicator is False
|
|
|
|
def test_typing_indicator_coerces_quoted_false(self):
|
|
restored = PlatformConfig.from_dict({"typing_indicator": "false"})
|
|
assert restored.typing_indicator is False
|
|
|
|
def test_typing_indicator_resolved_from_extra(self):
|
|
# The shared-key loop in load_gateway_config bridges the flag into
|
|
# extra; from_dict must honor it there too (mirrors _grn fallback).
|
|
restored = PlatformConfig.from_dict({"extra": {"typing_indicator": False}})
|
|
assert restored.typing_indicator is False
|
|
|
|
def test_typing_status_text_defaults_none(self):
|
|
assert PlatformConfig().typing_status_text is None
|
|
assert PlatformConfig.from_dict({}).typing_status_text is None
|
|
|
|
def test_typing_status_text_roundtrip(self):
|
|
pc = PlatformConfig(enabled=True, typing_status_text="is pouncing… 🐾")
|
|
restored = PlatformConfig.from_dict(pc.to_dict())
|
|
assert restored.typing_status_text == "is pouncing… 🐾"
|
|
|
|
def test_typing_status_text_resolved_from_extra(self):
|
|
# Same bridge route as typing_indicator: the shared-key loop copies a
|
|
# nested platforms.<plat> value into extra.
|
|
restored = PlatformConfig.from_dict(
|
|
{"extra": {"typing_status_text": "chasing yarn…"}}
|
|
)
|
|
assert restored.typing_status_text == "chasing yarn…"
|
|
|
|
def test_typing_status_text_omitted_from_to_dict_when_unset(self):
|
|
# None must not serialize — keeps existing config files byte-stable.
|
|
assert "typing_status_text" not in PlatformConfig().to_dict()
|
|
|
|
def test_channel_overrides_roundtrip(self):
|
|
pc = PlatformConfig(
|
|
enabled=True,
|
|
channel_overrides={
|
|
"1234567890": ChannelOverride(
|
|
model="openrouter/healer-alpha",
|
|
provider="openrouter",
|
|
system_prompt="You are a daily news summarizer.",
|
|
),
|
|
"9876543210": ChannelOverride(
|
|
model="anthropic/claude-opus-4.6",
|
|
provider="anthropic",
|
|
system_prompt="You are a coding assistant.",
|
|
),
|
|
},
|
|
)
|
|
d = pc.to_dict()
|
|
assert "channel_overrides" in d
|
|
assert d["channel_overrides"]["1234567890"]["model"] == "openrouter/healer-alpha"
|
|
assert d["channel_overrides"]["9876543210"]["system_prompt"] == "You are a coding assistant."
|
|
restored = PlatformConfig.from_dict(d)
|
|
assert restored.channel_overrides["1234567890"].model == "openrouter/healer-alpha"
|
|
assert restored.channel_overrides["9876543210"].provider == "anthropic"
|
|
|
|
def test_channel_overrides_from_dict_normalizes_channel_id_to_str(self):
|
|
"""YAML may have numeric channel IDs; we store as str."""
|
|
data = {
|
|
"enabled": True,
|
|
"channel_overrides": {
|
|
1234567890: {"model": "openrouter/healer-alpha"},
|
|
},
|
|
}
|
|
pc = PlatformConfig.from_dict(data)
|
|
assert "1234567890" in pc.channel_overrides
|
|
assert pc.channel_overrides["1234567890"].model == "openrouter/healer-alpha"
|
|
|
|
|
|
class TestChannelOverride:
|
|
def test_from_dict_empty(self):
|
|
assert ChannelOverride.from_dict({}).model is None
|
|
assert ChannelOverride.from_dict(None).model is None
|
|
|
|
def test_to_dict_omits_none(self):
|
|
ov = ChannelOverride(model="gpt-4", provider=None, system_prompt="Hi")
|
|
d = ov.to_dict()
|
|
assert d["model"] == "gpt-4"
|
|
assert "provider" not in d
|
|
assert d["system_prompt"] == "Hi"
|
|
|
|
|
|
class TestPlatformConfigMalformedSections:
|
|
def test_from_dict_ignores_malformed_nested_sections(self):
|
|
restored = PlatformConfig.from_dict(
|
|
{
|
|
"enabled": True,
|
|
"home_channel": "telegram:123",
|
|
"extra": "oops",
|
|
}
|
|
)
|
|
|
|
assert restored.enabled is True
|
|
assert restored.home_channel is None
|
|
assert restored.extra == {}
|
|
|
|
|
|
class TestGetConnectedPlatforms:
|
|
def test_returns_enabled_with_token(self):
|
|
config = GatewayConfig(
|
|
platforms={
|
|
Platform.TELEGRAM: PlatformConfig(enabled=True, token="t"),
|
|
Platform.DISCORD: PlatformConfig(enabled=False, token="d"),
|
|
Platform.SLACK: PlatformConfig(enabled=True), # no token
|
|
},
|
|
)
|
|
connected = config.get_connected_platforms()
|
|
assert Platform.TELEGRAM in connected
|
|
assert Platform.DISCORD not in connected
|
|
assert Platform.SLACK not in connected
|
|
|
|
def test_empty_platforms(self):
|
|
config = GatewayConfig()
|
|
assert config.get_connected_platforms() == []
|
|
|
|
def test_dingtalk_recognised_via_extras(self):
|
|
config = GatewayConfig(
|
|
platforms={
|
|
Platform.DINGTALK: PlatformConfig(
|
|
enabled=True,
|
|
extra={"client_id": "cid", "client_secret": "sec"},
|
|
),
|
|
},
|
|
)
|
|
assert Platform.DINGTALK in config.get_connected_platforms()
|
|
|
|
def test_dingtalk_recognised_via_env_vars(self, monkeypatch):
|
|
"""DingTalk configured via env vars (no extras) should still be
|
|
recognised as connected — covers the case where _apply_env_overrides
|
|
hasn't populated extras yet."""
|
|
monkeypatch.setenv("DINGTALK_CLIENT_ID", "env_cid")
|
|
monkeypatch.setenv("DINGTALK_CLIENT_SECRET", "env_sec")
|
|
config = GatewayConfig(
|
|
platforms={
|
|
Platform.DINGTALK: PlatformConfig(enabled=True, extra={}),
|
|
},
|
|
)
|
|
assert Platform.DINGTALK in config.get_connected_platforms()
|
|
|
|
def test_dingtalk_missing_creds_not_connected(self, monkeypatch):
|
|
monkeypatch.delenv("DINGTALK_CLIENT_ID", raising=False)
|
|
monkeypatch.delenv("DINGTALK_CLIENT_SECRET", raising=False)
|
|
config = GatewayConfig(
|
|
platforms={
|
|
Platform.DINGTALK: PlatformConfig(enabled=True, extra={}),
|
|
},
|
|
)
|
|
assert Platform.DINGTALK not in config.get_connected_platforms()
|
|
|
|
def test_dingtalk_disabled_not_connected(self):
|
|
config = GatewayConfig(
|
|
platforms={
|
|
Platform.DINGTALK: PlatformConfig(
|
|
enabled=False,
|
|
extra={"client_id": "cid", "client_secret": "sec"},
|
|
),
|
|
},
|
|
)
|
|
assert Platform.DINGTALK not in config.get_connected_platforms()
|
|
|
|
|
|
class TestSessionResetPolicy:
|
|
def test_roundtrip(self):
|
|
policy = SessionResetPolicy(mode="idle", at_hour=6, idle_minutes=120,
|
|
bg_process_max_age_hours=48)
|
|
d = policy.to_dict()
|
|
restored = SessionResetPolicy.from_dict(d)
|
|
assert restored.mode == "idle"
|
|
assert restored.at_hour == 6
|
|
assert restored.idle_minutes == 120
|
|
assert restored.bg_process_max_age_hours == 48
|
|
|
|
def test_defaults(self):
|
|
policy = SessionResetPolicy()
|
|
assert policy.mode == "none"
|
|
assert policy.at_hour == 4
|
|
assert policy.idle_minutes == 1440
|
|
assert policy.bg_process_max_age_hours == 24
|
|
|
|
def test_from_dict_treats_null_values_as_defaults(self):
|
|
restored = SessionResetPolicy.from_dict(
|
|
{"mode": None, "at_hour": None, "idle_minutes": None,
|
|
"bg_process_max_age_hours": None}
|
|
)
|
|
assert restored.mode == "none"
|
|
assert restored.at_hour == 4
|
|
assert restored.idle_minutes == 1440
|
|
assert restored.bg_process_max_age_hours == 24
|
|
|
|
def test_from_dict_coerces_quoted_false_notify(self):
|
|
restored = SessionResetPolicy.from_dict({"notify": "false"})
|
|
assert restored.notify is False
|
|
|
|
def test_from_dict_malformed_section_falls_back_to_defaults(self):
|
|
restored = SessionResetPolicy.from_dict("oops")
|
|
assert restored.mode == SessionResetPolicy().mode
|
|
assert restored.at_hour == 4
|
|
assert restored.idle_minutes == 1440
|
|
|
|
|
|
class TestStreamingConfig:
|
|
def test_defaults_to_auto_transport(self):
|
|
# "auto" prefers native draft streaming where the platform supports
|
|
# it (Telegram DMs) and falls back to edit-based everywhere else, so
|
|
# it is safe as the global out-of-the-box default.
|
|
restored = StreamingConfig.from_dict({"enabled": "true"})
|
|
assert restored.transport == "auto"
|
|
|
|
def test_from_dict_coerces_quoted_false_enabled(self):
|
|
restored = StreamingConfig.from_dict({"enabled": "false"})
|
|
assert restored.enabled is False
|
|
|
|
def test_from_dict_malformed_numeric_values_fall_back_to_defaults(self):
|
|
restored = StreamingConfig.from_dict(
|
|
{
|
|
"edit_interval": "oops",
|
|
"buffer_threshold": "oops",
|
|
"fresh_final_after_seconds": "oops",
|
|
}
|
|
)
|
|
assert restored.edit_interval == 0.8
|
|
assert restored.buffer_threshold == 24
|
|
assert restored.fresh_final_after_seconds == 0.0
|
|
|
|
def test_from_dict_malformed_section_falls_back_to_defaults(self):
|
|
restored = StreamingConfig.from_dict("enabled")
|
|
assert restored.enabled is False
|
|
assert restored.transport == "auto"
|
|
|
|
|
|
class TestGatewayConfigRoundtrip:
|
|
def test_full_roundtrip(self):
|
|
config = GatewayConfig(
|
|
platforms={
|
|
Platform.TELEGRAM: PlatformConfig(
|
|
enabled=True,
|
|
token="tok_123",
|
|
home_channel=HomeChannel(Platform.TELEGRAM, "123", "Home"),
|
|
),
|
|
},
|
|
reset_triggers=["/new"],
|
|
quick_commands={"limits": {"type": "exec", "command": "echo ok"}},
|
|
group_sessions_per_user=False,
|
|
thread_sessions_per_user=True,
|
|
systemd_watchdog_seconds=120,
|
|
)
|
|
d = config.to_dict()
|
|
restored = GatewayConfig.from_dict(d)
|
|
|
|
assert Platform.TELEGRAM in restored.platforms
|
|
assert restored.platforms[Platform.TELEGRAM].token == "tok_123"
|
|
assert restored.reset_triggers == ["/new"]
|
|
assert restored.quick_commands == {"limits": {"type": "exec", "command": "echo ok"}}
|
|
assert restored.group_sessions_per_user is False
|
|
assert restored.thread_sessions_per_user is True
|
|
assert restored.systemd_watchdog_seconds == 120
|
|
|
|
def test_systemd_watchdog_from_dict_disables_invalid_values(self):
|
|
invalid_values = [
|
|
None,
|
|
0,
|
|
-1,
|
|
True,
|
|
1.5,
|
|
float("nan"),
|
|
float("inf"),
|
|
"120.0",
|
|
"1e3",
|
|
"bad",
|
|
2_147_483_648,
|
|
]
|
|
|
|
for raw in invalid_values:
|
|
config = GatewayConfig.from_dict({"systemd_watchdog_seconds": raw})
|
|
assert config.systemd_watchdog_seconds == 0
|
|
|
|
def test_systemd_watchdog_from_dict_accepts_nested_positive_integer(self):
|
|
config = GatewayConfig.from_dict(
|
|
{"gateway": {"systemd_watchdog_seconds": "45"}}
|
|
)
|
|
|
|
assert config.systemd_watchdog_seconds == 45
|
|
|
|
def test_max_concurrent_sessions_from_dict_normalizes_disabled_values(self):
|
|
assert GatewayConfig.from_dict({}).max_concurrent_sessions is None
|
|
assert GatewayConfig.from_dict({"max_concurrent_sessions": None}).max_concurrent_sessions is None
|
|
assert GatewayConfig.from_dict({"max_concurrent_sessions": 0}).max_concurrent_sessions is None
|
|
assert GatewayConfig.from_dict({"max_concurrent_sessions": -1}).max_concurrent_sessions is None
|
|
|
|
def test_max_concurrent_sessions_from_dict_accepts_positive_integer(self):
|
|
config = GatewayConfig.from_dict({"max_concurrent_sessions": "3"})
|
|
|
|
assert config.max_concurrent_sessions == 3
|
|
|
|
def test_max_concurrent_sessions_from_dict_ignores_invalid_values(self, caplog):
|
|
caplog.set_level(logging.WARNING, logger="gateway.config")
|
|
|
|
config = GatewayConfig.from_dict({"max_concurrent_sessions": "many"})
|
|
|
|
assert config.max_concurrent_sessions is None
|
|
assert any(
|
|
"Ignoring invalid max_concurrent_sessions='many'" in record.message
|
|
for record in caplog.records
|
|
)
|
|
|
|
def test_max_concurrent_sessions_from_dict_accepts_nested_fallback(self):
|
|
config = GatewayConfig.from_dict({"gateway": {"max_concurrent_sessions": 4}})
|
|
|
|
assert config.max_concurrent_sessions == 4
|
|
|
|
def test_max_concurrent_sessions_top_level_overrides_nested(self):
|
|
config = GatewayConfig.from_dict(
|
|
{
|
|
"gateway": {"max_concurrent_sessions": 4},
|
|
"max_concurrent_sessions": 2,
|
|
}
|
|
)
|
|
|
|
assert config.max_concurrent_sessions == 2
|
|
|
|
def test_roundtrip_preserves_unauthorized_dm_behavior(self):
|
|
config = GatewayConfig(
|
|
unauthorized_dm_behavior="ignore",
|
|
platforms={
|
|
Platform.WHATSAPP: PlatformConfig(
|
|
enabled=True,
|
|
extra={"unauthorized_dm_behavior": "pair"},
|
|
),
|
|
},
|
|
)
|
|
|
|
restored = GatewayConfig.from_dict(config.to_dict())
|
|
|
|
assert restored.unauthorized_dm_behavior == "ignore"
|
|
assert restored.platforms[Platform.WHATSAPP].extra["unauthorized_dm_behavior"] == "pair"
|
|
|
|
def test_email_defaults_to_ignore_for_unauthorized_dm_behavior(self):
|
|
config = GatewayConfig(
|
|
platforms={Platform.EMAIL: PlatformConfig(enabled=True)},
|
|
)
|
|
|
|
assert config.get_unauthorized_dm_behavior(Platform.EMAIL) == "ignore"
|
|
|
|
def test_email_can_opt_into_pairing_for_unauthorized_dm_behavior(self):
|
|
config = GatewayConfig(
|
|
platforms={
|
|
Platform.EMAIL: PlatformConfig(
|
|
enabled=True,
|
|
extra={"unauthorized_dm_behavior": "pair"},
|
|
),
|
|
},
|
|
)
|
|
|
|
assert config.get_unauthorized_dm_behavior(Platform.EMAIL) == "pair"
|
|
|
|
def test_from_dict_coerces_quoted_false_always_log_local(self):
|
|
restored = GatewayConfig.from_dict({"always_log_local": "false"})
|
|
assert restored.always_log_local is False
|
|
|
|
def test_from_dict_ignores_malformed_nested_sections(self):
|
|
restored = GatewayConfig.from_dict(
|
|
{
|
|
"platforms": {
|
|
"telegram": "enabled",
|
|
"discord": {"enabled": True, "token": "tok"},
|
|
},
|
|
"default_reset_policy": "daily",
|
|
"reset_by_type": ["oops"],
|
|
"reset_by_platform": "oops",
|
|
"streaming": "enabled",
|
|
}
|
|
)
|
|
|
|
assert Platform.TELEGRAM not in restored.platforms
|
|
assert restored.platforms[Platform.DISCORD].enabled is True
|
|
assert restored.default_reset_policy.mode == SessionResetPolicy().mode
|
|
assert restored.reset_by_type == {}
|
|
assert restored.reset_by_platform == {}
|
|
assert restored.streaming.transport == "auto"
|
|
|
|
def test_get_notice_delivery_defaults_to_public(self):
|
|
config = GatewayConfig(
|
|
platforms={Platform.SLACK: PlatformConfig(enabled=True, token="***")}
|
|
)
|
|
|
|
assert config.get_notice_delivery(Platform.SLACK) == "public"
|
|
|
|
def test_get_notice_delivery_honors_platform_override(self):
|
|
config = GatewayConfig(
|
|
platforms={
|
|
Platform.SLACK: PlatformConfig(
|
|
enabled=True,
|
|
token="***",
|
|
extra={"notice_delivery": "private"},
|
|
),
|
|
}
|
|
)
|
|
|
|
assert config.get_notice_delivery(Platform.SLACK) == "private"
|
|
|
|
|
|
class TestLoadGatewayConfig:
|
|
def test_shipped_template_does_not_enable_auto_reset(self, tmp_path, monkeypatch):
|
|
"""A fresh install seeded from cli-config.yaml.example must not
|
|
auto-reset sessions.
|
|
|
|
Installers (scripts/install.sh, scripts/install.ps1,
|
|
docker/stage2-hook.sh, hermes doctor) copy the template verbatim to
|
|
~/.hermes/config.yaml, so whatever ``session_reset.mode`` the template
|
|
ships becomes an EXPLICIT user setting that overrides the code
|
|
default. After #60194 flipped the default to "none", the template
|
|
still said "both" — every new install kept 24h-idle resets on
|
|
(Luciano's report, July 2026). This pins the invariant: template
|
|
seed == no auto-reset.
|
|
"""
|
|
template = (
|
|
Path(__file__).resolve().parents[2] / "cli-config.yaml.example"
|
|
)
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
template.read_text(encoding="utf-8"), encoding="utf-8"
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.default_reset_policy.mode == "none"
|
|
|
|
def test_no_config_yaml_means_no_auto_reset(self, tmp_path, monkeypatch):
|
|
"""With no config.yaml at all, sessions must never auto-reset."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.default_reset_policy.mode == "none"
|
|
|
|
def test_session_reset_without_mode_means_no_auto_reset(self, tmp_path, monkeypatch):
|
|
"""A session_reset block that tunes knobs but omits mode stays off."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"session_reset:\n idle_minutes: 60\n", encoding="utf-8"
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.default_reset_policy.mode == "none"
|
|
assert config.default_reset_policy.idle_minutes == 60
|
|
|
|
def test_explicit_session_reset_opt_in_is_honored(self, tmp_path, monkeypatch):
|
|
"""Users who explicitly opt in to auto-reset keep their policy."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"session_reset:\n mode: idle\n idle_minutes: 30\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.default_reset_policy.mode == "idle"
|
|
assert config.default_reset_policy.idle_minutes == 30
|
|
|
|
def test_bridges_quick_commands_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"quick_commands:\n"
|
|
" limits:\n"
|
|
" type: exec\n"
|
|
" command: echo ok\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.quick_commands == {"limits": {"type": "exec", "command": "echo ok"}}
|
|
|
|
def test_typing_status_text_from_toplevel_platform_block(self, tmp_path, monkeypatch):
|
|
"""A top-level ``slack:`` block reaches PlatformConfig via the
|
|
shared-key bridge (bridged into extra, then the from_dict extra
|
|
fallback) — the route a bare ``hermes config set``-style YAML uses."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
'slack:\n typing_status_text: "is pouncing… 🐾"\n',
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert (
|
|
config.platforms[Platform.SLACK].typing_status_text
|
|
== "is pouncing… 🐾"
|
|
)
|
|
|
|
def test_typing_status_text_from_nested_platforms_block(self, tmp_path, monkeypatch):
|
|
"""``platforms.slack.typing_status_text`` reaches PlatformConfig via
|
|
_merge_platform_map + the from_dict top-level read."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"platforms:\n"
|
|
" slack:\n"
|
|
" enabled: true\n"
|
|
' typing_status_text: "chasing yarn…"\n',
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert (
|
|
config.platforms[Platform.SLACK].typing_status_text == "chasing yarn…"
|
|
)
|
|
|
|
def test_multiplex_profiles_from_nested_gateway_section(self, tmp_path, monkeypatch):
|
|
"""``gateway.multiplex_profiles: true`` (the nested form written by
|
|
``hermes config set gateway.multiplex_profiles true``) must enable
|
|
multiplexing when loaded via load_gateway_config().
|
|
|
|
Regression: load_gateway_config() only surfaced the *top-level*
|
|
``multiplex_profiles`` key into gw_data, so a config.yaml that pinned
|
|
the flag under the nested ``gateway:`` section silently loaded with
|
|
multiplex_profiles=False. (from_dict honors the nested fallback, but
|
|
load_gateway_config builds gw_data from the top-level keys before
|
|
calling from_dict, so the nested value never reached it.)
|
|
"""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n multiplex_profiles: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.multiplex_profiles is True
|
|
|
|
def test_discord_websocket_health_settings_seed_platform_extra(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"discord:\n"
|
|
" websocket_liveness_interval_seconds: 17\n"
|
|
" websocket_liveness_failure_threshold: 4\n"
|
|
" websocket_heartbeat_ack_max_age_seconds: 75\n"
|
|
" websocket_max_latency_seconds: 30\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
for key in (
|
|
"HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS",
|
|
"HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD",
|
|
):
|
|
monkeypatch.delenv(key, raising=False)
|
|
|
|
config = load_gateway_config()
|
|
|
|
extra = config.platforms[Platform.DISCORD].extra
|
|
assert extra["websocket_liveness_interval_seconds"] == 17
|
|
assert extra["websocket_liveness_failure_threshold"] == 4
|
|
assert extra["websocket_heartbeat_ack_max_age_seconds"] == 75
|
|
assert extra["websocket_max_latency_seconds"] == 30
|
|
|
|
def test_session_reset_from_nested_gateway_section(self, tmp_path, monkeypatch):
|
|
"""``gateway.session_reset`` (nested form) must reach default_reset_policy,
|
|
mirroring the gateway.multiplex_profiles precedent."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n session_reset:\n mode: idle\n idle_minutes: 30\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.default_reset_policy.mode == "idle"
|
|
assert config.default_reset_policy.idle_minutes == 30
|
|
|
|
def test_quick_commands_from_nested_gateway_section(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n quick_commands:\n limits:\n type: exec\n command: echo ok\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.quick_commands == {"limits": {"type": "exec", "command": "echo ok"}}
|
|
|
|
def test_stt_from_nested_gateway_section(self, tmp_path, monkeypatch):
|
|
"""Asserts False (not the True default) so the test fails if the
|
|
nested gateway.stt value never reaches from_dict() and silently
|
|
falls back to the class default instead."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n stt:\n enabled: false\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.stt_enabled is False
|
|
|
|
def test_stt_echo_transcripts_from_nested_gateway_section(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n stt_echo_transcripts: false\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.stt_echo_transcripts is False
|
|
|
|
def test_group_sessions_per_user_from_nested_gateway_section(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n group_sessions_per_user: false\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.group_sessions_per_user is False
|
|
|
|
def test_thread_sessions_per_user_from_nested_gateway_section(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n thread_sessions_per_user: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.thread_sessions_per_user is True
|
|
|
|
def test_reset_triggers_from_nested_gateway_section(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n reset_triggers:\n - /new\n - /clear\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.reset_triggers == ["/new", "/clear"]
|
|
|
|
def test_always_log_local_from_nested_gateway_section(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n always_log_local: false\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.always_log_local is False
|
|
|
|
def test_filter_silence_narration_from_nested_gateway_section(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n filter_silence_narration: false\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.filter_silence_narration is False
|
|
|
|
def test_unauthorized_dm_behavior_from_nested_gateway_section(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n unauthorized_dm_behavior: ignore\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.unauthorized_dm_behavior == "ignore"
|
|
|
|
def test_top_level_still_wins_over_nested_gateway_section(self, tmp_path, monkeypatch):
|
|
"""Top-level keys keep precedence over the nested gateway.* fallback
|
|
for every key this fix touches (matches the existing
|
|
gateway.streaming/write_sessions_json precedence contract)."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"always_log_local: true\n"
|
|
"gateway:\n"
|
|
" always_log_local: false\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.always_log_local is True
|
|
|
|
def test_present_empty_top_level_session_reset_blocks_nested_fallback(self, tmp_path, monkeypatch):
|
|
"""Key-presence precedence: a present (even empty) top-level
|
|
session_reset must NOT be replaced by gateway.session_reset —
|
|
the fallback fires only when the top-level key is absent."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"session_reset: {}\n"
|
|
"gateway:\n"
|
|
" session_reset:\n"
|
|
" mode: idle\n"
|
|
" idle_minutes: 30\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
# The nested value must not leak through the present top-level key.
|
|
assert config.default_reset_policy.mode != "idle"
|
|
|
|
def test_present_top_level_stt_blocks_nested_fallback(self, tmp_path, monkeypatch):
|
|
"""Key-presence precedence for stt: a present top-level stt (even
|
|
mistyped/non-dict) must not be replaced by gateway.stt."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"stt: {}\n"
|
|
"gateway:\n"
|
|
" stt:\n"
|
|
" enabled: false\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
# gateway.stt.enabled=false must NOT win over the present top-level stt.
|
|
assert config.stt_enabled is True
|
|
|
|
def test_relay_platform_enabled_from_env_url(self, tmp_path, monkeypatch):
|
|
"""GATEWAY_RELAY_URL must enable Platform.RELAY in config.platforms so
|
|
start_gateway()'s connect loop actually dials the connector. Registering
|
|
the adapter in the platform_registry is NOT enough — the connect loop
|
|
iterates config.platforms, so an un-enabled RELAY never connects (the
|
|
'relay registered but no inbound' bug)."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.setenv("GATEWAY_RELAY_URL", "https://connector.example/relay/")
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert Platform.RELAY in config.platforms
|
|
relay = config.platforms[Platform.RELAY]
|
|
assert relay.enabled is True
|
|
# Trailing slash stripped; mirrored into extra for the connected-checker.
|
|
assert relay.extra.get("relay_url") == "https://connector.example/relay"
|
|
assert Platform.RELAY in config.get_connected_platforms()
|
|
|
|
def test_relay_platform_absent_when_url_unset(self, tmp_path, monkeypatch):
|
|
"""No relay URL -> no RELAY platform, so direct/single-tenant gateways
|
|
are unaffected."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("GATEWAY_RELAY_URL", raising=False)
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert Platform.RELAY not in config.platforms
|
|
|
|
def test_relay_platform_enabled_from_config_yaml(self, tmp_path, monkeypatch):
|
|
"""gateway.relay_url in config.yaml also enables RELAY (env-less path)."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n platforms:\n relay:\n extra:\n relay_url: https://connector.example/relay\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("GATEWAY_RELAY_URL", raising=False)
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert Platform.RELAY in config.platforms
|
|
assert config.platforms[Platform.RELAY].enabled is True
|
|
|
|
def test_bridges_group_sessions_per_user_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text("group_sessions_per_user: false\n", encoding="utf-8")
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.group_sessions_per_user is False
|
|
|
|
def test_bridges_thread_sessions_per_user_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text("thread_sessions_per_user: true\n", encoding="utf-8")
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.thread_sessions_per_user is True
|
|
|
|
def test_thread_sessions_per_user_defaults_to_false(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text("{}\n", encoding="utf-8")
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.thread_sessions_per_user is False
|
|
|
|
def test_bridges_top_level_max_concurrent_sessions_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text("max_concurrent_sessions: 2\n", encoding="utf-8")
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.max_concurrent_sessions == 2
|
|
|
|
def test_bridges_nested_max_concurrent_sessions_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n"
|
|
" max_concurrent_sessions: 3\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.max_concurrent_sessions == 3
|
|
|
|
def test_top_level_max_concurrent_sessions_overrides_nested_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"max_concurrent_sessions: 2\n"
|
|
"gateway:\n"
|
|
" max_concurrent_sessions: 3\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.max_concurrent_sessions == 2
|
|
|
|
def test_scalar_gateway_section_does_not_crash_streaming_fallback(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text("gateway: disabled\n", encoding="utf-8")
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.streaming.transport == "auto"
|
|
|
|
def test_bridges_discord_thread_require_mention_from_config_yaml(self, tmp_path, monkeypatch):
|
|
"""discord.thread_require_mention in config.yaml should reach the runtime env var."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"discord:\n"
|
|
" thread_require_mention: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("DISCORD_THREAD_REQUIRE_MENTION", raising=False)
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.environ.get("DISCORD_THREAD_REQUIRE_MENTION") == "true"
|
|
|
|
def test_thread_require_mention_yaml_does_not_overwrite_env(self, tmp_path, monkeypatch):
|
|
"""Explicit env var should win over config.yaml (env > yaml precedence)."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"discord:\n"
|
|
" thread_require_mention: false\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.setenv("DISCORD_THREAD_REQUIRE_MENTION", "true") # user override
|
|
|
|
load_gateway_config()
|
|
|
|
# Env value preserved, not clobbered by yaml.
|
|
assert os.environ.get("DISCORD_THREAD_REQUIRE_MENTION") == "true"
|
|
|
|
def test_bridges_discord_bots_require_inline_mention_from_config_yaml(self, tmp_path, monkeypatch):
|
|
"""discord.bots_require_inline_mention should reach the runtime env var."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"discord:\n"
|
|
" bots_require_inline_mention: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("DISCORD_BOTS_REQUIRE_INLINE_MENTION", raising=False)
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.environ.get("DISCORD_BOTS_REQUIRE_INLINE_MENTION") == "true"
|
|
|
|
def test_bots_require_inline_mention_yaml_does_not_overwrite_env(self, tmp_path, monkeypatch):
|
|
"""Explicit env var should win over config.yaml for inline bot mention gating."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"discord:\n"
|
|
" bots_require_inline_mention: false\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.setenv("DISCORD_BOTS_REQUIRE_INLINE_MENTION", "true")
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.environ.get("DISCORD_BOTS_REQUIRE_INLINE_MENTION") == "true"
|
|
|
|
def test_bridges_discord_allow_from_from_config_yaml(self, tmp_path, monkeypatch):
|
|
"""discord.allow_from should populate DISCORD_ALLOWED_USERS for auth."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"discord:\n"
|
|
" allow_from:\n"
|
|
" - \"123456789012345678\"\n"
|
|
" - \"999888777666555444\"\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("DISCORD_ALLOWED_USERS", raising=False)
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.platforms[Platform.DISCORD].extra["allow_from"] == [
|
|
"123456789012345678",
|
|
"999888777666555444",
|
|
]
|
|
assert os.environ.get("DISCORD_ALLOWED_USERS") == (
|
|
"123456789012345678,999888777666555444"
|
|
)
|
|
|
|
def test_bridges_discord_platform_extra_allow_from_to_env(self, tmp_path, monkeypatch):
|
|
"""platforms.discord.extra.allow_from should reach DISCORD_ALLOWED_USERS too."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"platforms:\n"
|
|
" discord:\n"
|
|
" extra:\n"
|
|
" allow_from:\n"
|
|
" - \"123456789012345678\"\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("DISCORD_ALLOWED_USERS", raising=False)
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.platforms[Platform.DISCORD].extra["allow_from"] == [
|
|
"123456789012345678",
|
|
]
|
|
assert os.environ.get("DISCORD_ALLOWED_USERS") == "123456789012345678"
|
|
|
|
def test_bridges_nested_gateway_platforms_dingtalk_allowed_users_to_env(self, tmp_path, monkeypatch):
|
|
"""gateway.platforms.dingtalk.extra.allowed_users must reach
|
|
DINGTALK_ALLOWED_USERS — it's the documented config.yaml alternative
|
|
to the env var (website/docs/user-guide/messaging/dingtalk.md), the
|
|
adapter reads it from PlatformConfig.extra, but gateway auth
|
|
(_is_user_authorized) only consults the env var.
|
|
"""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n"
|
|
" platforms:\n"
|
|
" dingtalk:\n"
|
|
" enabled: true\n"
|
|
" extra:\n"
|
|
" allowed_users:\n"
|
|
" - user-id-1\n"
|
|
" - user-id-2\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("DINGTALK_ALLOWED_USERS", raising=False)
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.platforms[Platform.DINGTALK].extra["allowed_users"] == [
|
|
"user-id-1",
|
|
"user-id-2",
|
|
]
|
|
assert os.environ.get("DINGTALK_ALLOWED_USERS") == "user-id-1,user-id-2"
|
|
|
|
def test_bridges_platforms_dingtalk_extra_allowed_users_to_env(self, tmp_path, monkeypatch):
|
|
"""platforms.dingtalk.extra.allowed_users should reach DINGTALK_ALLOWED_USERS too."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"platforms:\n"
|
|
" dingtalk:\n"
|
|
" extra:\n"
|
|
" allowed_users:\n"
|
|
" - manager1234\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("DINGTALK_ALLOWED_USERS", raising=False)
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.platforms[Platform.DINGTALK].extra["allowed_users"] == [
|
|
"manager1234",
|
|
]
|
|
assert os.environ.get("DINGTALK_ALLOWED_USERS") == "manager1234"
|
|
|
|
def test_dingtalk_allowed_users_env_takes_precedence_over_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n"
|
|
" platforms:\n"
|
|
" dingtalk:\n"
|
|
" extra:\n"
|
|
" allowed_users:\n"
|
|
" - config-user\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.setenv("DINGTALK_ALLOWED_USERS", "env-user")
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.environ.get("DINGTALK_ALLOWED_USERS") == "env-user"
|
|
|
|
def test_top_level_dingtalk_allowed_users_wins_over_nested_extra(self, tmp_path, monkeypatch):
|
|
"""The legacy top-level dingtalk: block keeps precedence over the
|
|
nested platform extra when both define an allowlist."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"dingtalk:\n"
|
|
" allowed_users:\n"
|
|
" - top-level-user\n"
|
|
"gateway:\n"
|
|
" platforms:\n"
|
|
" dingtalk:\n"
|
|
" extra:\n"
|
|
" allowed_users:\n"
|
|
" - nested-user\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("DINGTALK_ALLOWED_USERS", raising=False)
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.environ.get("DINGTALK_ALLOWED_USERS") == "top-level-user"
|
|
|
|
def test_nested_dingtalk_allowlist_authorizes_listed_user_only(self, tmp_path, monkeypatch):
|
|
"""E2E for the documented setup: a nested-only allowlist must
|
|
authorize the listed user at the gateway and still deny others.
|
|
|
|
Before the bridge existed, the listed user passed the adapter's
|
|
_is_user_allowed() but _is_user_authorized() fell through to
|
|
default-deny because DINGTALK_ALLOWED_USERS was never populated.
|
|
"""
|
|
from types import SimpleNamespace
|
|
|
|
from gateway.run import GatewayRunner
|
|
from gateway.session import SessionSource
|
|
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n"
|
|
" platforms:\n"
|
|
" dingtalk:\n"
|
|
" enabled: true\n"
|
|
" extra:\n"
|
|
" allowed_users:\n"
|
|
" - user-id-1\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
for var in (
|
|
"DINGTALK_ALLOWED_USERS",
|
|
"DINGTALK_ALLOW_ALL_USERS",
|
|
"GATEWAY_ALLOWED_USERS",
|
|
"GATEWAY_ALLOW_ALL_USERS",
|
|
):
|
|
monkeypatch.delenv(var, raising=False)
|
|
|
|
config = load_gateway_config()
|
|
|
|
runner = object.__new__(GatewayRunner)
|
|
runner.pairing_store = SimpleNamespace(is_approved=lambda *_a, **_kw: False)
|
|
runner.config = config
|
|
|
|
def _dm_source(user_id):
|
|
return SessionSource(
|
|
platform=Platform.DINGTALK,
|
|
chat_id="dm-1",
|
|
chat_type="dm",
|
|
user_id=user_id,
|
|
user_name="someone",
|
|
)
|
|
|
|
assert runner._is_user_authorized(_dm_source("user-id-1")) is True
|
|
assert runner._is_user_authorized(_dm_source("intruder")) is False
|
|
|
|
def test_bridges_quoted_false_platform_enabled_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"platforms:\n"
|
|
" api_server:\n"
|
|
" enabled: \"false\"\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.platforms[Platform.API_SERVER].enabled is False
|
|
assert Platform.API_SERVER not in config.get_connected_platforms()
|
|
|
|
def test_bridges_nested_gateway_platforms_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n"
|
|
" platforms:\n"
|
|
" telegram:\n"
|
|
" enabled: true\n"
|
|
" token: nested-token\n"
|
|
" home_channel:\n"
|
|
" platform: telegram\n"
|
|
" chat_id: \"123\"\n"
|
|
" name: Nested Home\n"
|
|
" extra:\n"
|
|
" reply_prefix: nested\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
telegram = config.platforms[Platform.TELEGRAM]
|
|
assert telegram.enabled is True
|
|
assert telegram.token == "nested-token"
|
|
assert telegram.home_channel == HomeChannel(
|
|
platform=Platform.TELEGRAM,
|
|
chat_id="123",
|
|
name="Nested Home",
|
|
)
|
|
assert telegram.extra["reply_prefix"] == "nested"
|
|
|
|
def test_top_level_platforms_override_nested_gateway_platforms(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n"
|
|
" platforms:\n"
|
|
" telegram:\n"
|
|
" enabled: false\n"
|
|
" token: nested-token\n"
|
|
" extra:\n"
|
|
" reply_prefix: nested\n"
|
|
"platforms:\n"
|
|
" telegram:\n"
|
|
" enabled: true\n"
|
|
" token: top-token\n"
|
|
" extra:\n"
|
|
" reply_prefix: top\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
telegram = config.platforms[Platform.TELEGRAM]
|
|
assert telegram.enabled is True
|
|
assert telegram.token == "top-token"
|
|
assert telegram.extra["reply_prefix"] == "top"
|
|
|
|
def test_shared_key_loop_bridges_allow_from_from_nested_platforms(self, tmp_path, monkeypatch):
|
|
"""Regression: shared-key loop must bridge allow_from / require_mention
|
|
into PlatformConfig.extra even when the platform is configured only
|
|
under ``platforms:`` (no top-level ``telegram:`` block).
|
|
|
|
Before the fix, ``platform_cfg = yaml_cfg.get('telegram')`` returned
|
|
None for nested-only configs, so the loop skipped the platform entirely
|
|
and allow_from was silently ignored. The apply_yaml_config_fn dispatch
|
|
received the same fix in #44f3e51; the shared-key loop now mirrors it.
|
|
"""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"platforms:\n"
|
|
" telegram:\n"
|
|
" allow_from:\n"
|
|
" - \"111222333\"\n"
|
|
" - \"444555666\"\n"
|
|
" require_mention: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
telegram = config.platforms[Platform.TELEGRAM]
|
|
assert telegram.extra.get("allow_from") == ["111222333", "444555666"], (
|
|
"allow_from configured under platforms.telegram must be bridged "
|
|
"into PlatformConfig.extra by the shared-key loop"
|
|
)
|
|
assert telegram.extra.get("require_mention") is True, (
|
|
"require_mention configured under platforms.telegram must be "
|
|
"bridged into PlatformConfig.extra by the shared-key loop"
|
|
)
|
|
|
|
def test_shared_key_loop_bridges_allow_from_from_nested_gateway_platforms(self, tmp_path, monkeypatch):
|
|
"""Same regression check for ``gateway.platforms:`` path."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n"
|
|
" platforms:\n"
|
|
" telegram:\n"
|
|
" allow_from:\n"
|
|
" - \"777888999\"\n"
|
|
" require_mention: false\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
telegram = config.platforms[Platform.TELEGRAM]
|
|
assert telegram.extra.get("allow_from") == ["777888999"], (
|
|
"allow_from configured under plugins.platforms.telegram.adapter must be "
|
|
"bridged into PlatformConfig.extra by the shared-key loop"
|
|
)
|
|
assert telegram.extra.get("require_mention") is False
|
|
|
|
def test_bridges_quoted_false_session_notify_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"session_reset:\n"
|
|
" notify: \"false\"\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.default_reset_policy.notify is False
|
|
|
|
def test_bridges_quoted_false_always_log_local_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"always_log_local: \"false\"\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.always_log_local is False
|
|
|
|
def test_bridges_discord_channel_overrides_from_top_level_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"discord:\n"
|
|
" channel_overrides:\n"
|
|
' "1234567890":\n'
|
|
" model: openrouter/healer-alpha\n"
|
|
" provider: openrouter\n"
|
|
" system_prompt: Daily news summarizer\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
discord = config.platforms[Platform.DISCORD]
|
|
assert "1234567890" in discord.channel_overrides
|
|
ov = discord.channel_overrides["1234567890"]
|
|
assert ov.model == "openrouter/healer-alpha"
|
|
assert ov.provider == "openrouter"
|
|
assert ov.system_prompt == "Daily news summarizer"
|
|
|
|
def test_bridges_discord_channel_prompts_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"discord:\n"
|
|
" channel_prompts:\n"
|
|
" \"123\": Research mode\n"
|
|
" 456: Therapist mode\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.platforms[Platform.DISCORD].extra["channel_prompts"] == {
|
|
"123": "Research mode",
|
|
"456": "Therapist mode",
|
|
}
|
|
|
|
def test_bridges_discord_history_backfill_settings_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"discord:\n"
|
|
" history_backfill: true\n"
|
|
" history_backfill_limit: 17\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("DISCORD_HISTORY_BACKFILL", raising=False)
|
|
monkeypatch.delenv("DISCORD_HISTORY_BACKFILL_LIMIT", raising=False)
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.getenv("DISCORD_HISTORY_BACKFILL") == "true"
|
|
assert os.getenv("DISCORD_HISTORY_BACKFILL_LIMIT") == "17"
|
|
|
|
def test_bridges_telegram_channel_prompts_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"telegram:\n"
|
|
" channel_prompts:\n"
|
|
' "-1001234567": Research assistant\n'
|
|
" 789: Creative writing\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.platforms[Platform.TELEGRAM].extra["channel_prompts"] == {
|
|
"-1001234567": "Research assistant",
|
|
"789": "Creative writing",
|
|
}
|
|
|
|
def test_bridges_slack_channel_prompts_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"slack:\n"
|
|
" channel_prompts:\n"
|
|
' "C01ABC": Code review mode\n',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.platforms[Platform.SLACK].extra["channel_prompts"] == {
|
|
"C01ABC": "Code review mode",
|
|
}
|
|
|
|
def test_bridges_feishu_allow_bots_from_config_yaml_to_env(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"feishu:\n allow_bots: mentions\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("FEISHU_ALLOW_BOTS", raising=False)
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.environ.get("FEISHU_ALLOW_BOTS") == "mentions"
|
|
|
|
def test_feishu_allow_bots_env_takes_precedence_over_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"feishu:\n allow_bots: all\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.setenv("FEISHU_ALLOW_BOTS", "none")
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.environ.get("FEISHU_ALLOW_BOTS") == "none"
|
|
|
|
def test_bridges_telegram_allow_bots_from_config_yaml_to_env(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"telegram:\n allow_bots: mentions\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("TELEGRAM_ALLOW_BOTS", raising=False)
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.environ.get("TELEGRAM_ALLOW_BOTS") == "mentions"
|
|
|
|
def test_telegram_allow_bots_env_takes_precedence_over_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"telegram:\n allow_bots: all\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.setenv("TELEGRAM_ALLOW_BOTS", "none")
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.environ.get("TELEGRAM_ALLOW_BOTS") == "none"
|
|
|
|
def test_invalid_quick_commands_in_config_yaml_are_ignored(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text("quick_commands: not-a-mapping\n", encoding="utf-8")
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.quick_commands == {}
|
|
|
|
def test_bridges_unauthorized_dm_behavior_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"unauthorized_dm_behavior: ignore\n"
|
|
"whatsapp:\n"
|
|
" unauthorized_dm_behavior: pair\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.unauthorized_dm_behavior == "ignore"
|
|
assert config.platforms[Platform.WHATSAPP].extra["unauthorized_dm_behavior"] == "pair"
|
|
|
|
def test_bridges_telegram_disable_link_previews_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"telegram:\n"
|
|
" disable_link_previews: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.platforms[Platform.TELEGRAM].extra["disable_link_previews"] is True
|
|
|
|
def test_loads_telegram_rich_messages_from_gateway_platform_extra(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n"
|
|
" platforms:\n"
|
|
" telegram:\n"
|
|
" extra:\n"
|
|
" rich_messages: false\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.platforms[Platform.TELEGRAM].extra["rich_messages"] is False
|
|
|
|
def test_loads_telegram_rich_drafts_from_gateway_platform_extra(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n"
|
|
" platforms:\n"
|
|
" telegram:\n"
|
|
" extra:\n"
|
|
" rich_drafts: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.platforms[Platform.TELEGRAM].extra["rich_drafts"] is True
|
|
|
|
def test_load_config_default_keeps_telegram_rich_messages_opt_in(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
from hermes_cli.config import load_config
|
|
|
|
config = load_config()
|
|
|
|
assert config["telegram"]["extra"]["rich_messages"] is False
|
|
assert config["telegram"]["extra"]["rich_drafts"] is False
|
|
|
|
def test_bridges_telegram_extra_base_url_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"telegram:\n"
|
|
" extra:\n"
|
|
" base_url: https://custom-proxy.example.com/bot\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert (
|
|
config.platforms[Platform.TELEGRAM].extra["base_url"]
|
|
== "https://custom-proxy.example.com/bot"
|
|
)
|
|
|
|
def test_bridges_notice_delivery_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"slack:\n"
|
|
" notice_delivery: private\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.get_notice_delivery(Platform.SLACK) == "private"
|
|
|
|
def test_bridges_telegram_proxy_url_from_config_yaml(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"telegram:\n"
|
|
" proxy_url: socks5://127.0.0.1:1080\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("TELEGRAM_PROXY", raising=False)
|
|
|
|
load_gateway_config()
|
|
|
|
import os
|
|
assert os.environ.get("TELEGRAM_PROXY") == "socks5://127.0.0.1:1080"
|
|
|
|
def test_telegram_proxy_env_takes_precedence_over_config(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"telegram:\n"
|
|
" proxy_url: http://from-config:8080\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.setenv("TELEGRAM_PROXY", "socks5://from-env:1080")
|
|
|
|
load_gateway_config()
|
|
|
|
import os
|
|
assert os.environ.get("TELEGRAM_PROXY") == "socks5://from-env:1080"
|
|
|
|
def test_profile_scoped_env_overrides_do_not_fall_back_to_default_profile_env(
|
|
self,
|
|
tmp_path,
|
|
monkeypatch,
|
|
):
|
|
default_home = tmp_path / "default-home"
|
|
default_home.mkdir()
|
|
default_config = default_home / "config.yaml"
|
|
default_config.write_text(
|
|
"multiplex_profiles: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
secondary_home = tmp_path / "secondary-home"
|
|
secondary_home.mkdir()
|
|
secondary_config = secondary_home / "config.yaml"
|
|
secondary_config.write_text(
|
|
"multiplex_profiles: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(default_home))
|
|
monkeypatch.setenv("API_SERVER_ENABLED", "true")
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "default-token")
|
|
|
|
# Model the real multiplexed gateway: run.py flips the runtime flag at
|
|
# startup (set_multiplex_active) BEFORE any secondary-profile config
|
|
# loads. Without the flag, a scope miss legitimately falls through to
|
|
# os.environ (single-profile overlay semantics, #67827) and this test
|
|
# would no longer exercise the cross-profile isolation it's about.
|
|
set_multiplex_active(True)
|
|
home_token = set_hermes_home_override(str(secondary_home))
|
|
secret_token = set_secret_scope({"DISCORD_BOT_TOKEN": "worker-token"})
|
|
try:
|
|
config = load_gateway_config()
|
|
finally:
|
|
reset_secret_scope(secret_token)
|
|
reset_hermes_home_override(home_token)
|
|
set_multiplex_active(False)
|
|
|
|
assert config.multiplex_profiles is True
|
|
assert config.platforms[Platform.DISCORD].token == "worker-token"
|
|
assert Platform.API_SERVER not in config.platforms
|
|
|
|
|
|
class TestHomeChannelEnvOverrides:
|
|
"""Home channel env vars should apply even when the platform was already
|
|
configured via config.yaml (not just when credential env vars create it)."""
|
|
|
|
def test_existing_platform_configs_accept_home_channel_env_overrides(self):
|
|
cases = [
|
|
(
|
|
Platform.SLACK,
|
|
PlatformConfig(enabled=True, token="xoxb-from-config"),
|
|
{"SLACK_HOME_CHANNEL": "C123", "SLACK_HOME_CHANNEL_NAME": "Ops"},
|
|
("C123", "Ops"),
|
|
),
|
|
(
|
|
Platform.WHATSAPP,
|
|
PlatformConfig(enabled=True),
|
|
{
|
|
"WHATSAPP_HOME_CHANNEL": "1234567890@lid",
|
|
"WHATSAPP_HOME_CHANNEL_NAME": "Owner DM",
|
|
},
|
|
("1234567890@lid", "Owner DM"),
|
|
),
|
|
(
|
|
Platform.SIGNAL,
|
|
PlatformConfig(
|
|
enabled=True,
|
|
extra={"http_url": "http://localhost:9090", "account": "+15551234567"},
|
|
),
|
|
{"SIGNAL_HOME_CHANNEL": "+1555000", "SIGNAL_HOME_CHANNEL_NAME": "Phone"},
|
|
("+1555000", "Phone"),
|
|
),
|
|
(
|
|
Platform.MATTERMOST,
|
|
PlatformConfig(
|
|
enabled=True,
|
|
token="mm-token",
|
|
extra={"url": "https://mm.example.com"},
|
|
),
|
|
{"MATTERMOST_HOME_CHANNEL": "ch_abc123", "MATTERMOST_HOME_CHANNEL_NAME": "General"},
|
|
("ch_abc123", "General"),
|
|
),
|
|
(
|
|
Platform.MATRIX,
|
|
PlatformConfig(
|
|
enabled=True,
|
|
token="syt_abc123",
|
|
extra={"homeserver": "https://matrix.example.org"},
|
|
),
|
|
{"MATRIX_HOME_ROOM": "!room123:example.org", "MATRIX_HOME_ROOM_NAME": "Bot Room"},
|
|
("!room123:example.org", "Bot Room"),
|
|
),
|
|
(
|
|
Platform.EMAIL,
|
|
PlatformConfig(
|
|
enabled=True,
|
|
extra={
|
|
"address": "hermes@test.com",
|
|
"imap_host": "imap.test.com",
|
|
"smtp_host": "smtp.test.com",
|
|
},
|
|
),
|
|
{"EMAIL_HOME_ADDRESS": "user@test.com", "EMAIL_HOME_ADDRESS_NAME": "Inbox"},
|
|
("user@test.com", "Inbox"),
|
|
),
|
|
(
|
|
Platform.SMS,
|
|
PlatformConfig(enabled=True, api_key="token_abc"),
|
|
{"SMS_HOME_CHANNEL": "+15559876543", "SMS_HOME_CHANNEL_NAME": "My Phone"},
|
|
("+15559876543", "My Phone"),
|
|
),
|
|
]
|
|
|
|
for platform, platform_config, env, expected in cases:
|
|
config = GatewayConfig(platforms={platform: platform_config})
|
|
with patch.dict(os.environ, env, clear=True):
|
|
_apply_env_overrides(config)
|
|
|
|
home = config.platforms[platform].home_channel
|
|
assert home is not None, f"{platform.value}: home_channel should not be None"
|
|
assert (home.chat_id, home.name) == expected, platform.value
|
|
|
|
|
|
class TestMultiplexProfilesEnvOverride:
|
|
"""GATEWAY_MULTIPLEX_PROFILES env override — the 3-tier precedence chain.
|
|
|
|
env (recognized token) > config.yaml (top-level or nested gateway.*) >
|
|
default False. A blank / unrecognized env value is treated as UNSET and
|
|
falls through to config (the empty-secret trap: a provisioned-but-empty Fly
|
|
secret arrives as "" and must not shadow a config.yaml opt-in).
|
|
"""
|
|
|
|
def _load(self, tmp_path, monkeypatch, config_text=None):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir(exist_ok=True)
|
|
if config_text is not None:
|
|
(hermes_home / "config.yaml").write_text(config_text, encoding="utf-8")
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
return load_gateway_config()
|
|
|
|
# ── Tier 1: env wins ──────────────────────────────────────────────────
|
|
def test_env_true_forces_on_with_no_config(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("GATEWAY_MULTIPLEX_PROFILES", "true")
|
|
config = self._load(tmp_path, monkeypatch, config_text=None)
|
|
assert config.multiplex_profiles is True
|
|
|
|
def test_env_true_overrides_config_false(self, tmp_path, monkeypatch):
|
|
# THE discriminating test: env-set wins over an explicit config value.
|
|
monkeypatch.setenv("GATEWAY_MULTIPLEX_PROFILES", "1")
|
|
config = self._load(
|
|
tmp_path, monkeypatch, config_text="multiplex_profiles: false\n"
|
|
)
|
|
assert config.multiplex_profiles is True
|
|
|
|
def test_env_false_overrides_config_true(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("GATEWAY_MULTIPLEX_PROFILES", "off")
|
|
config = self._load(
|
|
tmp_path, monkeypatch, config_text="multiplex_profiles: true\n"
|
|
)
|
|
assert config.multiplex_profiles is False
|
|
|
|
# ── Tier 2: config.yaml when env unset ────────────────────────────────
|
|
def test_config_true_when_env_unset(self, tmp_path, monkeypatch):
|
|
monkeypatch.delenv("GATEWAY_MULTIPLEX_PROFILES", raising=False)
|
|
config = self._load(
|
|
tmp_path, monkeypatch, config_text="multiplex_profiles: true\n"
|
|
)
|
|
assert config.multiplex_profiles is True
|
|
|
|
# ── The empty / unrecognized env trap: fall through, don't force off ──
|
|
def test_empty_env_does_not_shadow_config_true(self, tmp_path, monkeypatch):
|
|
# Provisioned-but-unpopulated Fly secret arrives as "". It must NOT
|
|
# turn OFF a config.yaml opt-in.
|
|
monkeypatch.setenv("GATEWAY_MULTIPLEX_PROFILES", "")
|
|
config = self._load(
|
|
tmp_path, monkeypatch, config_text="multiplex_profiles: true\n"
|
|
)
|
|
assert config.multiplex_profiles is True
|
|
|
|
def test_whitespace_env_does_not_shadow_config_true(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("GATEWAY_MULTIPLEX_PROFILES", " ")
|
|
config = self._load(
|
|
tmp_path, monkeypatch, config_text="multiplex_profiles: true\n"
|
|
)
|
|
assert config.multiplex_profiles is True
|
|
|
|
def test_unrecognized_env_falls_through_to_config(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("GATEWAY_MULTIPLEX_PROFILES", "maybe")
|
|
config = self._load(
|
|
tmp_path, monkeypatch, config_text="multiplex_profiles: true\n"
|
|
)
|
|
assert config.multiplex_profiles is True
|
|
|
|
# ── Tier 3: default False ─────────────────────────────────────────────
|
|
def test_default_false_when_neither_set(self, tmp_path, monkeypatch):
|
|
monkeypatch.delenv("GATEWAY_MULTIPLEX_PROFILES", raising=False)
|
|
config = self._load(tmp_path, monkeypatch, config_text=None)
|
|
assert config.multiplex_profiles is False
|
|
|
|
# ── The resolver in isolation ─────────────────────────────────────────
|
|
def test_resolver_tristate(self, monkeypatch):
|
|
from gateway.config import _env_multiplex_profiles_override
|
|
|
|
monkeypatch.delenv("GATEWAY_MULTIPLEX_PROFILES", raising=False)
|
|
assert _env_multiplex_profiles_override() is None
|
|
for truthy in ("1", "true", "TRUE", "yes", "on"):
|
|
monkeypatch.setenv("GATEWAY_MULTIPLEX_PROFILES", truthy)
|
|
assert _env_multiplex_profiles_override() is True, truthy
|
|
for falsy in ("0", "false", "FALSE", "no", "off"):
|
|
monkeypatch.setenv("GATEWAY_MULTIPLEX_PROFILES", falsy)
|
|
assert _env_multiplex_profiles_override() is False, falsy
|
|
for noise in ("", " ", "maybe", "2"):
|
|
monkeypatch.setenv("GATEWAY_MULTIPLEX_PROFILES", noise)
|
|
assert _env_multiplex_profiles_override() is None, repr(noise)
|
|
|
|
|
|
class TestMultiplexProfilesConfig:
|
|
"""Tests for parsing multiplex_profiles (top-level and nested forms)."""
|
|
|
|
def test_multiplex_profiles_top_level(self, tmp_path, monkeypatch):
|
|
"""Top-level multiplex_profiles is honored."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"multiplex_profiles: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.multiplex_profiles is True
|
|
|
|
def test_multiplex_profiles_nested_under_gateway(self, tmp_path, monkeypatch):
|
|
"""gateway.multiplex_profiles (the form written by `hermes config set
|
|
gateway.multiplex_profiles true`) must be honored. Regression test for
|
|
the silent-fallback bug where the loader only forwarded the top-level
|
|
key, so users who wrote it under gateway: got multiplex_profiles=False
|
|
with no warning."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"gateway:\n multiplex_profiles: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.multiplex_profiles is True, (
|
|
"gateway.multiplex_profiles: true was silently ignored — "
|
|
"loader only forwarded the top-level form"
|
|
)
|
|
|
|
def test_multiplex_profiles_default_false(self, tmp_path, monkeypatch):
|
|
"""Default is False when neither form is present."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text("", encoding="utf-8")
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.multiplex_profiles is False
|
|
|
|
def test_multiplex_profiles_top_level_overrides_nested(self, tmp_path, monkeypatch):
|
|
"""When both forms are present, top-level wins (matches profile_routes
|
|
and other parity bridges in load_gateway_config)."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"multiplex_profiles: true\n"
|
|
"gateway:\n multiplex_profiles: false\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.multiplex_profiles is True
|
|
|
|
def test_multiplex_profiles_explicit_top_level_false_not_consulting_nested(
|
|
self, tmp_path, monkeypatch
|
|
):
|
|
"""Lock in the `is None` vs `is False` distinction: when top-level is
|
|
explicitly false, the loader must forward False WITHOUT consulting the
|
|
nested form (so a stale `gateway.multiplex_profiles: true` cannot
|
|
silently re-enable multiplexing). Guards against a future regression
|
|
that flips the check to `not _mp`."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"multiplex_profiles: false\n"
|
|
"gateway:\n multiplex_profiles: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.multiplex_profiles is False, (
|
|
"Explicit top-level false was overridden by nested true — "
|
|
"loader must respect top-level precedence when key is present"
|
|
)
|