from __future__ import annotations import asyncio from unittest.mock import AsyncMock, MagicMock, patch import pytest from yuxi.channels.adapters.irc.auth import ( nick_recover, nickserv_identify, sasl_authenticate, ) from yuxi.channels.adapters.irc.connection import IRCConnection from yuxi.channels.adapters.irc.adapter import IRCAdapter class TestAuthPasswordFix: """验证 auth.py 密码掩码修复 - 确保密码正确发送而非硬编码 ***""" @pytest.mark.asyncio async def test_nickserv_identify_sends_real_password(self): send_fn = MagicMock() await nickserv_identify(send_fn, "my_secret_password") send_fn.assert_called_once() call_arg = send_fn.call_args[0][0] assert "IDENTIFY" in call_arg assert "my_secret_password" in call_arg assert "***" not in call_arg @pytest.mark.asyncio async def test_nickserv_identify_sends_complex_password(self): send_fn = MagicMock() await nickserv_identify(send_fn, "P@ss!w0rd##2024") call_arg = send_fn.call_args[0][0] assert "P@ss!w0rd##2024" in call_arg assert "***" not in call_arg @pytest.mark.asyncio async def test_nick_recover_sends_real_password(self): send_fn = MagicMock() await nick_recover(send_fn, "mybot", "recovery_pass") ghost_call = send_fn.call_args_list[0][0][0] assert "GHOST" in ghost_call assert "mybot" in ghost_call assert "recovery_pass" in ghost_call assert "***" not in ghost_call @pytest.mark.asyncio async def test_nick_recover_resets_nick_after_ghost(self): send_fn = MagicMock() await nick_recover(send_fn, "original_nick", "pass") nick_call = send_fn.call_args_list[1][0][0] assert nick_call == "NICK original_nick" class TestConnectionPingTokenFix: """验证 connection.py PING token 修复 - 使用 server 名而非 keepalive 时间戳""" def test_send_line_truncation_logs_warning(self): conn = IRCConnection({"server": "irc.test.com", "port": 6667, "use_tls": False}) conn._writer = MagicMock() long_line = "A" * 600 with patch("yuxi.channels.adapters.irc.connection.logger") as mock_logger: conn.send_line(long_line) mock_logger.warning.assert_called_once() warning_msg = mock_logger.warning.call_args[0][0] assert "truncated" in warning_msg.lower() def test_send_line_no_truncation_for_short_message(self): conn = IRCConnection({"server": "irc.test.com", "port": 6667, "use_tls": False}) conn._writer = MagicMock() with patch("yuxi.channels.adapters.irc.connection.logger") as mock_logger: conn.send_line("short") mock_logger.warning.assert_not_called() def test_keepalive_ping_uses_server_name(self): conn = IRCConnection({"server": "irc.example.com", "port": 6697, "use_tls": True}) conn._writer = MagicMock() conn._last_pong = asyncio.get_event_loop().time() + 99999 with patch("yuxi.channels.adapters.irc.connection.logger"): with patch.object(conn, "_keepalive_loop"): pass assert conn._server == "irc.example.com" @pytest.mark.asyncio async def test_reconnect_triggered_via_create_task(self): """验证重连使用 asyncio.create_task 而非直接 await (避免阻塞 keepalive 循环)""" import inspect source = inspect.getsource(IRCConnection._keepalive_loop) assert "asyncio.create_task(self._reconnect())" in source assert "await self._reconnect()" not in source class TestTLSImprovements: """验证 TLS 安全性改进""" @pytest.mark.asyncio async def test_tls_context_has_check_hostname(self): with patch("asyncio.open_connection", new_callable=AsyncMock) as mock_open: mock_open.return_value = (MagicMock(), MagicMock()) conn = IRCConnection({"server": "irc.test.com", "port": 6697, "use_tls": True}) await conn.connect() call_kwargs = mock_open.call_args.kwargs assert "server_hostname" in call_kwargs assert call_kwargs["server_hostname"] == "irc.test.com" class TestMarkPongPublicMethod: """验证 IRCAdapter.mark_pong() 公共方法""" @pytest.mark.asyncio async def test_mark_pong_delegates_to_connection(self): adapter = IRCAdapter({"server": "irc.test.com", "port": 6667, "use_tls": False}) adapter._connection.mark_pong = MagicMock() adapter.mark_pong() adapter._connection.mark_pong.assert_called_once() class TestNickRetryStrategy: """验证备用昵称策略""" @pytest.mark.asyncio async def test_nick_retry_count_configured(self): from yuxi.channels.adapters.irc.adapter import _MAX_NICK_RETRIES assert _MAX_NICK_RETRIES == 3 @pytest.mark.asyncio async def test_fallback_nick_format(self): """验证备用昵称格式为 original_1, original_2, original_3""" adapter = IRCAdapter({ "server": "irc.test.com", "port": 6667, "use_tls": False, "nick": "MyBot", }) assert adapter.nick == "MyBot" from yuxi.channels.adapters.irc.adapter import _MAX_NICK_RETRIES for i in range(1, _MAX_NICK_RETRIES + 1): fallback = f"MyBot_{i}" assert fallback.startswith("MyBot_") assert str(i) in fallback class TestSASLAuthenticate: """验证 SASL 认证流程""" @pytest.mark.asyncio async def test_sasl_authenticate_send_chunks(self): send_fn = MagicMock() read_fn = AsyncMock() read_fn.side_effect = [ ":server CAP * ACK :sasl", ":server AUTHENTICATE +", ":server 900 nick nick!user@host :You are now logged in as nick", ] await sasl_authenticate(send_fn, read_fn, "testuser", "testpass", timeout=5.0) authenticate_calls = [ c[0][0] for c in send_fn.call_args_list if "AUTHENTICATE" in c[0][0] ] assert any("AUTHENTICATE PLAIN" == c for c in authenticate_calls) @pytest.mark.asyncio async def test_sasl_authenticate_sends_base64_credentials(self): import base64 send_fn = MagicMock() read_fn = AsyncMock() read_fn.side_effect = [ ":server CAP * ACK :sasl", ":server AUTHENTICATE +", ":server 900 nick nick!user@host :You are now logged in", ] username = "ircuser" password = "secret123" await sasl_authenticate(send_fn, read_fn, username, password, timeout=5.0) expected_plain = f"{username}\x00{username}\x00{password}" expected_b64 = base64.b64encode(expected_plain.encode()).decode() auth_chunks = [ c[0][0] for c in send_fn.call_args_list if c[0][0].startswith("AUTHENTICATE ") and c[0][0] != "AUTHENTICATE PLAIN" and c[0][0] != "AUTHENTICATE +" ] chunk_data = "".join(c.replace("AUTHENTICATE ", "") for c in auth_chunks) assert expected_b64 in chunk_data