ForcePilot/backend/test/unit/channels/test_irc_enhancement.py

from __future__ import annotations

from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from yuxi.channels.adapters.irc.accounts import has_configured_state
from yuxi.channels.adapters.irc.commands import sender_allowed_for_commands
from yuxi.channels.adapters.irc.runtime import (
    clear_irc_runtime,
    get_irc_runtime,
    record_irc_outbound_activity,
    set_irc_runtime,
)
from yuxi.channels.adapters.irc.security import (
    escape_irc_regex_literal,
    resolve_irc_group_sender_allowed,
)
from yuxi.channels.adapters.irc.splits import split_irc_text


class TestEscapeIrcRegexLiteral:
    """验证 escape_irc_regex_literal 安全正则构造"""

    def test_escapes_special_chars(self):
        escaped = escape_irc_regex_literal("nick[name]")
        assert "\\[" in escaped
        assert "\\]" in escaped

    def test_escapes_dot(self):
        escaped = escape_irc_regex_literal("bot.test")
        assert "\\." in escaped

    def test_handles_empty_string(self):
        assert escape_irc_regex_literal("") == ""

    def test_simple_nick_unchanged(self):
        escaped = escape_irc_regex_literal("MyBot")
        assert escaped == "MyBot"

    def test_escapes_parens(self):
        escaped = escape_irc_regex_literal("bot(test)")
        assert "\\(" in escaped
        assert "\\)" in escaped

    def test_escapes_star(self):
        escaped = escape_irc_regex_literal("bot*")
        assert "\\*" in escaped

    def test_escapes_plus(self):
        escaped = escape_irc_regex_literal("bot+")
        assert "\\+" in escaped

    def test_escapes_question(self):
        escaped = escape_irc_regex_literal("bot?")
        assert "\\?" in escaped

    def test_escapes_backslash(self):
        escaped = escape_irc_regex_literal("bot\\test")
        assert "\\\\" in escaped


class TestSenderAllowedForCommands:
    """验证 sender_allowed_for_commands 命令白名单匹配"""

    def test_exact_nick_match(self):
        assert sender_allowed_for_commands(
            "trusted!user@host", "trusted", ["trusted"]
        ) is True

    def test_full_mask_match(self):
        assert sender_allowed_for_commands(
            "trusted!user@host", "trusted", ["trusted!user@host"]
        ) is True

    def test_no_match(self):
        assert sender_allowed_for_commands(
            "untrusted!user@host", "untrusted", ["trusted"]
        ) is False

    def test_empty_allowlist(self):
        assert sender_allowed_for_commands(
            "anyone!user@host", "anyone", []
        ) is False

    def test_wildcard_match(self):
        assert sender_allowed_for_commands(
            "anyone!user@host", "anyone", ["*"]
        ) is True

    def test_nick_host_match(self):
        assert sender_allowed_for_commands(
            "someone!user@example.com", "someone", ["someone@example.com"]
        ) is True


class TestHasConfiguredState:
    """验证 has_configured_state 环境变量判定"""

    def test_returns_false_when_no_env(self):
        with patch("os.environ", {}):
            assert has_configured_state() is False

    def test_returns_true_when_both_set(self):
        env = {"IRC_HOST": "irc.libera.chat", "IRC_NICK": "MyBot"}
        with patch("os.environ", env):
            assert has_configured_state() is True

    def test_returns_false_when_only_host(self):
        env = {"IRC_HOST": "irc.libera.chat"}
        with patch("os.environ", env):
            assert has_configured_state() is False

    def test_returns_false_when_only_nick(self):
        env = {"IRC_NICK": "MyBot"}
        with patch("os.environ", env):
            assert has_configured_state() is False


class TestResolveIrcGroupSenderAllowed:
    """验证 resolve_irc_group_sender_allowed 三层回退"""

    def test_inner_allowlist_match(self):
        groups = {"#test": {"allowFrom": ["trusted!user@host"]}}
        allowed, layer, match = resolve_irc_group_sender_allowed(
            "#test", "trusted!user@host", "trusted", groups, [], "allowlist"
        )
        assert allowed is True
        assert layer == "group_inner_allowlist"
        assert match == "trusted!user@host"

    def test_outer_allowlist_match(self):
        allowed, layer, match = resolve_irc_group_sender_allowed(
            "#test", "trusted!user@host", "trusted", {}, ["trusted!user@host"], "allowlist"
        )
        assert allowed is True
        assert layer == "group_outer_allowlist"
        assert match == "trusted!user@host"

    def test_group_policy_open(self):
        allowed, layer, match = resolve_irc_group_sender_allowed(
            "#test", "anyone!user@host", "anyone", {}, [], "open"
        )
        assert allowed is True
        assert layer == "group_policy_open"
        assert match is None

    def test_channel_policy_open(self):
        groups = {"#test": {"groupPolicy": "open"}}
        allowed, layer, match = resolve_irc_group_sender_allowed(
            "#test", "anyone!user@host", "anyone", groups, [], "allowlist"
        )
        assert allowed is True
        assert layer == "channel_policy_open"
        assert match is None

    def test_blocked(self):
        allowed, layer, match = resolve_irc_group_sender_allowed(
            "#test", "untrusted!user@host", "untrusted", {}, [], "allowlist"
        )
        assert allowed is False
        assert layer == "blocked"
        assert match is None


class TestRuntimeStore:
    """验证 Runtime Store 机制"""

    def test_set_and_get(self):
        store = {"key": "value"}
        set_irc_runtime(store)
        assert get_irc_runtime() is store

    def test_clear(self):
        set_irc_runtime({"key": "value"})
        clear_irc_runtime()
        assert get_irc_runtime() is None

    def test_get_when_not_set(self):
        clear_irc_runtime()
        assert get_irc_runtime() is None

    def test_record_activity_adds_entry(self):
        store: dict = {}
        set_irc_runtime(store)
        record_irc_outbound_activity("#test", "hello world")
        assert "outbound_activity" in store
        assert len(store["outbound_activity"]) == 1
        assert store["outbound_activity"][0]["target"] == "#test"
        assert store["outbound_activity"][0]["text"] == "hello world"

    def test_record_activity_uninitialized(self):
        clear_irc_runtime()
        record_irc_outbound_activity("#test", "hello world")
        assert get_irc_runtime() is None


class TestSplitIrcText:
    """验证 split_irc_text 独立工具函数"""

    def test_short_text_unchanged(self):
        result = split_irc_text("hello", max_chars=350)
        assert result == ["hello"]

    def test_long_text_split(self):
        long_text = "A" * 2000
        result = split_irc_text(long_text, max_chars=50)
        assert len(result) > 1
        assert all(len(chunk.encode("utf-8")) <= 200 for chunk in result)

    def test_empty_text(self):
        assert split_irc_text("") == [""]

    def test_max_chars_zero(self):
        result = split_irc_text("hello", max_chars=0)
        assert "hello" in result[0]