1555 lines
62 KiB
Python
1555 lines
62 KiB
Python
"""Tests for gateway configuration management."""
|
|
|
|
import logging
|
|
import os
|
|
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,
|
|
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",
|
|
user_id="user-123",
|
|
scope_id="guild-456",
|
|
)
|
|
d = hc.to_dict()
|
|
restored = HomeChannel.from_dict(d)
|
|
|
|
assert restored.platform == Platform.DISCORD
|
|
assert restored.chat_id == "999"
|
|
assert restored.name == "general"
|
|
assert restored.user_id == "user-123"
|
|
assert restored.scope_id == "guild-456"
|
|
|
|
|
|
class TestPlatformConfigRoundtrip:
|
|
def test_toplevel_adapter_keys_promoted_into_extra(self):
|
|
"""Adapter settings written directly under the platform block (the documented
|
|
``platforms.webhook.port`` shape) reach ``extra``; an explicit ``extra:`` value wins and
|
|
typed fields never leak into ``extra`` (#10206)."""
|
|
pc = PlatformConfig.from_dict({
|
|
"enabled": True, "reply_to_mode": "all", "typing_indicator": False,
|
|
"port": 9100, "routes": {"gh": {"prompt": "x"}}, "extra": {"port": 9999},
|
|
})
|
|
assert pc.extra == {"port": 9999, "routes": {"gh": {"prompt": "x"}}}
|
|
assert pc.reply_to_mode == "all" and pc.typing_indicator is False
|
|
assert PlatformConfig.from_dict(pc.to_dict()).extra == pc.extra
|
|
|
|
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_from_dict_coerces_quoted_false_enabled(self):
|
|
restored = PlatformConfig.from_dict({"enabled": "false"})
|
|
assert restored.enabled is False
|
|
|
|
|
|
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_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_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"
|
|
|
|
|
|
class TestChannelOverride:
|
|
def test_from_dict_empty(self):
|
|
assert ChannelOverride.from_dict({}).model is None
|
|
assert ChannelOverride.from_dict(None).model is None
|
|
|
|
|
|
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_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()
|
|
|
|
|
|
class TestStreamingConfig:
|
|
|
|
|
|
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",
|
|
}
|
|
)
|
|
defaults = StreamingConfig()
|
|
assert restored.edit_interval == defaults.edit_interval
|
|
assert restored.buffer_threshold == defaults.buffer_threshold
|
|
assert restored.fresh_final_after_seconds == defaults.fresh_final_after_seconds
|
|
|
|
|
|
class TestGatewayConfigRoundtrip:
|
|
|
|
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_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
|
|
|
|
|
|
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"
|
|
|
|
|
|
class TestLoadGatewayConfig:
|
|
def test_platforms_env_refs_expanded_for_adapters(self, tmp_path, monkeypatch):
|
|
"""``${VAR}`` refs under ``platforms:`` reach the adapter config expanded — the gateway
|
|
YAML layer expands them the same way the CLI loader does (webhook secret used as the
|
|
HMAC key; api_server caller-auth key)."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"platforms:\n"
|
|
" webhook:\n"
|
|
" enabled: true\n"
|
|
" port: 8089\n"
|
|
" secret: ${WEBHOOK_SECRET}\n"
|
|
" api_server:\n"
|
|
" enabled: true\n"
|
|
" key: ${env:API_SERVER_KEY}\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.setenv("WEBHOOK_SECRET", "whsec-expanded")
|
|
monkeypatch.setenv("API_SERVER_KEY", "server-key-expanded")
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.platforms[Platform.WEBHOOK].extra["secret"] == "whsec-expanded"
|
|
assert (
|
|
config.platforms[Platform.API_SERVER].extra["key"] == "server-key-expanded"
|
|
)
|
|
|
|
def test_platforms_env_ref_unresolved_stays_literal(self, tmp_path, monkeypatch):
|
|
"""An unset env var keeps the literal placeholder (loader is fail-open; the adapter's
|
|
startup validation is what reports a bad secret)."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"platforms:\n"
|
|
" webhook:\n"
|
|
" enabled: true\n"
|
|
" secret: ${WEBHOOK_SECRET_UNSET_FOR_TEST}\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("WEBHOOK_SECRET_UNSET_FOR_TEST", raising=False)
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert (
|
|
config.platforms[Platform.WEBHOOK].extra["secret"]
|
|
== "${WEBHOOK_SECRET_UNSET_FOR_TEST}"
|
|
)
|
|
|
|
def test_platforms_plain_values_untouched_by_expansion(self, tmp_path, monkeypatch):
|
|
"""Configs without ``${VAR}`` refs load byte-identically through the same path."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"platforms:\n"
|
|
" webhook:\n"
|
|
" enabled: true\n"
|
|
" port: 8089\n"
|
|
" secret: plain-secret-value\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert (
|
|
config.platforms[Platform.WEBHOOK].extra["secret"] == "plain-secret-value"
|
|
)
|
|
assert config.platforms[Platform.WEBHOOK].extra["port"] == 8089
|
|
|
|
def test_slack_ignored_channels_config_sets_env_bridge(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"slack:\n"
|
|
" ignored_channels:\n"
|
|
" - C0123456789\n"
|
|
" - C0987654321\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("SLACK_IGNORED_CHANNELS", raising=False)
|
|
|
|
load_gateway_config()
|
|
|
|
assert os.getenv("SLACK_IGNORED_CHANNELS") == "C0123456789,C0987654321"
|
|
|
|
|
|
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_stale_multiplex_allowlist_key_is_ignored(self, tmp_path, monkeypatch):
|
|
# The removed ``multiplex_profile_allowlist`` key may linger in an un-migrated
|
|
# config.yaml; it must not break loading or the multiplex flag.
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"gateway:\n"
|
|
" multiplex_profiles: true\n"
|
|
" multiplex_profile_allowlist:\n"
|
|
" - worker\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.multiplex_profiles is True
|
|
assert not hasattr(config, "multiplex_profile_allowlist")
|
|
|
|
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"
|
|
" websocket_event_max_silence_seconds: 7200\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
|
|
assert extra["websocket_event_max_silence_seconds"] == 7200
|
|
|
|
|
|
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
|
|
|
|
|
|
@staticmethod
|
|
def _clear_api_server_env(monkeypatch):
|
|
"""Keep _apply_env_overrides from masking the YAML path under test."""
|
|
for key in (
|
|
"API_SERVER_ENABLED",
|
|
"API_SERVER_KEY",
|
|
"API_SERVER_PORT",
|
|
"API_SERVER_HOST",
|
|
"API_SERVER_CORS_ORIGINS",
|
|
"API_SERVER_MODEL_NAME",
|
|
):
|
|
monkeypatch.delenv(key, raising=False)
|
|
|
|
def test_api_server_from_nested_gateway_section(self, tmp_path, monkeypatch):
|
|
"""``gateway.api_server:`` (nested YAML form) must be discovered and
|
|
enable the platform.
|
|
|
|
Regression for #66630: load_gateway_config handled gateway.streaming
|
|
and gateway.platforms.* but silently dropped nested gateway.<platform>
|
|
blocks, so ``gateway.api_server.enabled: true`` never started the API
|
|
server unless API_SERVER_* env vars were also set.
|
|
"""
|
|
self._clear_api_server_env(monkeypatch)
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"gateway:\n api_server:\n enabled: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert Platform.API_SERVER in config.platforms
|
|
assert config.platforms[Platform.API_SERVER].enabled is True
|
|
|
|
def test_api_server_port_bridged_into_extra(self, tmp_path, monkeypatch):
|
|
"""``gateway.api_server.port`` must land in PlatformConfig.extra —
|
|
the adapter reads port/key/host/cors_origins/model_name from extra
|
|
(gateway/platforms/api_server.py), and from_dict discards unknown
|
|
top-level keys, so without the bridge the port is silently lost."""
|
|
self._clear_api_server_env(monkeypatch)
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"gateway:\n"
|
|
" api_server:\n"
|
|
" enabled: true\n"
|
|
" port: 8642\n"
|
|
" host: 0.0.0.0\n"
|
|
" key: sekrit\n"
|
|
" model_name: my-hermes\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
extra = config.platforms[Platform.API_SERVER].extra
|
|
assert extra["port"] == 8642
|
|
assert extra["host"] == "0.0.0.0"
|
|
assert extra["key"] == "sekrit"
|
|
assert extra["model_name"] == "my-hermes"
|
|
|
|
def test_room_link_url_from_nested_gateway_section(self, tmp_path, monkeypatch):
|
|
"""The supported config path advertises no endpoint until restart."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"gateway:\n"
|
|
" room_link_url: https://peer.example.test/hermes\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.room_link_url == "https://peer.example.test/hermes"
|
|
assert GatewayConfig.from_dict(config.to_dict()).room_link_url == (
|
|
"https://peer.example.test/hermes"
|
|
)
|
|
|
|
|
|
def test_non_platform_gateway_keys_not_misparsed_as_platforms(self, tmp_path, monkeypatch):
|
|
"""Nested-platform discovery must only pick up keys matching the
|
|
Platform enum: ``gateway.streaming`` / ``gateway.timeout`` must not
|
|
be turned into phantom platform entries or break loading."""
|
|
self._clear_api_server_env(monkeypatch)
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"gateway:\n"
|
|
" streaming:\n"
|
|
" enabled: false\n"
|
|
" timeout: 120\n"
|
|
" api_server:\n"
|
|
" enabled: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
config = load_gateway_config()
|
|
|
|
# streaming still parsed as the streaming subsection, not a platform
|
|
assert config.streaming.enabled is False
|
|
# only real platforms present; nothing named streaming/timeout leaked in
|
|
assert all(isinstance(p, Platform) for p in config.platforms)
|
|
assert config.platforms[Platform.API_SERVER].enabled is True
|
|
|
|
|
|
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_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_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_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_env_url_disables_other_messaging_platforms(self, tmp_path, monkeypatch):
|
|
"""A GATEWAY_RELAY_URL env stamp means the connector owns all platform
|
|
connections: directly-connected messaging platforms must be disabled,
|
|
even when explicitly enabled in config.yaml, while non-messaging
|
|
surfaces (api_server et al.) survive."""
|
|
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"
|
|
" bot_token: '123:abc'\n"
|
|
" api_server:\n"
|
|
" enabled: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.setenv("GATEWAY_RELAY_URL", "https://connector.example/relay")
|
|
# Credential-based auto-enable path must be suppressed too.
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "fake-token-for-test")
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.platforms[Platform.RELAY].enabled is True
|
|
assert config.platforms[Platform.TELEGRAM].enabled is False
|
|
discord_cfg = config.platforms.get(Platform.DISCORD)
|
|
assert discord_cfg is None or discord_cfg.enabled is False
|
|
# Non-messaging surfaces are untouched.
|
|
assert config.platforms[Platform.API_SERVER].enabled is True
|
|
|
|
|
|
def test_relay_yaml_url_keeps_other_platforms_enabled(self, tmp_path, monkeypatch):
|
|
"""gateway.relay_url in config.yaml (no env stamp) keeps the old
|
|
additive behavior: relay runs beside directly-connected platforms."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"gateway:\n"
|
|
" platforms:\n"
|
|
" relay:\n"
|
|
" enabled: true\n"
|
|
" extra:\n"
|
|
" relay_url: https://connector.example/relay\n"
|
|
" telegram:\n"
|
|
" enabled: true\n"
|
|
" bot_token: '123:abc'\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("GATEWAY_RELAY_URL", raising=False)
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.platforms[Platform.RELAY].enabled is True
|
|
assert config.platforms[Platform.TELEGRAM].enabled is True
|
|
|
|
|
|
def test_relay_env_url_opt_out_keeps_direct_platforms(self, tmp_path, monkeypatch):
|
|
"""GATEWAY_RELAY_ALLOW_DIRECT_PLATFORMS=true opts a deployment out of
|
|
the relay-exclusive sweep: direct adapters stay enabled beside the
|
|
relay even with the GATEWAY_RELAY_URL env stamp present."""
|
|
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"
|
|
" bot_token: '123:abc'\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.setenv("GATEWAY_RELAY_URL", "https://connector.example/relay")
|
|
monkeypatch.setenv("GATEWAY_RELAY_ALLOW_DIRECT_PLATFORMS", "true")
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert config.platforms[Platform.RELAY].enabled is True
|
|
assert config.platforms[Platform.TELEGRAM].enabled is True
|
|
|
|
|
|
def test_relay_exclusive_reads_profile_scoped_env(self, tmp_path, monkeypatch):
|
|
"""GATEWAY_RELAY_URL is a process-global deployment stamp (like the
|
|
API_SERVER listener vars, #69379): the scoped runner reload in a
|
|
multiplexed gateway must keep seeing it, so config's relay enablement
|
|
and sweep agree with gateway.relay.relay_url()/register_relay_adapter()
|
|
(which read os.environ). A secondary profile's .env stamp does NOT
|
|
override the shared process stamp — an isolated multiplex scope is
|
|
never consulted for globals. (A single-profile gateway, or the profile
|
|
the process is launched under, still activates via its .env because
|
|
load_hermes_dotenv exports that file into os.environ at startup.)"""
|
|
from agent import secret_scope as ss
|
|
|
|
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"
|
|
" bot_token: '123:abc'\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
# Managed-deploy stamp in the process env; the profile .env has none.
|
|
monkeypatch.setenv("GATEWAY_RELAY_URL", "https://deploy.example/relay")
|
|
|
|
profile_dir = tmp_path / "profile-a"
|
|
profile_dir.mkdir()
|
|
(profile_dir / ".env").write_text("", encoding="utf-8")
|
|
|
|
ss.set_multiplex_active(True)
|
|
tok = ss.set_secret_scope(ss.build_profile_secret_scope(profile_dir))
|
|
try:
|
|
config = load_gateway_config()
|
|
finally:
|
|
ss.reset_secret_scope(tok)
|
|
ss.set_multiplex_active(False)
|
|
|
|
# The deploy stamp survives the profile scope: relay enabled, sweep
|
|
# ran — matching what register_relay_adapter() sees in os.environ.
|
|
assert config.platforms[Platform.RELAY].enabled is True
|
|
assert config.platforms[Platform.TELEGRAM].enabled is False
|
|
|
|
# A profile-only stamp does NOT activate relay: globals read only the
|
|
# process env, so config and the relay transport can never disagree.
|
|
monkeypatch.delenv("GATEWAY_RELAY_URL", raising=False)
|
|
(profile_dir / ".env").write_text(
|
|
"GATEWAY_RELAY_URL=https://profile.example/relay\n", encoding="utf-8"
|
|
)
|
|
ss.set_multiplex_active(True)
|
|
tok = ss.set_secret_scope(ss.build_profile_secret_scope(profile_dir))
|
|
try:
|
|
config = load_gateway_config()
|
|
finally:
|
|
ss.reset_secret_scope(tok)
|
|
ss.set_multiplex_active(False)
|
|
|
|
relay_cfg = config.platforms.get(Platform.RELAY)
|
|
assert relay_cfg is None or relay_cfg.enabled is False
|
|
assert config.platforms[Platform.TELEGRAM].enabled is True
|
|
|
|
|
|
def test_relay_exclusive_sweep_log_levels_and_marker_cleanup(self, tmp_path, monkeypatch, caplog):
|
|
"""Explicitly-enabled platforms are disabled at WARNING, auto-enabled
|
|
ones at INFO, and the _enabled_explicit marker never survives config
|
|
load."""
|
|
import logging as _logging
|
|
|
|
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"
|
|
" bot_token: '123:abc'\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.setenv("GATEWAY_RELAY_URL", "https://connector.example/relay")
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "fake-token-for-test")
|
|
|
|
with caplog.at_level(_logging.INFO, logger="gateway.config"):
|
|
config = load_gateway_config()
|
|
|
|
warnings = [
|
|
r for r in caplog.records
|
|
if r.levelno == _logging.WARNING and "telegram" in r.getMessage()
|
|
]
|
|
assert warnings, "explicit platform must be disabled at WARNING"
|
|
discord_infos = [
|
|
r for r in caplog.records
|
|
if r.levelno == _logging.INFO and "discord" in r.getMessage()
|
|
]
|
|
if config.platforms.get(Platform.DISCORD) is not None:
|
|
assert discord_infos, "auto-enabled platform must be disabled at INFO"
|
|
|
|
for pc in config.platforms.values():
|
|
assert "_enabled_explicit" not in (pc.extra or {})
|
|
|
|
|
|
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_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"
|
|
|
|
@pytest.mark.parametrize("yaml_text", ["gateway:\n allow_all_users: true\n", "allow_all_users: true\n"])
|
|
def test_allow_all_users_yaml_reaches_the_authz_gate(self, tmp_path, monkeypatch, yaml_text):
|
|
"""Both spellings must open the gate for an unknown sender; before #110690 the key was inert
|
|
because every allow-all reader consults GATEWAY_ALLOW_ALL_USERS only."""
|
|
from gateway.authz_mixin import GatewayAuthorizationMixin
|
|
from gateway.session import SessionSource
|
|
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(yaml_text, encoding="utf-8")
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("GATEWAY_ALLOW_ALL_USERS", raising=False)
|
|
|
|
runner = object.__new__(GatewayAuthorizationMixin)
|
|
runner.config = load_gateway_config()
|
|
runner.adapters = {}
|
|
stranger = SessionSource(platform=Platform.TELEGRAM, user_id="999", chat_id="999", chat_type="dm")
|
|
assert runner._is_user_authorized(stranger) is True
|
|
|
|
@pytest.mark.parametrize("yaml_text, env, expected", [
|
|
("gateway:\n allow_all_users: false\n", None, None), # only a truthy grant is exported
|
|
("gateway:\n allow_all_users: true\n", "false", "false"), # explicit env wins over YAML
|
|
("gateway: {}\n", None, None),
|
|
])
|
|
def test_allow_all_users_yaml_never_widens_past_env(self, tmp_path, monkeypatch, yaml_text, env, expected):
|
|
from gateway.authz_mixin import GatewayAuthorizationMixin
|
|
from gateway.session import SessionSource
|
|
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(yaml_text, encoding="utf-8")
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
if env is None:
|
|
monkeypatch.delenv("GATEWAY_ALLOW_ALL_USERS", raising=False)
|
|
else:
|
|
monkeypatch.setenv("GATEWAY_ALLOW_ALL_USERS", env)
|
|
|
|
runner = object.__new__(GatewayAuthorizationMixin)
|
|
runner.config = load_gateway_config()
|
|
runner.adapters = {}
|
|
assert os.environ.get("GATEWAY_ALLOW_ALL_USERS") == expected
|
|
stranger = SessionSource(platform=Platform.TELEGRAM, user_id="999", chat_id="999", chat_type="dm")
|
|
assert runner._is_user_authorized(stranger) is False
|
|
|
|
def test_bridged_allow_all_users_does_not_survive_a_config_flip_or_restart(self, tmp_path, monkeypatch):
|
|
"""The bridge owns what it wrote: flipping config.yaml to false and reloading closes the gate,
|
|
and the restart/dashboard child envs never carry the bridged value (a sticky env var would make
|
|
the restarted gateway ignore the flipped config and stay open)."""
|
|
from gateway.run_shutdown import GatewayShutdownMixin
|
|
from hermes_cli.web_server_gateway import _profile_action_environment
|
|
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text("gateway:\n allow_all_users: true\n", encoding="utf-8")
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("GATEWAY_ALLOW_ALL_USERS", raising=False)
|
|
load_gateway_config()
|
|
assert os.environ.get("GATEWAY_ALLOW_ALL_USERS") == "true"
|
|
assert "GATEWAY_ALLOW_ALL_USERS" not in GatewayShutdownMixin._restart_watcher_env()
|
|
assert "GATEWAY_ALLOW_ALL_USERS" not in _profile_action_environment(["gateway", "restart"])
|
|
|
|
(hermes_home / "config.yaml").write_text("gateway:\n allow_all_users: false\n", encoding="utf-8")
|
|
load_gateway_config()
|
|
assert os.environ.get("GATEWAY_ALLOW_ALL_USERS") is None
|
|
|
|
def test_allow_all_users_yaml_reaches_the_default_profile_under_multiplex(self, tmp_path, monkeypatch):
|
|
"""Default-profile events are authorized inside its secret scope, where gate readers never fall to
|
|
os.environ: the bridged grant must be part of that profile's scope (and only that profile's)."""
|
|
from agent.secret_scope import build_profile_secret_scope, set_multiplex_active
|
|
from gateway.authz_mixin import GatewayAuthorizationMixin
|
|
from gateway.run import _profile_runtime_scope
|
|
from gateway.session import SessionSource
|
|
|
|
hermes_home = tmp_path / ".hermes"
|
|
secondary = hermes_home / "profiles" / "other"
|
|
secondary.mkdir(parents=True)
|
|
(hermes_home / "config.yaml").write_text(
|
|
"gateway:\n allow_all_users: true\n multiplex_profiles: true\n", encoding="utf-8")
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("GATEWAY_ALLOW_ALL_USERS", raising=False)
|
|
runner = object.__new__(GatewayAuthorizationMixin)
|
|
runner.config = load_gateway_config()
|
|
runner.adapters = {}
|
|
set_multiplex_active(True)
|
|
try:
|
|
stranger = SessionSource(platform=Platform.TELEGRAM, user_id="999", chat_id="999", chat_type="dm")
|
|
with _profile_runtime_scope(hermes_home):
|
|
assert runner._is_user_authorized(stranger) is True
|
|
assert "GATEWAY_ALLOW_ALL_USERS" not in build_profile_secret_scope(secondary)
|
|
finally:
|
|
set_multiplex_active(False)
|
|
|
|
|
|
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_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_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_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 TestWebhookPortBridging:
|
|
"""Top-level port/host in the YAML platform section must be bridged into
|
|
PlatformConfig.extra so the webhook/api_server adapters can read them.
|
|
|
|
The adapters (WebhookAdapter, ApiServerAdapter) read port/host from
|
|
``config.extra``, but users naturally write them at the top level of the
|
|
platform section in config.yaml:
|
|
|
|
platforms:
|
|
webhook:
|
|
enabled: true
|
|
host: 0.0.0.0
|
|
port: 8649
|
|
|
|
Without bridging, the port silently falls back to DEFAULT_PORT (8644),
|
|
causing port conflicts between profiles that configure different ports."""
|
|
|
|
def test_webhook_port_bridged_from_toplevel(self, tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"platforms:\n"
|
|
" webhook:\n"
|
|
" enabled: true\n"
|
|
" host: 0.0.0.0\n"
|
|
" port: 8649\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("WEBHOOK_ENABLED", raising=False)
|
|
monkeypatch.delenv("WEBHOOK_PORT", raising=False)
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert Platform.WEBHOOK in config.platforms
|
|
wh = config.platforms[Platform.WEBHOOK]
|
|
assert wh.enabled is True
|
|
assert wh.extra.get("port") == 8649
|
|
assert wh.extra.get("host") == "0.0.0.0"
|
|
|
|
|
|
def test_root_level_platform_block_adapter_keys_reach_extra(self, tmp_path, monkeypatch):
|
|
"""A ROOT-level ``webhook:`` block (not under ``platforms:``) is a supported spelling; its
|
|
adapter keys must reach ``extra`` like the nested form, with nested ``extra:`` winning."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"webhook:\n enabled: true\n port: 9100\n host: 127.0.0.2\n secret: fixture\n"
|
|
" extra:\n port: 9999\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("WEBHOOK_PORT", raising=False)
|
|
wh = load_gateway_config().platforms[Platform.WEBHOOK]
|
|
assert (wh.extra.get("port"), wh.extra.get("host"), wh.extra.get("secret")) == (9999, "127.0.0.2", "fixture")
|
|
|
|
def test_msgraph_webhook_port_host_secret_bridged_from_toplevel(self, tmp_path, monkeypatch):
|
|
"""msgraph_webhook top-level port/host/secret must be bridged into extra,
|
|
with an explicit extra: value still winning over the top-level one."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
config_path = hermes_home / "config.yaml"
|
|
config_path.write_text(
|
|
"platforms:\n"
|
|
" msgraph_webhook:\n"
|
|
" enabled: true\n"
|
|
" host: 0.0.0.0\n"
|
|
" port: 8651\n"
|
|
" secret: toplevel-secret\n"
|
|
" extra:\n"
|
|
" client_state: my-client-state\n"
|
|
" secret: extra-secret\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("MSGRAPH_WEBHOOK_ENABLED", raising=False)
|
|
monkeypatch.delenv("MSGRAPH_WEBHOOK_PORT", raising=False)
|
|
monkeypatch.delenv("MSGRAPH_WEBHOOK_CLIENT_STATE", raising=False)
|
|
|
|
config = load_gateway_config()
|
|
|
|
assert Platform.MSGRAPH_WEBHOOK in config.platforms
|
|
ms = config.platforms[Platform.MSGRAPH_WEBHOOK]
|
|
assert ms.enabled is True
|
|
assert ms.extra.get("port") == 8651
|
|
assert ms.extra.get("host") == "0.0.0.0"
|
|
# explicit extra: wins over top-level
|
|
assert ms.extra.get("secret") == "extra-secret"
|
|
assert ms.extra.get("client_state") == "my-client-state"
|
|
|
|
|
|
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_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
|
|
|
|
|
|
# ── 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
|
|
|
|
|
|
# ── Tier 3: default False ─────────────────────────────────────────────
|
|
|
|
# ── The resolver in isolation ─────────────────────────────────────────
|
|
|
|
|
|
class TestMultiplexProfilesConfig:
|
|
"""Tests for parsing multiplex_profiles (top-level and nested forms)."""
|
|
|
|
|
|
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_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"
|
|
)
|
|
|
|
|
|
class TestApiServerEnvOverride:
|
|
def test_env_key_does_not_reenable_explicitly_disabled_api_server(self):
|
|
"""An explicit ``platforms.api_server.enabled: false`` must survive
|
|
_apply_env_overrides() even when API_SERVER_KEY is present in the env.
|
|
|
|
Regression: _apply_env_overrides() force-set api_server.enabled = True
|
|
whenever API_SERVER_KEY (or API_SERVER_ENABLED) was set. In multiplex
|
|
mode a secondary profile pins ``api_server.enabled: false`` so it shares
|
|
the default profile's listener instead of binding its own port, but it
|
|
still inherits the process-level API_SERVER_KEY. The unconditional
|
|
re-enable flipped it back on and tripped the MultiplexConfigError check.
|
|
|
|
The fix honors the explicit disable, flagged by ``_enabled_explicit`` in
|
|
the platform's extra (set when the config.yaml pins enabled).
|
|
"""
|
|
config = GatewayConfig(
|
|
platforms={
|
|
Platform.API_SERVER: PlatformConfig(
|
|
enabled=False,
|
|
extra={"_enabled_explicit": True},
|
|
),
|
|
},
|
|
)
|
|
|
|
api_server_key = "secret-key-at-least-16"
|
|
with patch.dict(os.environ, {"API_SERVER_KEY": api_server_key}, clear=True):
|
|
_apply_env_overrides(config)
|
|
|
|
# Explicit disable wins over the env-var presence.
|
|
assert config.platforms[Platform.API_SERVER].enabled is False
|
|
# The key is still wired through for the shared listener.
|
|
assert config.platforms[Platform.API_SERVER].extra.get("key") == api_server_key
|
|
|
|
|
|
class TestWebhookEnvOverride:
|
|
def test_config_enabled_webhook_reads_env_port_and_secret(self, tmp_path, monkeypatch):
|
|
"""A config.yaml-enabled webhook still receives its .env listener settings."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"platforms:\n"
|
|
" webhook:\n"
|
|
" enabled: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
monkeypatch.delenv("WEBHOOK_ENABLED", raising=False)
|
|
monkeypatch.setenv("WEBHOOK_PORT", "9012")
|
|
monkeypatch.setenv("WEBHOOK_SECRET", "webhook-env-secret")
|
|
|
|
webhook = load_gateway_config().platforms[Platform.WEBHOOK]
|
|
|
|
assert webhook.enabled is True
|
|
assert webhook.extra["port"] == 9012
|
|
assert webhook.extra["secret"] == "webhook-env-secret"
|
|
|
|
def test_env_key_does_not_reenable_explicitly_disabled_webhook(self):
|
|
"""An explicit ``platforms.webhook.enabled: false`` must survive
|
|
_apply_env_overrides() even when WEBHOOK_ENABLED is truthy in the env.
|
|
|
|
Regression (#85637): _apply_env_overrides() force-set
|
|
webhook.enabled = True whenever WEBHOOK_ENABLED was truthy. In
|
|
multiplex mode a secondary profile pins ``webhook.enabled: false`` so
|
|
it shares the default profile's listener instead of binding its own
|
|
port, but it still inherits the process-level WEBHOOK_ENABLED
|
|
(or carries one in its own .env). The unconditional re-enable
|
|
flipped it back on and tripped the MultiplexConfigError check.
|
|
|
|
The fix honors the explicit disable, flagged by ``_enabled_explicit``
|
|
in the platform's extra (set when the config.yaml pins enabled).
|
|
The MSGRAPH_WEBHOOK branch shares the shape and the fix.
|
|
"""
|
|
config = GatewayConfig(
|
|
platforms={
|
|
Platform.WEBHOOK: PlatformConfig(
|
|
enabled=False,
|
|
extra={"_enabled_explicit": True},
|
|
),
|
|
Platform.MSGRAPH_WEBHOOK: PlatformConfig(
|
|
enabled=False,
|
|
extra={"_enabled_explicit": True},
|
|
),
|
|
},
|
|
)
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"WEBHOOK_ENABLED": "true",
|
|
"WEBHOOK_PORT": "9999",
|
|
"WEBHOOK_SECRET": "shared-secret",
|
|
"MSGRAPH_WEBHOOK_ENABLED": "true",
|
|
"MSGRAPH_WEBHOOK_PORT": "9998",
|
|
},
|
|
clear=True,
|
|
):
|
|
_apply_env_overrides(config)
|
|
|
|
# Explicit disable wins over the env-var presence.
|
|
assert config.platforms[Platform.WEBHOOK].enabled is False
|
|
assert config.platforms[Platform.MSGRAPH_WEBHOOK].enabled is False
|
|
assert config.platforms[Platform.MSGRAPH_WEBHOOK].extra.get("port") == 9998
|
|
# Port/secret are still wired through for the shared listener.
|
|
assert config.platforms[Platform.WEBHOOK].extra.get("port") == 9999
|
|
assert (
|
|
config.platforms[Platform.WEBHOOK].extra.get("secret")
|
|
== "shared-secret"
|
|
)
|