"""Tests for the platform adapter registry and dynamic Platform enum.""" import os import pytest from unittest.mock import MagicMock from gateway.platform_registry import PlatformRegistry, PlatformEntry from gateway.config import Platform, GatewayConfig # ── Platform enum dynamic members ───────────────────────────────────────── class TestPlatformEnumDynamic: """Test that Platform enum accepts unknown values for plugin platforms.""" def test_builtin_members_still_work(self): assert Platform.TELEGRAM.value == "telegram" assert Platform("telegram") is Platform.TELEGRAM def test_dynamic_member_case_normalised(self): """Mixed case normalised to lowercase.""" a = Platform("IRC") b = Platform("irc") assert a is b assert a.value == "irc" def test_dynamic_member_with_hyphens(self): """Registered plugin platforms with hyphens work once registered.""" from gateway.platform_registry import platform_registry as _reg entry = PlatformEntry( name="my-platform", label="My Platform", adapter_factory=lambda cfg: MagicMock(), check_fn=lambda: True, source="plugin", ) _reg.register(entry) try: p = Platform("my-platform") assert p.value == "my-platform" assert p.name == "MY_PLATFORM" finally: _reg.unregister("my-platform") # ── PlatformRegistry ────────────────────────────────────────────────────── class TestPlatformRegistry: """Test the PlatformRegistry itself.""" def _make_entry(self, name="test", check_ok=True, validate_ok=True, factory_ok=True): adapter_mock = MagicMock() return PlatformEntry( name=name, label=name.title(), adapter_factory=lambda cfg, _m=adapter_mock: _m if factory_ok else (_ for _ in ()).throw(RuntimeError("factory error")), check_fn=lambda: check_ok, validate_config=lambda cfg: validate_ok, required_env=[], source="plugin", ), adapter_mock def test_register_and_get(self): reg = PlatformRegistry() entry, _ = self._make_entry("alpha") reg.register(entry) assert reg.get("alpha") is entry assert reg.is_registered("alpha") def test_get_unknown_returns_none(self): reg = PlatformRegistry() assert reg.get("nonexistent") is None def test_unregister(self): reg = PlatformRegistry() entry, _ = self._make_entry("beta") reg.register(entry) assert reg.unregister("beta") is True assert reg.get("beta") is None assert reg.unregister("beta") is False # already gone def test_create_adapter_no_validate(self): """When validate_config is None, skip validation.""" reg = PlatformRegistry() mock_adapter = MagicMock() entry = PlatformEntry( name="novalidate", label="NoValidate", adapter_factory=lambda cfg: mock_adapter, check_fn=lambda: True, validate_config=None, source="plugin", ) reg.register(entry) assert reg.create_adapter("novalidate", MagicMock()) is mock_adapter # ── GatewayConfig integration ──────────────────────────────────────────── class TestGatewayConfigPluginPlatform: """Test that GatewayConfig parses and validates plugin platforms.""" def test_get_connected_platforms_includes_registered_plugin(self): """Plugin platform with registry entry passes get_connected_platforms.""" # Register a fake plugin platform from gateway.platform_registry import platform_registry as _reg test_entry = PlatformEntry( name="testplat", label="TestPlat", adapter_factory=lambda cfg: MagicMock(), check_fn=lambda: True, validate_config=lambda cfg: bool(cfg.extra.get("token")), source="plugin", ) _reg.register(test_entry) try: data = { "platforms": { "testplat": {"enabled": True, "extra": {"token": "abc"}}, } } cfg = GatewayConfig.from_dict(data) connected = cfg.get_connected_platforms() connected_values = {p.value for p in connected} assert "testplat" in connected_values finally: _reg.unregister("testplat") # ── Extended PlatformEntry fields ───────────────────────────────────── class TestPlatformEntryExtendedFields: """Test the auth, message length, and display fields on PlatformEntry.""" def test_default_field_values(self): entry = PlatformEntry( name="test", label="Test", adapter_factory=lambda cfg: None, check_fn=lambda: True, ) assert entry.allowed_users_env == "" assert entry.allow_all_env == "" assert entry.max_message_length == 0 assert entry.pii_safe is False assert entry.emoji == "🔌" assert entry.allow_update_command is True # ── Cron platform resolution ───────────────────────────────────────── class TestCronPlatformResolution: """Test that cron delivery accepts plugin platform names.""" def test_builtin_platform_resolves(self): """Built-in platform names resolve via Platform() call.""" p = Platform("telegram") assert p is Platform.TELEGRAM # ── platforms.py integration ────────────────────────────────────────── class TestPlatformsMerge: """Test get_all_platforms() merges with registry.""" def test_get_all_platforms_includes_plugin(self): from hermes_cli.platforms import get_all_platforms from gateway.platform_registry import platform_registry as _reg _reg.register(PlatformEntry( name="testmerge", label="TestMerge", adapter_factory=lambda cfg: None, check_fn=lambda: True, source="plugin", emoji="🧪", )) try: merged = get_all_platforms() assert "testmerge" in merged assert "TestMerge" in merged["testmerge"].label finally: _reg.unregister("testmerge") # ── apply_yaml_config_fn (PlatformEntry field + load_gateway_config dispatch) ── class TestApplyYamlConfigFnField: """The hook field itself — defaults, custom values, signature.""" def test_default_is_none(self): entry = PlatformEntry( name="test", label="Test", adapter_factory=lambda cfg: None, check_fn=lambda: True, ) assert entry.apply_yaml_config_fn is None class TestApplyYamlConfigFnDispatch: """End-to-end dispatch through load_gateway_config(). Each test registers a temporary PlatformEntry, writes a config.yaml in a tmp HERMES_HOME, calls load_gateway_config(), and asserts the hook was invoked correctly. Cleanup unregisters the entry. """ def _write_config(self, tmp_path, content: str): hermes_home = tmp_path / ".hermes" hermes_home.mkdir() (hermes_home / "config.yaml").write_text(content, encoding="utf-8") return hermes_home def _register_hook(self, name, hook_fn): from gateway.platform_registry import platform_registry as _reg entry = PlatformEntry( name=name, label=name.title(), adapter_factory=lambda cfg: None, check_fn=lambda: True, source="plugin", apply_yaml_config_fn=hook_fn, ) _reg.register(entry) return _reg def test_hook_exception_swallowed(self, tmp_path, monkeypatch): """A misbehaving hook never aborts load_gateway_config().""" def _bad_hook(yaml_cfg, platform_cfg): raise RuntimeError("plugin author bug") # Also register a well-behaved hook to ensure dispatch continues # iterating after a bad one. good_called = {"count": 0} def _good_hook(yaml_cfg, platform_cfg): good_called["count"] += 1 return None from gateway.platform_registry import platform_registry as _reg _reg.register(PlatformEntry( name="mybadplat", label="MyBad", adapter_factory=lambda cfg: None, check_fn=lambda: True, source="plugin", apply_yaml_config_fn=_bad_hook, )) _reg.register(PlatformEntry( name="mygoodplat", label="MyGood", adapter_factory=lambda cfg: None, check_fn=lambda: True, source="plugin", apply_yaml_config_fn=_good_hook, )) try: home = self._write_config( tmp_path, "mybadplat:\n k: v\n" "mygoodplat:\n k: v\n", ) monkeypatch.setenv("HERMES_HOME", str(home)) # Must not raise. from gateway.config import load_gateway_config load_gateway_config() assert good_called["count"] == 1 finally: _reg.unregister("mybadplat") _reg.unregister("mygoodplat") def test_env_var_takes_precedence_when_hook_uses_getenv_guard( self, tmp_path, monkeypatch ): """The standard `not os.getenv(...)` guard preserves env > YAML.""" env_var = "MYPRECPLAT_FLAG" monkeypatch.setenv(env_var, "preexisting") def _hook(yaml_cfg, platform_cfg): if "flag" in platform_cfg and not os.getenv(env_var): os.environ[env_var] = str(platform_cfg["flag"]).lower() return None reg = self._register_hook("myprecplat", _hook) try: home = self._write_config( tmp_path, "myprecplat:\n flag: yaml-value\n", ) monkeypatch.setenv("HERMES_HOME", str(home)) from gateway.config import load_gateway_config load_gateway_config() # Pre-existing env var was NOT clobbered by the hook. assert os.environ.get(env_var) == "preexisting" finally: reg.unregister("myprecplat") os.environ.pop(env_var, None) class TestPluginPlatformSharedKeyBridge: """Plugin-registered platforms get the same shared-key bridging as built-ins. Without this, plugin authors using ``apply_yaml_config_fn`` would have to re-implement bridging for every common key (``unauthorized_dm_behavior``, ``notice_delivery``, ``reply_prefix``, ``require_mention``, ``dm_policy``, ``allow_from``, etc.) — defeating the hook's whole point of letting plugins focus on their *platform-specific* keys. """ def _write_config(self, tmp_path, content: str): hermes_home = tmp_path / ".hermes" hermes_home.mkdir() (hermes_home / "config.yaml").write_text(content, encoding="utf-8") return hermes_home def test_shared_keys_bridged_for_plugin_platform(self, tmp_path, monkeypatch): """A plugin platform's ``require_mention``/``dm_policy``/etc. flow into ``PlatformConfig.extra`` without the plugin needing its own bridge.""" from gateway.platform_registry import platform_registry as _reg _reg.register(PlatformEntry( name="mysharedplat", label="MySharedPlat", adapter_factory=lambda cfg: None, check_fn=lambda: True, source="plugin", )) try: home = self._write_config( tmp_path, "mysharedplat:\n" " require_mention: true\n" " dm_policy: allow\n" " reply_prefix: \"→ \"\n" " allow_from: [\"alice\", \"bob\"]\n", ) monkeypatch.setenv("HERMES_HOME", str(home)) from gateway.config import load_gateway_config, Platform cfg = load_gateway_config() plat = Platform("mysharedplat") assert plat in cfg.platforms extra = cfg.platforms[plat].extra assert extra.get("require_mention") is True assert extra.get("dm_policy") == "allow" assert extra.get("reply_prefix") == "→ " assert extra.get("allow_from") == ["alice", "bob"] finally: _reg.unregister("mysharedplat") class TestPluginEnablementGate: """Plugin platforms must NOT auto-enable on check_fn alone (#31116). When a plugin registers ``is_connected`` (the "did the user actually configure credentials" probe), ``load_gateway_config`` must consult it before flipping ``enabled = True``. Without this gate, ``check_fn`` semantics ("the SDK is importable") get conflated with "the user wants this platform on", and the gateway tries to connect to e.g. Discord with no token — emitting noisy retry-forever errors on every fresh install that has the plugin loaded. """ def _write_config(self, tmp_path, content: str = ""): hermes_home = tmp_path / ".hermes" hermes_home.mkdir() (hermes_home / "config.yaml").write_text(content, encoding="utf-8") return hermes_home def test_plugin_with_is_connected_false_is_NOT_enabled( self, tmp_path, monkeypatch ): """check_fn=True + is_connected=False must NOT enable the platform. Reproduces #31116: Discord plugin loads, its check_fn lazy-installs discord.py and returns True, but the user has no DISCORD_BOT_TOKEN. Previously this auto-enabled Discord and the gateway spammed ``ERROR ... [Discord] No bot token configured`` on every reconnect. """ from gateway.platform_registry import platform_registry as _reg _reg.register(PlatformEntry( name="myunconfiguredplat", label="MyUnconfigured", adapter_factory=lambda cfg: None, check_fn=lambda: True, # SDK available is_connected=lambda cfg: False, # but user hasn't set credentials source="plugin", )) try: home = self._write_config(tmp_path) monkeypatch.setenv("HERMES_HOME", str(home)) from gateway.config import load_gateway_config, Platform cfg = load_gateway_config() plat = Platform("myunconfiguredplat") # Either absent entirely, or present but explicitly disabled. if plat in cfg.platforms: assert cfg.platforms[plat].enabled is False, ( "Plugin with is_connected=False must NOT be auto-enabled" ) finally: _reg.unregister("myunconfiguredplat") def test_is_connected_raises_does_not_enable(self, tmp_path, monkeypatch): """A buggy is_connected must not silently enable the platform. Treat a raising is_connected as "configuration unknown" — refuse to enable, log, and move on. Anything else would re-introduce the #31116 bug for plugins whose probe has a transient failure. """ from gateway.platform_registry import platform_registry as _reg def _bad_probe(cfg): raise RuntimeError("plugin bug") _reg.register(PlatformEntry( name="mybadprobeplat", label="MyBadProbe", adapter_factory=lambda cfg: None, check_fn=lambda: True, is_connected=_bad_probe, source="plugin", )) try: home = self._write_config(tmp_path) monkeypatch.setenv("HERMES_HOME", str(home)) from gateway.config import load_gateway_config, Platform cfg = load_gateway_config() plat = Platform("mybadprobeplat") if plat in cfg.platforms: assert cfg.platforms[plat].enabled is False finally: _reg.unregister("mybadprobeplat") def test_is_connected_failed_gate_does_not_leak_extras( self, tmp_path, monkeypatch ): """When the gate rejects, env-seeded extras must NOT leak onto ``config.platforms``. A rejected plugin should be invisible, not present-but-partially-populated. """ from gateway.platform_registry import platform_registry as _reg _reg.register(PlatformEntry( name="myrejectedplat", label="MyRejected", adapter_factory=lambda cfg: None, check_fn=lambda: True, is_connected=lambda cfg: False, env_enablement_fn=lambda: {"some_key": "should-not-leak"}, source="plugin", )) try: home = self._write_config(tmp_path) monkeypatch.setenv("HERMES_HOME", str(home)) from gateway.config import load_gateway_config, Platform cfg = load_gateway_config() plat = Platform("myrejectedplat") if plat in cfg.platforms: assert cfg.platforms[plat].enabled is False assert "some_key" not in cfg.platforms[plat].extra, ( "Rejected plugin's env-seeded extras leaked onto " "config.platforms" ) finally: _reg.unregister("myrejectedplat")