test_irc_adapter.py
1 """Tests for the IRC platform adapter plugin.""" 2 3 import asyncio 4 import os 5 import sys 6 import pytest 7 from pathlib import Path 8 from unittest.mock import AsyncMock, MagicMock, patch 9 10 from tests.gateway._plugin_adapter_loader import load_plugin_adapter 11 12 # Load plugins/platforms/irc/adapter.py under a unique module name 13 # (plugin_adapter_irc) so it cannot collide with other plugin adapters 14 # loaded by sibling tests in the same xdist worker. 15 _irc_mod = load_plugin_adapter("irc") 16 17 _parse_irc_message = _irc_mod._parse_irc_message 18 _extract_nick = _irc_mod._extract_nick 19 IRCAdapter = _irc_mod.IRCAdapter 20 check_requirements = _irc_mod.check_requirements 21 validate_config = _irc_mod.validate_config 22 register = _irc_mod.register 23 24 25 class TestIRCProtocolHelpers: 26 27 def test_parse_simple_command(self): 28 msg = _parse_irc_message("PING :server.example.com") 29 assert msg["command"] == "PING" 30 assert msg["params"] == ["server.example.com"] 31 assert msg["prefix"] == "" 32 33 def test_parse_prefixed_message(self): 34 msg = _parse_irc_message(":nick!user@host PRIVMSG #channel :Hello world") 35 assert msg["prefix"] == "nick!user@host" 36 assert msg["command"] == "PRIVMSG" 37 assert msg["params"] == ["#channel", "Hello world"] 38 39 def test_parse_numeric_reply(self): 40 msg = _parse_irc_message(":server 001 hermes-bot :Welcome to IRC") 41 assert msg["prefix"] == "server" 42 assert msg["command"] == "001" 43 assert msg["params"] == ["hermes-bot", "Welcome to IRC"] 44 45 def test_parse_nick_collision(self): 46 msg = _parse_irc_message(":server 433 * hermes-bot :Nickname is already in use") 47 assert msg["command"] == "433" 48 49 def test_extract_nick_full_prefix(self): 50 assert _extract_nick("nick!user@host") == "nick" 51 52 def test_extract_nick_bare(self): 53 assert _extract_nick("server.example.com") == "server.example.com" 54 55 56 # ── IRC Adapter ────────────────────────────────────────────────────────── 57 58 59 class TestIRCAdapterInit: 60 61 def test_init_from_env(self, monkeypatch): 62 monkeypatch.setenv("IRC_SERVER", "irc.test.net") 63 monkeypatch.setenv("IRC_PORT", "6667") 64 monkeypatch.setenv("IRC_NICKNAME", "testbot") 65 monkeypatch.setenv("IRC_CHANNEL", "#test") 66 monkeypatch.setenv("IRC_USE_TLS", "false") 67 68 from gateway.config import PlatformConfig 69 cfg = PlatformConfig(enabled=True) 70 adapter = IRCAdapter(cfg) 71 72 assert adapter.server == "irc.test.net" 73 assert adapter.port == 6667 74 assert adapter.nickname == "testbot" 75 assert adapter.channel == "#test" 76 assert adapter.use_tls is False 77 78 def test_init_from_config_extra(self, monkeypatch): 79 # Clear any env vars 80 for key in ("IRC_SERVER", "IRC_PORT", "IRC_NICKNAME", "IRC_CHANNEL", "IRC_USE_TLS"): 81 monkeypatch.delenv(key, raising=False) 82 83 from gateway.config import PlatformConfig 84 cfg = PlatformConfig( 85 enabled=True, 86 extra={ 87 "server": "irc.libera.chat", 88 "port": 6697, 89 "nickname": "hermes", 90 "channel": "#hermes-dev", 91 "use_tls": True, 92 }, 93 ) 94 adapter = IRCAdapter(cfg) 95 96 assert adapter.server == "irc.libera.chat" 97 assert adapter.port == 6697 98 assert adapter.nickname == "hermes" 99 assert adapter.channel == "#hermes-dev" 100 assert adapter.use_tls is True 101 102 def test_env_overrides_config(self, monkeypatch): 103 monkeypatch.setenv("IRC_SERVER", "env-server.net") 104 105 from gateway.config import PlatformConfig 106 cfg = PlatformConfig( 107 enabled=True, 108 extra={"server": "config-server.net", "channel": "#ch"}, 109 ) 110 adapter = IRCAdapter(cfg) 111 assert adapter.server == "env-server.net" 112 113 114 class TestIRCAdapterSend: 115 116 @pytest.fixture 117 def adapter(self, monkeypatch): 118 for key in ("IRC_SERVER", "IRC_PORT", "IRC_NICKNAME", "IRC_CHANNEL", "IRC_USE_TLS"): 119 monkeypatch.delenv(key, raising=False) 120 from gateway.config import PlatformConfig 121 cfg = PlatformConfig( 122 enabled=True, 123 extra={ 124 "server": "localhost", 125 "port": 6667, 126 "nickname": "testbot", 127 "channel": "#test", 128 "use_tls": False, 129 }, 130 ) 131 return IRCAdapter(cfg) 132 133 @pytest.mark.asyncio 134 async def test_send_not_connected(self, adapter): 135 result = await adapter.send("#test", "hello") 136 assert result.success is False 137 assert "Not connected" in result.error 138 139 @pytest.mark.asyncio 140 async def test_send_success(self, adapter): 141 writer = MagicMock() 142 writer.is_closing = MagicMock(return_value=False) 143 writer.write = MagicMock() 144 writer.drain = AsyncMock() 145 adapter._writer = writer 146 147 result = await adapter.send("#test", "hello world") 148 assert result.success is True 149 assert result.message_id is not None 150 # Verify PRIVMSG was sent 151 writer.write.assert_called() 152 sent_data = writer.write.call_args[0][0] 153 assert b"PRIVMSG #test :hello world" in sent_data 154 155 @pytest.mark.asyncio 156 async def test_send_splits_long_messages(self, adapter): 157 writer = MagicMock() 158 writer.is_closing = MagicMock(return_value=False) 159 writer.write = MagicMock() 160 writer.drain = AsyncMock() 161 adapter._writer = writer 162 163 long_msg = "x" * 1000 164 result = await adapter.send("#test", long_msg) 165 assert result.success is True 166 # Should have been split into multiple PRIVMSG calls 167 assert writer.write.call_count > 1 168 169 170 class TestIRCAdapterMessageParsing: 171 172 @pytest.fixture 173 def adapter(self, monkeypatch): 174 for key in ("IRC_SERVER", "IRC_PORT", "IRC_NICKNAME", "IRC_CHANNEL", "IRC_USE_TLS"): 175 monkeypatch.delenv(key, raising=False) 176 from gateway.config import PlatformConfig 177 cfg = PlatformConfig( 178 enabled=True, 179 extra={ 180 "server": "localhost", 181 "port": 6667, 182 "nickname": "hermes", 183 "channel": "#test", 184 "use_tls": False, 185 }, 186 ) 187 a = IRCAdapter(cfg) 188 a._current_nick = "hermes" 189 a._registered = True 190 return a 191 192 @pytest.mark.asyncio 193 async def test_handle_ping(self, adapter): 194 writer = MagicMock() 195 writer.is_closing = MagicMock(return_value=False) 196 writer.write = MagicMock() 197 writer.drain = AsyncMock() 198 adapter._writer = writer 199 200 await adapter._handle_line("PING :test-server") 201 sent = writer.write.call_args[0][0] 202 assert b"PONG :test-server" in sent 203 204 @pytest.mark.asyncio 205 async def test_handle_welcome(self, adapter): 206 adapter._registered = False 207 adapter._registration_event = asyncio.Event() 208 209 await adapter._handle_line(":server 001 hermes :Welcome to IRC") 210 assert adapter._registered is True 211 assert adapter._registration_event.is_set() 212 213 @pytest.mark.asyncio 214 async def test_handle_nick_collision(self, adapter): 215 writer = MagicMock() 216 writer.is_closing = MagicMock(return_value=False) 217 writer.write = MagicMock() 218 writer.drain = AsyncMock() 219 adapter._writer = writer 220 221 await adapter._handle_line(":server 433 * hermes :Nickname in use") 222 assert adapter._current_nick == "hermes_" 223 sent = writer.write.call_args[0][0] 224 assert b"NICK hermes_" in sent 225 226 @pytest.mark.asyncio 227 async def test_handle_addressed_channel_message(self, adapter): 228 """Messages addressed to the bot (nick: msg) should be dispatched.""" 229 handler = AsyncMock(return_value="response") 230 adapter._message_handler = handler 231 232 # Mock handle_message to capture the event 233 dispatched = [] 234 original_dispatch = adapter._dispatch_message 235 236 async def capture_dispatch(**kwargs): 237 dispatched.append(kwargs) 238 239 adapter._dispatch_message = capture_dispatch 240 241 await adapter._handle_line(":user!u@host PRIVMSG #test :hermes: hello there") 242 assert len(dispatched) == 1 243 assert dispatched[0]["text"] == "hello there" 244 assert dispatched[0]["chat_id"] == "#test" 245 246 @pytest.mark.asyncio 247 async def test_ignores_unaddressed_channel_message(self, adapter): 248 dispatched = [] 249 250 async def capture_dispatch(**kwargs): 251 dispatched.append(kwargs) 252 253 adapter._dispatch_message = capture_dispatch 254 adapter._message_handler = AsyncMock() 255 256 await adapter._handle_line(":user!u@host PRIVMSG #test :just talking") 257 assert len(dispatched) == 0 258 259 @pytest.mark.asyncio 260 async def test_handle_dm(self, adapter): 261 """DMs (target == bot nick) should always be dispatched.""" 262 dispatched = [] 263 264 async def capture_dispatch(**kwargs): 265 dispatched.append(kwargs) 266 267 adapter._dispatch_message = capture_dispatch 268 adapter._message_handler = AsyncMock() 269 270 await adapter._handle_line(":user!u@host PRIVMSG hermes :private message") 271 assert len(dispatched) == 1 272 assert dispatched[0]["text"] == "private message" 273 assert dispatched[0]["chat_type"] == "dm" 274 assert dispatched[0]["chat_id"] == "user" 275 276 @pytest.mark.asyncio 277 async def test_ignores_own_messages(self, adapter): 278 dispatched = [] 279 280 async def capture_dispatch(**kwargs): 281 dispatched.append(kwargs) 282 283 adapter._dispatch_message = capture_dispatch 284 adapter._message_handler = AsyncMock() 285 286 await adapter._handle_line(":hermes!bot@host PRIVMSG #test :my own msg") 287 assert len(dispatched) == 0 288 289 @pytest.mark.asyncio 290 async def test_ctcp_action_converted(self, adapter): 291 """CTCP ACTION (/me) should be converted to text.""" 292 dispatched = [] 293 294 async def capture_dispatch(**kwargs): 295 dispatched.append(kwargs) 296 297 adapter._dispatch_message = capture_dispatch 298 adapter._message_handler = AsyncMock() 299 300 await adapter._handle_line(":user!u@host PRIVMSG hermes :\x01ACTION waves\x01") 301 assert len(dispatched) == 1 302 assert dispatched[0]["text"] == "* user waves" 303 304 @pytest.mark.asyncio 305 async def test_allowed_users_case_insensitive(self, monkeypatch): 306 """Allowlist should match nicks case-insensitively.""" 307 for key in ("IRC_SERVER", "IRC_PORT", "IRC_NICKNAME", "IRC_CHANNEL", "IRC_USE_TLS"): 308 monkeypatch.delenv(key, raising=False) 309 from gateway.config import PlatformConfig 310 cfg = PlatformConfig( 311 enabled=True, 312 extra={ 313 "server": "localhost", 314 "port": 6667, 315 "nickname": "hermes", 316 "channel": "#test", 317 "use_tls": False, 318 "allowed_users": ["Admin", "BOB"], 319 }, 320 ) 321 adapter = IRCAdapter(cfg) 322 adapter._current_nick = "hermes" 323 adapter._registered = True 324 dispatched = [] 325 326 async def capture_dispatch(**kwargs): 327 dispatched.append(kwargs) 328 329 adapter._dispatch_message = capture_dispatch 330 adapter._message_handler = AsyncMock() 331 332 # "admin" matches "Admin" in allowlist 333 await adapter._handle_line(":admin!u@host PRIVMSG #test :hermes: hello") 334 assert len(dispatched) == 1 335 assert dispatched[0]["text"] == "hello" 336 337 @pytest.mark.asyncio 338 async def test_unauthorized_user_blocked(self, monkeypatch): 339 """Nicks not in allowlist should be ignored.""" 340 for key in ("IRC_SERVER", "IRC_PORT", "IRC_NICKNAME", "IRC_CHANNEL", "IRC_USE_TLS"): 341 monkeypatch.delenv(key, raising=False) 342 from gateway.config import PlatformConfig 343 cfg = PlatformConfig( 344 enabled=True, 345 extra={ 346 "server": "localhost", 347 "port": 6667, 348 "nickname": "hermes", 349 "channel": "#test", 350 "use_tls": False, 351 "allowed_users": ["Admin", "BOB"], 352 }, 353 ) 354 adapter = IRCAdapter(cfg) 355 adapter._current_nick = "hermes" 356 adapter._registered = True 357 dispatched = [] 358 359 async def capture_dispatch(**kwargs): 360 dispatched.append(kwargs) 361 362 adapter._dispatch_message = capture_dispatch 363 adapter._message_handler = AsyncMock() 364 365 await adapter._handle_line(":eve!u@host PRIVMSG #test :hermes: hello") 366 assert len(dispatched) == 0 367 368 @pytest.mark.asyncio 369 async def test_nick_collision_retry(self, adapter): 370 """Multiple 433 responses should keep incrementing the suffix.""" 371 writer = MagicMock() 372 writer.is_closing = MagicMock(return_value=False) 373 writer.write = MagicMock() 374 writer.drain = AsyncMock() 375 adapter._writer = writer 376 377 await adapter._handle_line(":server 433 * hermes :Nickname in use") 378 assert adapter._current_nick == "hermes_" 379 await adapter._handle_line(":server 433 * hermes_ :Nickname in use") 380 assert adapter._current_nick == "hermes_1" 381 await adapter._handle_line(":server 433 * hermes_1 :Nickname in use") 382 assert adapter._current_nick == "hermes_2" 383 384 385 class TestIRCAdapterSplitting: 386 387 def test_split_respects_byte_limit(self): 388 """Multi-byte characters should not exceed IRC byte limit.""" 389 # 100 japanese chars = 300 bytes in utf-8 390 text = "あ" * 100 391 from gateway.config import PlatformConfig 392 cfg = PlatformConfig(enabled=True, extra={"server": "x", "channel": "#x"}) 393 adapter = IRCAdapter(cfg) 394 adapter._current_nick = "bot" 395 lines = adapter._split_message(text, "#test") 396 for line in lines: 397 overhead = len(f"PRIVMSG #test :{line}\r\n".encode("utf-8")) 398 assert overhead <= 512, f"line over 512 bytes: {overhead}" 399 400 def test_split_prefers_word_boundary(self): 401 text = "hello world foo bar baz qux" 402 from gateway.config import PlatformConfig 403 cfg = PlatformConfig(enabled=True, extra={"server": "x", "channel": "#x"}) 404 adapter = IRCAdapter(cfg) 405 adapter._current_nick = "bot" 406 lines = adapter._split_message(text, "#test") 407 # Should not split in the middle of "world" 408 assert any("hello" in ln for ln in lines) 409 assert any("world" in ln for ln in lines) 410 411 412 class TestIRCProtocolHelpersExtra: 413 414 def test_parse_malformed_no_space(self): 415 """A line starting with : but no space should not crash.""" 416 msg = _parse_irc_message(":justaprefix") 417 assert msg["prefix"] == "justaprefix" 418 assert msg["command"] == "" 419 assert msg["params"] == [] 420 421 def test_parse_empty(self): 422 msg = _parse_irc_message("") 423 assert msg["prefix"] == "" 424 assert msg["command"] == "" 425 assert msg["params"] == [] 426 427 428 class TestIRCAdapterMarkdown: 429 430 def test_strip_bold(self): 431 assert IRCAdapter._strip_markdown("**bold**") == "bold" 432 433 def test_strip_italic(self): 434 assert IRCAdapter._strip_markdown("*italic*") == "italic" 435 436 def test_strip_code(self): 437 assert IRCAdapter._strip_markdown("`code`") == "code" 438 439 def test_strip_link(self): 440 result = IRCAdapter._strip_markdown("[click here](https://example.com)") 441 assert result == "click here (https://example.com)" 442 443 def test_strip_image(self): 444 result = IRCAdapter._strip_markdown("") 445 assert result == "https://example.com/img.png" 446 447 448 # ── Requirements / validation ──────────────────────────────────────────── 449 450 451 class TestIRCRequirements: 452 453 def test_check_requirements_with_env(self, monkeypatch): 454 monkeypatch.setenv("IRC_SERVER", "irc.test.net") 455 monkeypatch.setenv("IRC_CHANNEL", "#test") 456 assert check_requirements() is True 457 458 def test_check_requirements_missing_server(self, monkeypatch): 459 monkeypatch.delenv("IRC_SERVER", raising=False) 460 monkeypatch.setenv("IRC_CHANNEL", "#test") 461 assert check_requirements() is False 462 463 def test_check_requirements_missing_channel(self, monkeypatch): 464 monkeypatch.setenv("IRC_SERVER", "irc.test.net") 465 monkeypatch.delenv("IRC_CHANNEL", raising=False) 466 assert check_requirements() is False 467 468 def test_validate_config_from_extra(self, monkeypatch): 469 for key in ("IRC_SERVER", "IRC_CHANNEL"): 470 monkeypatch.delenv(key, raising=False) 471 from gateway.config import PlatformConfig 472 cfg = PlatformConfig(extra={"server": "irc.test.net", "channel": "#test"}) 473 assert validate_config(cfg) is True 474 475 def test_validate_config_missing(self, monkeypatch): 476 for key in ("IRC_SERVER", "IRC_CHANNEL"): 477 monkeypatch.delenv(key, raising=False) 478 from gateway.config import PlatformConfig 479 cfg = PlatformConfig(extra={}) 480 assert validate_config(cfg) is False 481 482 483 # ── Plugin registration ────────────────────────────────────────────────── 484 485 486 class TestIRCPluginRegistration: 487 """Test the register() entry point.""" 488 489 def test_register_adds_to_registry(self, monkeypatch): 490 monkeypatch.setenv("IRC_SERVER", "irc.test.net") 491 monkeypatch.setenv("IRC_CHANNEL", "#test") 492 493 from gateway.platform_registry import platform_registry 494 495 # Clean up if already registered 496 platform_registry.unregister("irc") 497 498 ctx = MagicMock() 499 register(ctx) 500 ctx.register_platform.assert_called_once() 501 call_kwargs = ctx.register_platform.call_args 502 assert call_kwargs[1]["name"] == "irc" or call_kwargs[0][0] == "irc" if call_kwargs[0] else call_kwargs[1]["name"] == "irc"