test_platform_registry.py
1 """Tests for the platform adapter registry and dynamic Platform enum.""" 2 3 import os 4 import pytest 5 from unittest.mock import MagicMock, patch 6 from dataclasses import dataclass 7 8 from gateway.platform_registry import PlatformRegistry, PlatformEntry, platform_registry 9 from gateway.config import Platform, PlatformConfig, GatewayConfig 10 11 12 # ── Platform enum dynamic members ───────────────────────────────────────── 13 14 15 class TestPlatformEnumDynamic: 16 """Test that Platform enum accepts unknown values for plugin platforms.""" 17 18 def test_builtin_members_still_work(self): 19 assert Platform.TELEGRAM.value == "telegram" 20 assert Platform("telegram") is Platform.TELEGRAM 21 22 def test_dynamic_member_created(self): 23 p = Platform("irc") 24 assert p.value == "irc" 25 assert p.name == "IRC" 26 27 def test_dynamic_member_identity_stable(self): 28 """Same value returns same object (cached).""" 29 a = Platform("irc") 30 b = Platform("irc") 31 assert a is b 32 33 def test_dynamic_member_case_normalised(self): 34 """Mixed case normalised to lowercase.""" 35 a = Platform("IRC") 36 b = Platform("irc") 37 assert a is b 38 assert a.value == "irc" 39 40 def test_dynamic_member_with_hyphens(self): 41 """Registered plugin platforms with hyphens work once registered.""" 42 from gateway.platform_registry import platform_registry as _reg 43 44 entry = PlatformEntry( 45 name="my-platform", 46 label="My Platform", 47 adapter_factory=lambda cfg: MagicMock(), 48 check_fn=lambda: True, 49 source="plugin", 50 ) 51 _reg.register(entry) 52 try: 53 p = Platform("my-platform") 54 assert p.value == "my-platform" 55 assert p.name == "MY_PLATFORM" 56 finally: 57 _reg.unregister("my-platform") 58 59 def test_dynamic_member_rejects_unregistered(self): 60 """Arbitrary strings are rejected to prevent enum pollution.""" 61 with pytest.raises(ValueError): 62 Platform("totally-fake-platform") 63 64 def test_dynamic_member_rejects_non_string(self): 65 with pytest.raises(ValueError): 66 Platform(123) 67 68 def test_dynamic_member_rejects_empty(self): 69 with pytest.raises(ValueError): 70 Platform("") 71 72 def test_dynamic_member_rejects_whitespace_only(self): 73 with pytest.raises(ValueError): 74 Platform(" ") 75 76 77 # ── PlatformRegistry ────────────────────────────────────────────────────── 78 79 80 class TestPlatformRegistry: 81 """Test the PlatformRegistry itself.""" 82 83 def _make_entry(self, name="test", check_ok=True, validate_ok=True, factory_ok=True): 84 adapter_mock = MagicMock() 85 return PlatformEntry( 86 name=name, 87 label=name.title(), 88 adapter_factory=lambda cfg, _m=adapter_mock: _m if factory_ok else (_ for _ in ()).throw(RuntimeError("factory error")), 89 check_fn=lambda: check_ok, 90 validate_config=lambda cfg: validate_ok, 91 required_env=[], 92 source="plugin", 93 ), adapter_mock 94 95 def test_register_and_get(self): 96 reg = PlatformRegistry() 97 entry, _ = self._make_entry("alpha") 98 reg.register(entry) 99 assert reg.get("alpha") is entry 100 assert reg.is_registered("alpha") 101 102 def test_get_unknown_returns_none(self): 103 reg = PlatformRegistry() 104 assert reg.get("nonexistent") is None 105 106 def test_unregister(self): 107 reg = PlatformRegistry() 108 entry, _ = self._make_entry("beta") 109 reg.register(entry) 110 assert reg.unregister("beta") is True 111 assert reg.get("beta") is None 112 assert reg.unregister("beta") is False # already gone 113 114 def test_create_adapter_success(self): 115 reg = PlatformRegistry() 116 entry, mock_adapter = self._make_entry("gamma") 117 reg.register(entry) 118 result = reg.create_adapter("gamma", MagicMock()) 119 assert result is mock_adapter 120 121 def test_create_adapter_unknown_name(self): 122 reg = PlatformRegistry() 123 assert reg.create_adapter("unknown", MagicMock()) is None 124 125 def test_create_adapter_check_fails(self): 126 reg = PlatformRegistry() 127 entry, _ = self._make_entry("delta", check_ok=False) 128 reg.register(entry) 129 assert reg.create_adapter("delta", MagicMock()) is None 130 131 def test_create_adapter_validate_fails(self): 132 reg = PlatformRegistry() 133 entry, _ = self._make_entry("epsilon", validate_ok=False) 134 reg.register(entry) 135 assert reg.create_adapter("epsilon", MagicMock()) is None 136 137 def test_create_adapter_factory_exception(self): 138 reg = PlatformRegistry() 139 entry = PlatformEntry( 140 name="broken", 141 label="Broken", 142 adapter_factory=lambda cfg: (_ for _ in ()).throw(RuntimeError("boom")), 143 check_fn=lambda: True, 144 validate_config=None, 145 source="plugin", 146 ) 147 reg.register(entry) 148 # factory raises → create_adapter returns None instead of propagating 149 assert reg.create_adapter("broken", MagicMock()) is None 150 151 def test_create_adapter_no_validate(self): 152 """When validate_config is None, skip validation.""" 153 reg = PlatformRegistry() 154 mock_adapter = MagicMock() 155 entry = PlatformEntry( 156 name="novalidate", 157 label="NoValidate", 158 adapter_factory=lambda cfg: mock_adapter, 159 check_fn=lambda: True, 160 validate_config=None, 161 source="plugin", 162 ) 163 reg.register(entry) 164 assert reg.create_adapter("novalidate", MagicMock()) is mock_adapter 165 166 def test_all_entries(self): 167 reg = PlatformRegistry() 168 e1, _ = self._make_entry("one") 169 e2, _ = self._make_entry("two") 170 reg.register(e1) 171 reg.register(e2) 172 names = {e.name for e in reg.all_entries()} 173 assert names == {"one", "two"} 174 175 def test_plugin_entries(self): 176 reg = PlatformRegistry() 177 plugin_entry, _ = self._make_entry("plugged") 178 builtin_entry = PlatformEntry( 179 name="core", 180 label="Core", 181 adapter_factory=lambda cfg: MagicMock(), 182 check_fn=lambda: True, 183 source="builtin", 184 ) 185 reg.register(plugin_entry) 186 reg.register(builtin_entry) 187 plugin_names = {e.name for e in reg.plugin_entries()} 188 assert plugin_names == {"plugged"} 189 190 def test_re_register_replaces(self): 191 reg = PlatformRegistry() 192 entry1, mock1 = self._make_entry("dup") 193 entry2 = PlatformEntry( 194 name="dup", 195 label="Dup v2", 196 adapter_factory=lambda cfg: "v2", 197 check_fn=lambda: True, 198 source="plugin", 199 ) 200 reg.register(entry1) 201 reg.register(entry2) 202 assert reg.get("dup").label == "Dup v2" 203 204 205 # ── GatewayConfig integration ──────────────────────────────────────────── 206 207 208 class TestGatewayConfigPluginPlatform: 209 """Test that GatewayConfig parses and validates plugin platforms.""" 210 211 def test_from_dict_accepts_plugin_platform(self): 212 data = { 213 "platforms": { 214 "telegram": {"enabled": True, "token": "test-token"}, 215 "irc": {"enabled": True, "extra": {"server": "irc.libera.chat"}}, 216 } 217 } 218 cfg = GatewayConfig.from_dict(data) 219 platform_values = {p.value for p in cfg.platforms} 220 assert "telegram" in platform_values 221 assert "irc" in platform_values 222 223 def test_get_connected_platforms_includes_registered_plugin(self): 224 """Plugin platform with registry entry passes get_connected_platforms.""" 225 # Register a fake plugin platform 226 from gateway.platform_registry import platform_registry as _reg 227 228 test_entry = PlatformEntry( 229 name="testplat", 230 label="TestPlat", 231 adapter_factory=lambda cfg: MagicMock(), 232 check_fn=lambda: True, 233 validate_config=lambda cfg: bool(cfg.extra.get("token")), 234 source="plugin", 235 ) 236 _reg.register(test_entry) 237 try: 238 data = { 239 "platforms": { 240 "testplat": {"enabled": True, "extra": {"token": "abc"}}, 241 } 242 } 243 cfg = GatewayConfig.from_dict(data) 244 connected = cfg.get_connected_platforms() 245 connected_values = {p.value for p in connected} 246 assert "testplat" in connected_values 247 finally: 248 _reg.unregister("testplat") 249 250 def test_get_connected_platforms_excludes_unregistered_plugin(self): 251 """Plugin platform without registry entry is excluded.""" 252 data = { 253 "platforms": { 254 "unknown_plugin": {"enabled": True, "extra": {"token": "abc"}}, 255 } 256 } 257 cfg = GatewayConfig.from_dict(data) 258 connected = cfg.get_connected_platforms() 259 connected_values = {p.value for p in connected} 260 assert "unknown_plugin" not in connected_values 261 262 def test_get_connected_platforms_excludes_invalid_config(self): 263 """Plugin platform with failing validate_config is excluded.""" 264 from gateway.platform_registry import platform_registry as _reg 265 266 test_entry = PlatformEntry( 267 name="badconfig", 268 label="BadConfig", 269 adapter_factory=lambda cfg: MagicMock(), 270 check_fn=lambda: True, 271 validate_config=lambda cfg: False, # always fails 272 source="plugin", 273 ) 274 _reg.register(test_entry) 275 try: 276 data = { 277 "platforms": { 278 "badconfig": {"enabled": True, "extra": {}}, 279 } 280 } 281 cfg = GatewayConfig.from_dict(data) 282 connected = cfg.get_connected_platforms() 283 connected_values = {p.value for p in connected} 284 assert "badconfig" not in connected_values 285 finally: 286 _reg.unregister("badconfig") 287 288 289 # ── Extended PlatformEntry fields ───────────────────────────────────── 290 291 292 class TestPlatformEntryExtendedFields: 293 """Test the auth, message length, and display fields on PlatformEntry.""" 294 295 def test_default_field_values(self): 296 entry = PlatformEntry( 297 name="test", 298 label="Test", 299 adapter_factory=lambda cfg: None, 300 check_fn=lambda: True, 301 ) 302 assert entry.allowed_users_env == "" 303 assert entry.allow_all_env == "" 304 assert entry.max_message_length == 0 305 assert entry.pii_safe is False 306 assert entry.emoji == "🔌" 307 assert entry.allow_update_command is True 308 309 def test_custom_auth_fields(self): 310 entry = PlatformEntry( 311 name="irc", 312 label="IRC", 313 adapter_factory=lambda cfg: None, 314 check_fn=lambda: True, 315 allowed_users_env="IRC_ALLOWED_USERS", 316 allow_all_env="IRC_ALLOW_ALL_USERS", 317 max_message_length=450, 318 pii_safe=False, 319 emoji="💬", 320 ) 321 assert entry.allowed_users_env == "IRC_ALLOWED_USERS" 322 assert entry.allow_all_env == "IRC_ALLOW_ALL_USERS" 323 assert entry.max_message_length == 450 324 assert entry.emoji == "💬" 325 326 327 # ── Cron platform resolution ───────────────────────────────────────── 328 329 330 class TestCronPlatformResolution: 331 """Test that cron delivery accepts plugin platform names.""" 332 333 def test_builtin_platform_resolves(self): 334 """Built-in platform names resolve via Platform() call.""" 335 p = Platform("telegram") 336 assert p is Platform.TELEGRAM 337 338 def test_plugin_platform_resolves(self): 339 """Plugin platform names create dynamic enum members.""" 340 p = Platform("irc") 341 assert p.value == "irc" 342 343 def test_invalid_platform_type_rejected(self): 344 """Non-string values are still rejected.""" 345 with pytest.raises(ValueError): 346 Platform(None) 347 348 349 # ── platforms.py integration ────────────────────────────────────────── 350 351 352 class TestPlatformsMerge: 353 """Test get_all_platforms() merges with registry.""" 354 355 def test_get_all_platforms_includes_builtins(self): 356 from hermes_cli.platforms import get_all_platforms, PLATFORMS 357 merged = get_all_platforms() 358 for key in PLATFORMS: 359 assert key in merged 360 361 def test_get_all_platforms_includes_plugin(self): 362 from hermes_cli.platforms import get_all_platforms 363 from gateway.platform_registry import platform_registry as _reg 364 365 _reg.register(PlatformEntry( 366 name="testmerge", 367 label="TestMerge", 368 adapter_factory=lambda cfg: None, 369 check_fn=lambda: True, 370 source="plugin", 371 emoji="🧪", 372 )) 373 try: 374 merged = get_all_platforms() 375 assert "testmerge" in merged 376 assert "TestMerge" in merged["testmerge"].label 377 finally: 378 _reg.unregister("testmerge") 379 380 def test_platform_label_plugin_fallback(self): 381 from hermes_cli.platforms import platform_label 382 from gateway.platform_registry import platform_registry as _reg 383 384 _reg.register(PlatformEntry( 385 name="labeltest", 386 label="LabelTest", 387 adapter_factory=lambda cfg: None, 388 check_fn=lambda: True, 389 source="plugin", 390 emoji="🏷️", 391 )) 392 try: 393 label = platform_label("labeltest") 394 assert "LabelTest" in label 395 finally: 396 _reg.unregister("labeltest")