test_config.py
1 """Tests for gateway configuration management.""" 2 3 import os 4 from unittest.mock import patch 5 6 from gateway.config import ( 7 GatewayConfig, 8 HomeChannel, 9 Platform, 10 PlatformConfig, 11 SessionResetPolicy, 12 StreamingConfig, 13 _apply_env_overrides, 14 load_gateway_config, 15 ) 16 17 18 class TestHomeChannelRoundtrip: 19 def test_to_dict_from_dict(self): 20 hc = HomeChannel(platform=Platform.DISCORD, chat_id="999", name="general") 21 d = hc.to_dict() 22 restored = HomeChannel.from_dict(d) 23 24 assert restored.platform == Platform.DISCORD 25 assert restored.chat_id == "999" 26 assert restored.name == "general" 27 28 29 class TestPlatformConfigRoundtrip: 30 def test_to_dict_from_dict(self): 31 pc = PlatformConfig( 32 enabled=True, 33 token="tok_123", 34 home_channel=HomeChannel( 35 platform=Platform.TELEGRAM, 36 chat_id="555", 37 name="Home", 38 ), 39 extra={"foo": "bar"}, 40 ) 41 d = pc.to_dict() 42 restored = PlatformConfig.from_dict(d) 43 44 assert restored.enabled is True 45 assert restored.token == "tok_123" 46 assert restored.home_channel.chat_id == "555" 47 assert restored.extra == {"foo": "bar"} 48 49 def test_disabled_no_token(self): 50 pc = PlatformConfig() 51 d = pc.to_dict() 52 restored = PlatformConfig.from_dict(d) 53 assert restored.enabled is False 54 assert restored.token is None 55 56 def test_from_dict_coerces_quoted_false_enabled(self): 57 restored = PlatformConfig.from_dict({"enabled": "false"}) 58 assert restored.enabled is False 59 60 61 class TestGetConnectedPlatforms: 62 def test_returns_enabled_with_token(self): 63 config = GatewayConfig( 64 platforms={ 65 Platform.TELEGRAM: PlatformConfig(enabled=True, token="t"), 66 Platform.DISCORD: PlatformConfig(enabled=False, token="d"), 67 Platform.SLACK: PlatformConfig(enabled=True), # no token 68 }, 69 ) 70 connected = config.get_connected_platforms() 71 assert Platform.TELEGRAM in connected 72 assert Platform.DISCORD not in connected 73 assert Platform.SLACK not in connected 74 75 def test_empty_platforms(self): 76 config = GatewayConfig() 77 assert config.get_connected_platforms() == [] 78 79 def test_dingtalk_recognised_via_extras(self): 80 config = GatewayConfig( 81 platforms={ 82 Platform.DINGTALK: PlatformConfig( 83 enabled=True, 84 extra={"client_id": "cid", "client_secret": "sec"}, 85 ), 86 }, 87 ) 88 assert Platform.DINGTALK in config.get_connected_platforms() 89 90 def test_dingtalk_recognised_via_env_vars(self, monkeypatch): 91 """DingTalk configured via env vars (no extras) should still be 92 recognised as connected — covers the case where _apply_env_overrides 93 hasn't populated extras yet.""" 94 monkeypatch.setenv("DINGTALK_CLIENT_ID", "env_cid") 95 monkeypatch.setenv("DINGTALK_CLIENT_SECRET", "env_sec") 96 config = GatewayConfig( 97 platforms={ 98 Platform.DINGTALK: PlatformConfig(enabled=True, extra={}), 99 }, 100 ) 101 assert Platform.DINGTALK in config.get_connected_platforms() 102 103 def test_dingtalk_missing_creds_not_connected(self, monkeypatch): 104 monkeypatch.delenv("DINGTALK_CLIENT_ID", raising=False) 105 monkeypatch.delenv("DINGTALK_CLIENT_SECRET", raising=False) 106 config = GatewayConfig( 107 platforms={ 108 Platform.DINGTALK: PlatformConfig(enabled=True, extra={}), 109 }, 110 ) 111 assert Platform.DINGTALK not in config.get_connected_platforms() 112 113 def test_dingtalk_disabled_not_connected(self): 114 config = GatewayConfig( 115 platforms={ 116 Platform.DINGTALK: PlatformConfig( 117 enabled=False, 118 extra={"client_id": "cid", "client_secret": "sec"}, 119 ), 120 }, 121 ) 122 assert Platform.DINGTALK not in config.get_connected_platforms() 123 124 125 class TestSessionResetPolicy: 126 def test_roundtrip(self): 127 policy = SessionResetPolicy(mode="idle", at_hour=6, idle_minutes=120) 128 d = policy.to_dict() 129 restored = SessionResetPolicy.from_dict(d) 130 assert restored.mode == "idle" 131 assert restored.at_hour == 6 132 assert restored.idle_minutes == 120 133 134 def test_defaults(self): 135 policy = SessionResetPolicy() 136 assert policy.mode == "both" 137 assert policy.at_hour == 4 138 assert policy.idle_minutes == 1440 139 140 def test_from_dict_treats_null_values_as_defaults(self): 141 restored = SessionResetPolicy.from_dict( 142 {"mode": None, "at_hour": None, "idle_minutes": None} 143 ) 144 assert restored.mode == "both" 145 assert restored.at_hour == 4 146 assert restored.idle_minutes == 1440 147 148 def test_from_dict_coerces_quoted_false_notify(self): 149 restored = SessionResetPolicy.from_dict({"notify": "false"}) 150 assert restored.notify is False 151 152 153 class TestStreamingConfig: 154 def test_from_dict_coerces_quoted_false_enabled(self): 155 restored = StreamingConfig.from_dict({"enabled": "false"}) 156 assert restored.enabled is False 157 158 def test_from_dict_malformed_numeric_values_fall_back_to_defaults(self): 159 restored = StreamingConfig.from_dict( 160 { 161 "edit_interval": "oops", 162 "buffer_threshold": "oops", 163 "fresh_final_after_seconds": "oops", 164 } 165 ) 166 assert restored.edit_interval == 1.0 167 assert restored.buffer_threshold == 40 168 assert restored.fresh_final_after_seconds == 60.0 169 170 171 class TestGatewayConfigRoundtrip: 172 def test_full_roundtrip(self): 173 config = GatewayConfig( 174 platforms={ 175 Platform.TELEGRAM: PlatformConfig( 176 enabled=True, 177 token="tok_123", 178 home_channel=HomeChannel(Platform.TELEGRAM, "123", "Home"), 179 ), 180 }, 181 reset_triggers=["/new"], 182 quick_commands={"limits": {"type": "exec", "command": "echo ok"}}, 183 group_sessions_per_user=False, 184 thread_sessions_per_user=True, 185 ) 186 d = config.to_dict() 187 restored = GatewayConfig.from_dict(d) 188 189 assert Platform.TELEGRAM in restored.platforms 190 assert restored.platforms[Platform.TELEGRAM].token == "tok_123" 191 assert restored.reset_triggers == ["/new"] 192 assert restored.quick_commands == {"limits": {"type": "exec", "command": "echo ok"}} 193 assert restored.group_sessions_per_user is False 194 assert restored.thread_sessions_per_user is True 195 196 def test_roundtrip_preserves_unauthorized_dm_behavior(self): 197 config = GatewayConfig( 198 unauthorized_dm_behavior="ignore", 199 platforms={ 200 Platform.WHATSAPP: PlatformConfig( 201 enabled=True, 202 extra={"unauthorized_dm_behavior": "pair"}, 203 ), 204 }, 205 ) 206 207 restored = GatewayConfig.from_dict(config.to_dict()) 208 209 assert restored.unauthorized_dm_behavior == "ignore" 210 assert restored.platforms[Platform.WHATSAPP].extra["unauthorized_dm_behavior"] == "pair" 211 212 def test_from_dict_coerces_quoted_false_always_log_local(self): 213 restored = GatewayConfig.from_dict({"always_log_local": "false"}) 214 assert restored.always_log_local is False 215 216 def test_get_notice_delivery_defaults_to_public(self): 217 config = GatewayConfig( 218 platforms={Platform.SLACK: PlatformConfig(enabled=True, token="***")} 219 ) 220 221 assert config.get_notice_delivery(Platform.SLACK) == "public" 222 223 def test_get_notice_delivery_honors_platform_override(self): 224 config = GatewayConfig( 225 platforms={ 226 Platform.SLACK: PlatformConfig( 227 enabled=True, 228 token="***", 229 extra={"notice_delivery": "private"}, 230 ), 231 } 232 ) 233 234 assert config.get_notice_delivery(Platform.SLACK) == "private" 235 236 237 class TestLoadGatewayConfig: 238 def test_bridges_quick_commands_from_config_yaml(self, tmp_path, monkeypatch): 239 hermes_home = tmp_path / ".hermes" 240 hermes_home.mkdir() 241 config_path = hermes_home / "config.yaml" 242 config_path.write_text( 243 "quick_commands:\n" 244 " limits:\n" 245 " type: exec\n" 246 " command: echo ok\n", 247 encoding="utf-8", 248 ) 249 250 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 251 252 config = load_gateway_config() 253 254 assert config.quick_commands == {"limits": {"type": "exec", "command": "echo ok"}} 255 256 def test_bridges_group_sessions_per_user_from_config_yaml(self, tmp_path, monkeypatch): 257 hermes_home = tmp_path / ".hermes" 258 hermes_home.mkdir() 259 config_path = hermes_home / "config.yaml" 260 config_path.write_text("group_sessions_per_user: false\n", encoding="utf-8") 261 262 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 263 264 config = load_gateway_config() 265 266 assert config.group_sessions_per_user is False 267 268 def test_bridges_thread_sessions_per_user_from_config_yaml(self, tmp_path, monkeypatch): 269 hermes_home = tmp_path / ".hermes" 270 hermes_home.mkdir() 271 config_path = hermes_home / "config.yaml" 272 config_path.write_text("thread_sessions_per_user: true\n", encoding="utf-8") 273 274 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 275 276 config = load_gateway_config() 277 278 assert config.thread_sessions_per_user is True 279 280 def test_thread_sessions_per_user_defaults_to_false(self, tmp_path, monkeypatch): 281 hermes_home = tmp_path / ".hermes" 282 hermes_home.mkdir() 283 config_path = hermes_home / "config.yaml" 284 config_path.write_text("{}\n", encoding="utf-8") 285 286 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 287 288 config = load_gateway_config() 289 290 assert config.thread_sessions_per_user is False 291 292 def test_bridges_quoted_false_platform_enabled_from_config_yaml(self, tmp_path, monkeypatch): 293 hermes_home = tmp_path / ".hermes" 294 hermes_home.mkdir() 295 config_path = hermes_home / "config.yaml" 296 config_path.write_text( 297 "platforms:\n" 298 " api_server:\n" 299 " enabled: \"false\"\n", 300 encoding="utf-8", 301 ) 302 303 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 304 305 config = load_gateway_config() 306 307 assert config.platforms[Platform.API_SERVER].enabled is False 308 assert Platform.API_SERVER not in config.get_connected_platforms() 309 310 def test_bridges_quoted_false_session_notify_from_config_yaml(self, tmp_path, monkeypatch): 311 hermes_home = tmp_path / ".hermes" 312 hermes_home.mkdir() 313 config_path = hermes_home / "config.yaml" 314 config_path.write_text( 315 "session_reset:\n" 316 " notify: \"false\"\n", 317 encoding="utf-8", 318 ) 319 320 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 321 322 config = load_gateway_config() 323 324 assert config.default_reset_policy.notify is False 325 326 def test_bridges_quoted_false_always_log_local_from_config_yaml(self, tmp_path, monkeypatch): 327 hermes_home = tmp_path / ".hermes" 328 hermes_home.mkdir() 329 config_path = hermes_home / "config.yaml" 330 config_path.write_text( 331 "always_log_local: \"false\"\n", 332 encoding="utf-8", 333 ) 334 335 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 336 337 config = load_gateway_config() 338 339 assert config.always_log_local is False 340 341 def test_bridges_discord_channel_prompts_from_config_yaml(self, tmp_path, monkeypatch): 342 hermes_home = tmp_path / ".hermes" 343 hermes_home.mkdir() 344 config_path = hermes_home / "config.yaml" 345 config_path.write_text( 346 "discord:\n" 347 " channel_prompts:\n" 348 " \"123\": Research mode\n" 349 " 456: Therapist mode\n", 350 encoding="utf-8", 351 ) 352 353 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 354 355 config = load_gateway_config() 356 357 assert config.platforms[Platform.DISCORD].extra["channel_prompts"] == { 358 "123": "Research mode", 359 "456": "Therapist mode", 360 } 361 362 def test_bridges_telegram_channel_prompts_from_config_yaml(self, tmp_path, monkeypatch): 363 hermes_home = tmp_path / ".hermes" 364 hermes_home.mkdir() 365 config_path = hermes_home / "config.yaml" 366 config_path.write_text( 367 "telegram:\n" 368 " channel_prompts:\n" 369 ' "-1001234567": Research assistant\n' 370 " 789: Creative writing\n", 371 encoding="utf-8", 372 ) 373 374 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 375 376 config = load_gateway_config() 377 378 assert config.platforms[Platform.TELEGRAM].extra["channel_prompts"] == { 379 "-1001234567": "Research assistant", 380 "789": "Creative writing", 381 } 382 383 def test_bridges_slack_channel_prompts_from_config_yaml(self, tmp_path, monkeypatch): 384 hermes_home = tmp_path / ".hermes" 385 hermes_home.mkdir() 386 config_path = hermes_home / "config.yaml" 387 config_path.write_text( 388 "slack:\n" 389 " channel_prompts:\n" 390 ' "C01ABC": Code review mode\n', 391 encoding="utf-8", 392 ) 393 394 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 395 396 config = load_gateway_config() 397 398 assert config.platforms[Platform.SLACK].extra["channel_prompts"] == { 399 "C01ABC": "Code review mode", 400 } 401 402 def test_bridges_feishu_allow_bots_from_config_yaml_to_env(self, tmp_path, monkeypatch): 403 hermes_home = tmp_path / ".hermes" 404 hermes_home.mkdir() 405 config_path = hermes_home / "config.yaml" 406 config_path.write_text( 407 "feishu:\n allow_bots: mentions\n", 408 encoding="utf-8", 409 ) 410 411 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 412 monkeypatch.delenv("FEISHU_ALLOW_BOTS", raising=False) 413 414 load_gateway_config() 415 416 assert os.environ.get("FEISHU_ALLOW_BOTS") == "mentions" 417 418 def test_feishu_allow_bots_env_takes_precedence_over_config_yaml(self, tmp_path, monkeypatch): 419 hermes_home = tmp_path / ".hermes" 420 hermes_home.mkdir() 421 config_path = hermes_home / "config.yaml" 422 config_path.write_text( 423 "feishu:\n allow_bots: all\n", 424 encoding="utf-8", 425 ) 426 427 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 428 monkeypatch.setenv("FEISHU_ALLOW_BOTS", "none") 429 430 load_gateway_config() 431 432 assert os.environ.get("FEISHU_ALLOW_BOTS") == "none" 433 434 def test_invalid_quick_commands_in_config_yaml_are_ignored(self, tmp_path, monkeypatch): 435 hermes_home = tmp_path / ".hermes" 436 hermes_home.mkdir() 437 config_path = hermes_home / "config.yaml" 438 config_path.write_text("quick_commands: not-a-mapping\n", encoding="utf-8") 439 440 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 441 442 config = load_gateway_config() 443 444 assert config.quick_commands == {} 445 446 def test_bridges_unauthorized_dm_behavior_from_config_yaml(self, tmp_path, monkeypatch): 447 hermes_home = tmp_path / ".hermes" 448 hermes_home.mkdir() 449 config_path = hermes_home / "config.yaml" 450 config_path.write_text( 451 "unauthorized_dm_behavior: ignore\n" 452 "whatsapp:\n" 453 " unauthorized_dm_behavior: pair\n", 454 encoding="utf-8", 455 ) 456 457 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 458 459 config = load_gateway_config() 460 461 assert config.unauthorized_dm_behavior == "ignore" 462 assert config.platforms[Platform.WHATSAPP].extra["unauthorized_dm_behavior"] == "pair" 463 464 def test_bridges_telegram_disable_link_previews_from_config_yaml(self, tmp_path, monkeypatch): 465 hermes_home = tmp_path / ".hermes" 466 hermes_home.mkdir() 467 config_path = hermes_home / "config.yaml" 468 config_path.write_text( 469 "telegram:\n" 470 " disable_link_previews: true\n", 471 encoding="utf-8", 472 ) 473 474 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 475 476 config = load_gateway_config() 477 478 assert config.platforms[Platform.TELEGRAM].extra["disable_link_previews"] is True 479 480 def test_bridges_notice_delivery_from_config_yaml(self, tmp_path, monkeypatch): 481 hermes_home = tmp_path / ".hermes" 482 hermes_home.mkdir() 483 config_path = hermes_home / "config.yaml" 484 config_path.write_text( 485 "slack:\n" 486 " notice_delivery: private\n", 487 encoding="utf-8", 488 ) 489 490 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 491 492 config = load_gateway_config() 493 494 assert config.get_notice_delivery(Platform.SLACK) == "private" 495 496 def test_bridges_telegram_proxy_url_from_config_yaml(self, tmp_path, monkeypatch): 497 hermes_home = tmp_path / ".hermes" 498 hermes_home.mkdir() 499 config_path = hermes_home / "config.yaml" 500 config_path.write_text( 501 "telegram:\n" 502 " proxy_url: socks5://127.0.0.1:1080\n", 503 encoding="utf-8", 504 ) 505 506 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 507 monkeypatch.delenv("TELEGRAM_PROXY", raising=False) 508 509 load_gateway_config() 510 511 import os 512 assert os.environ.get("TELEGRAM_PROXY") == "socks5://127.0.0.1:1080" 513 514 def test_telegram_proxy_env_takes_precedence_over_config(self, tmp_path, monkeypatch): 515 hermes_home = tmp_path / ".hermes" 516 hermes_home.mkdir() 517 config_path = hermes_home / "config.yaml" 518 config_path.write_text( 519 "telegram:\n" 520 " proxy_url: http://from-config:8080\n", 521 encoding="utf-8", 522 ) 523 524 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 525 monkeypatch.setenv("TELEGRAM_PROXY", "socks5://from-env:1080") 526 527 load_gateway_config() 528 529 import os 530 assert os.environ.get("TELEGRAM_PROXY") == "socks5://from-env:1080" 531 532 533 class TestHomeChannelEnvOverrides: 534 """Home channel env vars should apply even when the platform was already 535 configured via config.yaml (not just when credential env vars create it).""" 536 537 def test_existing_platform_configs_accept_home_channel_env_overrides(self): 538 cases = [ 539 ( 540 Platform.SLACK, 541 PlatformConfig(enabled=True, token="xoxb-from-config"), 542 {"SLACK_HOME_CHANNEL": "C123", "SLACK_HOME_CHANNEL_NAME": "Ops"}, 543 ("C123", "Ops"), 544 ), 545 ( 546 Platform.WHATSAPP, 547 PlatformConfig(enabled=True), 548 { 549 "WHATSAPP_HOME_CHANNEL": "1234567890@lid", 550 "WHATSAPP_HOME_CHANNEL_NAME": "Owner DM", 551 }, 552 ("1234567890@lid", "Owner DM"), 553 ), 554 ( 555 Platform.SIGNAL, 556 PlatformConfig( 557 enabled=True, 558 extra={"http_url": "http://localhost:9090", "account": "+15551234567"}, 559 ), 560 {"SIGNAL_HOME_CHANNEL": "+1555000", "SIGNAL_HOME_CHANNEL_NAME": "Phone"}, 561 ("+1555000", "Phone"), 562 ), 563 ( 564 Platform.MATTERMOST, 565 PlatformConfig( 566 enabled=True, 567 token="mm-token", 568 extra={"url": "https://mm.example.com"}, 569 ), 570 {"MATTERMOST_HOME_CHANNEL": "ch_abc123", "MATTERMOST_HOME_CHANNEL_NAME": "General"}, 571 ("ch_abc123", "General"), 572 ), 573 ( 574 Platform.MATRIX, 575 PlatformConfig( 576 enabled=True, 577 token="syt_abc123", 578 extra={"homeserver": "https://matrix.example.org"}, 579 ), 580 {"MATRIX_HOME_ROOM": "!room123:example.org", "MATRIX_HOME_ROOM_NAME": "Bot Room"}, 581 ("!room123:example.org", "Bot Room"), 582 ), 583 ( 584 Platform.EMAIL, 585 PlatformConfig( 586 enabled=True, 587 extra={ 588 "address": "hermes@test.com", 589 "imap_host": "imap.test.com", 590 "smtp_host": "smtp.test.com", 591 }, 592 ), 593 {"EMAIL_HOME_ADDRESS": "user@test.com", "EMAIL_HOME_ADDRESS_NAME": "Inbox"}, 594 ("user@test.com", "Inbox"), 595 ), 596 ( 597 Platform.SMS, 598 PlatformConfig(enabled=True, api_key="token_abc"), 599 {"SMS_HOME_CHANNEL": "+15559876543", "SMS_HOME_CHANNEL_NAME": "My Phone"}, 600 ("+15559876543", "My Phone"), 601 ), 602 ] 603 604 for platform, platform_config, env, expected in cases: 605 config = GatewayConfig(platforms={platform: platform_config}) 606 with patch.dict(os.environ, env, clear=True): 607 _apply_env_overrides(config) 608 609 home = config.platforms[platform].home_channel 610 assert home is not None, f"{platform.value}: home_channel should not be None" 611 assert (home.chat_id, home.name) == expected, platform.value