from __future__ import annotations from yuxi.channels.adapters.irc.protocol import ( extract_nick, format_ctcp, is_ctcp_message, parse_ctcp, parse_irc_line, parse_irc_prefix, parse_irc_tags, ) from yuxi.channels.adapters.irc.splits import calc_effective_limit, split_utf8 from yuxi.channels.adapters.irc.normalizer import normalize_inbound from yuxi.channels.adapters.irc.session import ( is_channel, normalize_irc_messaging_target, resolve_chat_id, resolve_channel_user_id, resolve_targets, resolve_thread_id, ) from yuxi.channels.adapters.irc.sanitize import ( decode_literal_escapes, has_irc_control_chars, has_literal_escapes, is_irc_control_char, sanitize_inbound_text, sanitize_outbound_text, sanitize_irc_target, strip_irc_control_chars, ) from yuxi.channels.adapters.irc.setup import prepare_setup_wizard, run_setup_wizard from yuxi.channels.adapters.irc.config_schema import validate_config as schema_validate_config from yuxi.channels.adapters.irc.channel_runtime import ChannelRuntime from yuxi.channels.models import ChannelType, ChatType, EventType, MessageType class TestIRCProtocol: def test_parse_simple_privmsg(self): line = ":nick!user@host PRIVMSG #channel :hello world" prefix, command, args = parse_irc_line(line) assert prefix == "nick!user@host" assert command == "PRIVMSG" assert args == ["#channel", "hello world"] def test_parse_ping(self): prefix, command, args = parse_irc_line("PING :server.example.com") assert prefix is None assert command == "PING" assert args == [":server.example.com"] def test_parse_notice(self): line = ":NickServ!service@services NOTICE user :Password accepted" prefix, command, args = parse_irc_line(line) assert prefix == "NickServ!service@services" assert command == "NOTICE" assert args == ["user", "Password accepted"] def test_parse_numeric(self): line = ":irc.server 376 botnick :End of /MOTD command" prefix, command, args = parse_irc_line(line) assert prefix == "irc.server" assert command == "376" assert args == ["botnick", "End of /MOTD command"] def test_parse_with_tags(self): line = "@msgid=123;time=2024-01-01 :nick!user@host PRIVMSG #chan :hi" prefix, command, args = parse_irc_line(line) assert prefix == "nick!user@host" assert command == "PRIVMSG" assert args == ["#chan", "hi"] def test_parse_invalid_line(self): prefix, command, args = parse_irc_line("garbage") assert prefix is None assert args == [] def test_extract_nick_simple(self): assert extract_nick("nick!user@host") == "nick" def test_extract_nick_no_user(self): assert extract_nick("nick") == "nick" def test_extract_nick_none(self): assert extract_nick(None) == "unknown" def test_parse_prefix_full(self): result = parse_irc_prefix("nick!user@host") assert result == {"nick": "nick", "user": "user", "host": "host"} def test_parse_prefix_nick_only(self): result = parse_irc_prefix("nick") assert result == {"nick": "nick", "user": None, "host": None} def test_parse_prefix_none(self): result = parse_irc_prefix(None) assert result == {"nick": None, "user": None, "host": None} def test_parse_tags(self): tags = parse_irc_tags("@msgid=abc;time=2024-01-01 PING") assert tags == {"msgid": "abc", "time": "2024-01-01"} def test_parse_tags_no_tags(self): assert parse_irc_tags("PING :server") is None def test_parse_tags_flag_only(self): tags = parse_irc_tags("@+typing PRIVMSG #chan :hi") assert tags == {"+typing": True} class TestCTCP: def test_is_ctcp(self): assert is_ctcp_message("\x01PING 123\x01") is True def test_is_not_ctcp(self): assert is_ctcp_message("hello") is False assert is_ctcp_message("\x01not") is False assert is_ctcp_message("ctcp\x01") is False def test_parse_ctcp_with_args(self): command, args = parse_ctcp("\x01PING 42\x01") assert command == "PING" assert args == "42" def test_parse_ctcp_no_args(self): command, args = parse_ctcp("\x01VERSION\x01") assert command == "VERSION" assert args == "" def test_format_ctcp_with_args(self): result = format_ctcp("PING", "42") assert result == "\x01PING 42\x01" def test_format_ctcp_no_args(self): result = format_ctcp("VERSION") assert result == "\x01VERSION\x01" class TestUTF8Split: def test_short_text_no_split(self): chunks = split_utf8("hello", 500) assert len(chunks) == 1 assert chunks[0] == "hello" def test_exact_limit(self): text = "a" * 100 limit = len(text.encode("utf-8")) chunks = split_utf8(text, limit) assert len(chunks) == 1 def test_over_limit_splits(self): text = "A" * 200 limit = 100 chunks = split_utf8(text, limit) assert len(chunks) > 1 for chunk in chunks: assert len(chunk.encode("utf-8")) <= limit def test_chinese_text_split(self): text = "测试" * 50 limit = 50 chunks = split_utf8(text, limit) for chunk in chunks: assert len(chunk.encode("utf-8")) <= limit def test_zero_limit_returns_full_text(self): chunks = split_utf8("hello", 0) assert chunks == ["hello"] def test_negative_limit_returns_full_text(self): chunks = split_utf8("hello", -1) assert chunks == ["hello"] def test_empty_text(self): chunks = split_utf8("", 500) assert chunks == [""] def test_calc_effective_limit(self): limit = calc_effective_limit("#chan", nick="bot", username="bot", server="irc.example.com") assert limit > 0 assert limit < 510 class TestChannelRuntime: def test_add_channel(self): rt = ChannelRuntime() rt.add_channel("#test") assert "#test" in rt.joined_channels def test_remove_channel(self): rt = ChannelRuntime() rt.add_channel("#test") rt.remove_channel("#test") assert "#test" not in rt.joined_channels def test_handle_names(self): rt = ChannelRuntime() rt.add_channel("#test") rt.handle_names("#test", "@Alice +Bob Charlie") members = rt.get_members("#test") assert members == {"Alice", "Bob", "Charlie"} def test_handle_join(self): rt = ChannelRuntime() rt.add_channel("#test") rt.handle_names("#test", "Alice") rt.handle_join("#test", "Bob") assert "Bob" in rt.get_members("#test") def test_handle_part(self): rt = ChannelRuntime() rt.add_channel("#test") rt.handle_names("#test", "Alice Bob") rt.handle_part("#test", "Alice") assert "Alice" not in rt.get_members("#test") def test_handle_kick(self): rt = ChannelRuntime() rt.add_channel("#test") rt.handle_names("#test", "Alice Bob") rt.handle_kick("#test", "Alice") assert "Alice" not in rt.get_members("#test") def test_get_members_unknown_channel(self): rt = ChannelRuntime() assert rt.get_members("#nonexistent") == set() class TestNormalizer: def test_direct_message(self): raw = { "sender_nick": "alice", "target": "ForcePilotBot", "text": "hello bot", "prefix": "alice!user@host", "command": "PRIVMSG", "raw_line": ":alice!user@host PRIVMSG ForcePilotBot :hello bot", } msg = normalize_inbound(raw, "irc", ChannelType.IRC, "ForcePilotBot") assert msg.identity.channel_user_id == "alice" assert msg.identity.channel_chat_id == "dm_alice" assert msg.chat_type == ChatType.DIRECT assert msg.content == "hello bot" assert msg.message_type == MessageType.TEXT assert msg.event_type == EventType.MESSAGE_RECEIVED def test_channel_message(self): raw = { "sender_nick": "bob", "target": "#general", "text": "hello everyone", "prefix": "bob!user@host", "command": "PRIVMSG", "raw_line": ":bob!user@host PRIVMSG #general :hello everyone", } msg = normalize_inbound(raw, "irc", ChannelType.IRC, "ForcePilotBot") assert msg.identity.channel_chat_id == "#general" assert msg.chat_type == ChatType.GROUP def test_bot_mentioned(self): raw = { "sender_nick": "alice", "target": "#general", "text": "hey ForcePilotBot help me", "prefix": "alice!user@host", "command": "PRIVMSG", "raw_line": ":alice!user@host PRIVMSG #general :hey ForcePilotBot help me", } msg = normalize_inbound(raw, "irc", ChannelType.IRC, "ForcePilotBot") assert msg.mentions is not None assert msg.mentions.is_bot_mentioned is True def test_bot_not_mentioned(self): raw = { "sender_nick": "alice", "target": "#general", "text": "hello everyone", "prefix": "alice!user@host", "command": "PRIVMSG", "raw_line": ":alice!user@host PRIVMSG #general :hello everyone", } msg = normalize_inbound(raw, "irc", ChannelType.IRC, "ForcePilotBot") assert msg.mentions is not None assert msg.mentions.is_bot_mentioned is False def test_explicit_mentions(self): raw = { "sender_nick": "alice", "target": "#general", "text": "@bob @charlie check this out", "prefix": "alice!user@host", "command": "PRIVMSG", "raw_line": ":alice!user@host PRIVMSG #general :@bob @charlie check this out", } msg = normalize_inbound(raw, "irc", ChannelType.IRC, "ForcePilotBot") assert msg.mentions is not None assert "bob" in msg.mentions.mentioned_user_ids assert "charlie" in msg.mentions.mentioned_user_ids def test_message_id_unique(self): raw = { "sender_nick": "alice", "target": "ForcePilotBot", "text": "hello", "prefix": "alice!user@host", "command": "PRIVMSG", "raw_line": ":alice!user@host PRIVMSG ForcePilotBot :hello", } msg1 = normalize_inbound(raw, "irc", ChannelType.IRC, "ForcePilotBot") msg2 = normalize_inbound(raw, "irc", ChannelType.IRC, "ForcePilotBot") assert msg1.identity.channel_message_id != msg2.identity.channel_message_id class TestSession: def test_resolve_chat_id_channel(self): assert resolve_chat_id("#general", "alice") == "#general" assert resolve_chat_id("&support", "alice") == "&support" def test_resolve_chat_id_dm(self): assert resolve_chat_id("ForcePilotBot", "alice") == "dm_alice" def test_resolve_channel_user_id(self): assert resolve_channel_user_id("alice") == "alice" def test_resolve_thread_id_channel(self): tid = resolve_thread_id("agent1", "#general") assert tid == "agent:agent1:irc:channel:general" def test_resolve_thread_id_dm(self): tid = resolve_thread_id("agent1", "dm_alice") assert tid == "agent:agent1:irc:dm:alice" def test_is_channel(self): assert is_channel("#general") is True assert is_channel("&support") is True assert is_channel("ForcePilotBot") is False class TestNormalizeIrcMessagingTarget: def test_raw_channel(self): result = normalize_irc_messaging_target("#general") assert result == {"kind": "group", "target": "#general"} def test_raw_ampersand_channel(self): result = normalize_irc_messaging_target("&support") assert result == {"kind": "group", "target": "&support"} def test_raw_user(self): result = normalize_irc_messaging_target("alice") assert result == {"kind": "user", "target": "alice"} def test_protocol_prefix_channel(self): result = normalize_irc_messaging_target("irc:channel:#general") assert result == {"kind": "group", "target": "#general"} def test_protocol_prefix_user(self): result = normalize_irc_messaging_target("irc:user:alice") assert result == {"kind": "user", "target": "alice"} def test_empty_target(self): assert normalize_irc_messaging_target("") is None assert normalize_irc_messaging_target(" ") is None def test_protocol_empty_channel(self): assert normalize_irc_messaging_target("irc:channel:") is None def test_protocol_empty_user(self): assert normalize_irc_messaging_target("irc:user:") is None def test_protocol_unknown_prefix(self): assert normalize_irc_messaging_target("irc:unknown:stuff") is None class TestResolveTargets: def test_group_auto_prepend_hash(self): result = resolve_targets("group", "general") assert result == {"kind": "group", "target": "#general"} def test_group_with_hash(self): result = resolve_targets("group", "#general") assert result == {"kind": "group", "target": "#general"} def test_group_with_ampersand(self): result = resolve_targets("group", "&support") assert result == {"kind": "group", "target": "&support"} def test_user_reject_channel(self): result = resolve_targets("user", "#general") assert result is None def test_user_reject_ampersand(self): result = resolve_targets("user", "&support") assert result is None def test_user_accepts_nick(self): result = resolve_targets("user", "alice") assert result == {"kind": "user", "target": "alice"} def test_empty_target(self): assert resolve_targets("group", "") is None assert resolve_targets("user", " ") is None def test_unknown_kind(self): assert resolve_targets("unknown", "stuff") is None class TestSanitize: def test_decode_literal_escapes_strips_x01(self): text = "hello\\x01world" result = decode_literal_escapes(text) assert result == "helloworld" def test_decode_literal_escapes_strips_multiple(self): text = "\\x01hidden\\x02ctrl\\x1fend" result = decode_literal_escapes(text) assert result == "hiddenctrlend" def test_decode_literal_escapes_case_insensitive(self): text = "\\x0ATEST\\x0aend" result = decode_literal_escapes(text) assert result == "TESTend" def test_decode_literal_escapes_no_escapes(self): text = "normal message without escapes" result = decode_literal_escapes(text) assert result == text def test_has_literal_escapes_true(self): assert has_literal_escapes("test\\x01") is True def test_has_literal_escapes_false(self): assert has_literal_escapes("normal text") is False def test_has_literal_escapes_backslash_not_escape(self): assert has_literal_escapes("just a \\ backslash") is False def test_sanitize_inbound_text_removes_escapes_and_ctrl(self): text = "\\x01hello\x02world" result = sanitize_inbound_text(text) assert "\x02" not in result assert "\\x01" not in result def test_strip_irc_control_chars(self): assert "\x01" not in strip_irc_control_chars("\x01hello\x02") assert strip_irc_control_chars("clean") == "clean" def test_has_irc_control_chars(self): assert has_irc_control_chars("\x01test") is True assert has_irc_control_chars("normal") is False def test_is_irc_control_char(self): assert is_irc_control_char("\x01") is True assert is_irc_control_char("\x7F") is True assert is_irc_control_char("a") is False def test_sanitize_outbound_text(self): assert "\n" not in sanitize_outbound_text("line1\nline2") assert "\r" not in sanitize_outbound_text("line1\rline2") def test_sanitize_irc_target(self): result = sanitize_irc_target("test target") assert " " not in result class TestSetupWizardPrepare: def test_prepare_empty_env(self, monkeypatch): for key in ("IRC_HOST", "IRC_NICK", "IRC_TLS", "IRC_CHANNELS"): monkeypatch.delenv(key, raising=False) answers = prepare_setup_wizard() assert answers == {} def test_prepare_detects_host(self, monkeypatch): monkeypatch.setenv("IRC_HOST", "irc.libera.chat") monkeypatch.delenv("IRC_NICK", raising=False) answers = prepare_setup_wizard() assert answers.get("server") == "irc.libera.chat" def test_prepare_detects_port(self, monkeypatch): monkeypatch.setenv("IRC_PORT", "7000") answers = prepare_setup_wizard() assert answers.get("port") == 7000 def test_prepare_detects_tls(self, monkeypatch): monkeypatch.setenv("IRC_TLS", "false") answers = prepare_setup_wizard() assert answers.get("use_tls") is False class TestConfigSchemaCrossField: def test_nickserv_register_requires_email(self): errors = schema_validate_config({ "server": "irc.libera.chat", "nick": "testbot", "nickserv_register": True, }) assert any("nickserv_register_email" in e for e in errors) def test_nickserv_register_with_email_ok(self): errors = schema_validate_config({ "server": "irc.libera.chat", "nick": "testbot", "nickserv_register": True, "nickserv_register_email": "bot@example.com", }) assert not any("nickserv_register_email" in e for e in errors) def test_dm_policy_open_requires_allowfrom_star(self): errors = schema_validate_config({ "server": "irc.libera.chat", "nick": "testbot", "dm_policy": "open", "allowFrom": [], }) assert any("allowFrom" in e for e in errors) def test_dm_policy_open_with_star_ok(self): errors = schema_validate_config({ "server": "irc.libera.chat", "nick": "testbot", "dm_policy": "open", "allowFrom": ["*"], }) assert not any("allowFrom" in e for e in errors) def test_dm_policy_pairing_no_warning(self): errors = schema_validate_config({ "server": "irc.libera.chat", "nick": "testbot", "dm_policy": "pairing", "allowFrom": [], }) assert len(errors) == 0 class TestSetupWizardGroupAccess: def test_group_access_sets_allowfrom(self): answers = { "server": "irc.libera.chat", "nick": "testbot", "groupAccess": "user1!*@*, user2!*@*", } result = run_setup_wizard(answers) assert result.success assert result.config.get("groupAllowFrom") == ["user1!*@*", "user2!*@*"] def test_allow_from_sets_from_answers(self): answers = { "server": "irc.libera.chat", "nick": "testbot", "allowFrom": "alice!*@*, bob!*@*", } result = run_setup_wizard(answers) assert result.success assert result.config.get("allowFrom") == ["alice!*@*", "bob!*@*"]