"""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. 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_slack_disable_dms_config_sets_env_bridge(self, tmp_path, monkeypatch): hermes_home = tmp_path / ".hermes" hermes_home.mkdir() config_path = hermes_home / "config.yaml" config_path.write_text( "slack:\n" " disable_dms: true\n", encoding="utf-8", ) monkeypatch.setenv("HERMES_HOME", str(hermes_home)) monkeypatch.delenv("SLACK_DISABLE_DMS", raising=False) load_gateway_config() assert os.getenv("SLACK_DISABLE_DMS") == "true" 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_slack_ignored_channels_env_takes_precedence(self, tmp_path, monkeypatch): """An explicit SLACK_IGNORED_CHANNELS env var must not be overwritten by the config.yaml bridge.""" hermes_home = tmp_path / ".hermes" hermes_home.mkdir() (hermes_home / "config.yaml").write_text( "slack:\n" " ignored_channels: C_FROM_YAML\n", encoding="utf-8", ) monkeypatch.setenv("HERMES_HOME", str(hermes_home)) monkeypatch.setenv("SLACK_IGNORED_CHANNELS", "C_FROM_ENV") load_gateway_config() assert os.getenv("SLACK_IGNORED_CHANNELS") == "C_FROM_ENV" 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 @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. 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_api_server_explicit_extra_wins_over_toplevel_key(self, tmp_path, monkeypatch): """An explicit ``extra: {port: X}`` must beat a sibling top-level ``port:`` β€” the bridge's ``not in _api_extra`` guard must never clobber a value the user placed in extra deliberately.""" 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: 9999\n" " extra:\n" " port: 8642\n", encoding="utf-8", ) monkeypatch.setenv("HERMES_HOME", str(hermes_home)) config = load_gateway_config() assert config.platforms[Platform.API_SERVER].extra["port"] == 8642 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_api_server_via_gateway_platforms_block_still_works(self, tmp_path, monkeypatch): """No regression: the pre-existing ``gateway.platforms.api_server`` path must keep working alongside the new nested discovery, and its api_server keys get the same extra bridge.""" self._clear_api_server_env(monkeypatch) hermes_home = tmp_path / ".hermes" hermes_home.mkdir() (hermes_home / "config.yaml").write_text( "gateway:\n" " platforms:\n" " api_server:\n" " enabled: true\n" " port: 8643\n", encoding="utf-8", ) monkeypatch.setenv("HERMES_HOME", str(hermes_home)) config = load_gateway_config() assert config.platforms[Platform.API_SERVER].enabled is True assert config.platforms[Platform.API_SERVER].extra["port"] == 8643 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 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_webhook_port_in_extra_not_overwritten_by_toplevel(self, tmp_path, monkeypatch): """If port is already under extra, the top-level value must not clobber it.""" 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" " extra:\n" " port: 8650\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() wh = config.platforms[Platform.WEBHOOK] assert wh.extra.get("port") == 8650 def test_api_server_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" " api_server:\n" " enabled: true\n" " host: 127.0.0.1\n" " port: 8648\n", encoding="utf-8", ) monkeypatch.setenv("HERMES_HOME", str(hermes_home)) monkeypatch.delenv("API_SERVER_ENABLED", raising=False) monkeypatch.delenv("API_SERVER_PORT", raising=False) config = load_gateway_config() assert Platform.API_SERVER in config.platforms api = config.platforms[Platform.API_SERVER] assert api.extra.get("port") == 8648 assert api.extra.get("host") == "127.0.0.1" def test_webhook_port_defaults_when_not_configured(self, tmp_path, monkeypatch): """No port anywhere -> adapter uses its hardcoded DEFAULT_PORT.""" 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", 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() wh = config.platforms[Platform.WEBHOOK] assert "port" not in wh.extra or wh.extra.get("port") is None 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_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" ) 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