from __future__ import annotations import hashlib import time from yuxi.channels.adapters.mattermost.accounts import ( DEFAULT_ACCOUNT_ID, MattermostAccount, MultiAccountConfig, merge_config, ) from yuxi.channels.adapters.mattermost.approval import ( APPROVAL_TIMEOUT_S, ApprovalConfig, ApprovalManager, ApprovalRequest, ) from yuxi.channels.adapters.mattermost.cache import ( LRUCache, MattermostChannelCache, SentMessageCache, TTLCache, ) from yuxi.channels.adapters.mattermost.config_schema import ( MATTERMOST_CONFIG_SCHEMA, get_config_schema, validate_config_schema, ) from yuxi.channels.adapters.mattermost.config_writes import ConfigWritesManager from yuxi.channels.adapters.mattermost.debounce import InboundDebouncer from yuxi.channels.adapters.mattermost.directory import ( DirectoryGroup, DirectoryPeer, DirectorySnapshot, build_group_from_channel_data, build_peer_from_user_data, ) from yuxi.channels.adapters.mattermost.model_picker import ( MODEL_PICKER_PAGE_SIZE, ModelOption, PickerState, build_model_picker_actions, build_model_picker_attachment, build_model_select_attachment, get_available_providers, get_default_model_options, ) from yuxi.channels.adapters.mattermost.monitor import ( MODE_ONCHAR, MODE_ONMESSAGE, MentionGate, MentionGateConfig, is_control_command, resolve_control_command_gate, ) from yuxi.channels.adapters.mattermost.normalizer import ( check_bot_mentioned, detect_message_type, extract_attachments, extract_mentions, extract_urls, parse_channel_json, parse_post_json, ) from yuxi.channels.adapters.mattermost.pairing import ( PAIRING_CODE_LENGTH, PAIRING_REQUEST_TTL_S, MattermostPairingManager, PairingRequest, build_pairing_notification, ) from yuxi.channels.adapters.mattermost.polls import ( build_poll_props, build_yes_no_poll_props, parse_poll_action, ) from yuxi.channels.adapters.mattermost.reply import ( REPLY_ALL, REPLY_OFF, ReplyConfig, ReplyManager, ) from yuxi.channels.adapters.mattermost.secret import ( SecretDescriptor, build_secret_target_registry_entries, collect_runtime_config_assignments, get_secret_contract, list_configured_secrets, resolve_secret, ) from yuxi.channels.adapters.mattermost.security import ( DM_POLICY_DISABLED, DM_POLICY_OPEN, DM_POLICY_PAIRING, GROUP_POLICY_ALLOWLIST, GROUP_POLICY_DISABLED, GROUP_POLICY_OPEN, MattermostSecurity, MattermostSecurityConfig, is_allow_entry_match, normalize_allow_entry, ) from yuxi.channels.adapters.mattermost.send import ( _split_hard, build_patch_options, build_post_options, chunk_text_for_outbound, ) from yuxi.channels.adapters.mattermost.session import ( build_thread_id, resolve_chat_id, resolve_chat_type, ) from yuxi.channels.adapters.mattermost.slash import ( DEFAULT_COMMANDS, ManagedSlashCommand, SlashCommand, SlashCommandConfig, build_command_payload, build_skill_commands, get_supported_commands, ) from yuxi.channels.adapters.mattermost.target_resolution import ( looks_like_mattermost_target_id, parse_target, resolve_channel_target, resolve_user_target, ) from yuxi.channels.models import ChatType, MessageType # ============================================================ # config_schema.py # ============================================================ class TestValidateConfigSchema: def test_valid_minimal_config(self): config = {"server_url": "https://example.com", "bot_token": "token123"} errors = validate_config_schema(config) assert errors == [] def test_missing_server_url(self): config = {"bot_token": "token123"} errors = validate_config_schema(config) assert any("server_url" in e for e in errors) def test_missing_bot_token(self): config = {"server_url": "https://example.com"} errors = validate_config_schema(config) assert any("bot_token" in e for e in errors) def test_invalid_dm_policy(self): config = {"server_url": "https://example.com", "bot_token": "x", "dm_policy": "invalid"} errors = validate_config_schema(config) assert any("dm_policy" in e for e in errors) def test_invalid_group_policy(self): config = {"server_url": "https://example.com", "bot_token": "x", "group_policy": "invalid"} errors = validate_config_schema(config) assert any("group_policy" in e for e in errors) def test_invalid_chatmode(self): config = {"server_url": "https://example.com", "bot_token": "x", "chatmode": "invalid"} errors = validate_config_schema(config) assert any("chatmode" in e for e in errors) def test_text_chunk_limit_out_of_range(self): config = {"server_url": "https://example.com", "bot_token": "x", "text_chunk_limit": 500} errors = validate_config_schema(config) assert any("text_chunk_limit" in e for e in errors) config2 = {"server_url": "https://example.com", "bot_token": "x", "text_chunk_limit": 20000} errors2 = validate_config_schema(config2) assert any("text_chunk_limit" in e for e in errors2) def test_valid_boundary_text_chunk_limit(self): config = {"server_url": "https://example.com", "bot_token": "x", "text_chunk_limit": 1000} errors = validate_config_schema(config) assert not any("text_chunk_limit" in e for e in errors) config2 = {"server_url": "https://example.com", "bot_token": "x", "text_chunk_limit": 16383} errors2 = validate_config_schema(config2) assert not any("text_chunk_limit" in e for e in errors2) def test_all_valid_policies(self): for dm in ["open", "allowlist", "pairing", "disabled"]: for gp in ["open", "allowlist", "disabled"]: for cm in ["oncall", "onmessage", "onchar"]: config = { "server_url": "https://example.com", "bot_token": "x", "dm_policy": dm, "group_policy": gp, "chatmode": cm, } errors = validate_config_schema(config) assert errors == [], f"Failed for dm={dm} gp={gp} cm={cm}: {errors}" def test_empty_config(self): errors = validate_config_schema({}) assert len(errors) == 2 class TestGetConfigSchema: def test_returns_dict(self): result = get_config_schema() assert isinstance(result, dict) assert result == MATTERMOST_CONFIG_SCHEMA def test_required_keys(self): result = get_config_schema() assert "required" in result assert "server_url" in result["required"] assert "bot_token" in result["required"] def test_schema_version(self): result = get_config_schema() assert result["$schema"].startswith("http://json-schema.org") # ============================================================ # normalizer.py # ============================================================ class TestParseJsonFields: def test_parse_post_json_valid(self): data = {"post": '{"id":"m1","message":"hello"}'} result = parse_post_json(data) assert result == {"id": "m1", "message": "hello"} def test_parse_post_json_invalid(self): data = {"post": "not-json"} result = parse_post_json(data) assert result == {} def test_parse_post_json_missing(self): data = {} result = parse_post_json(data) assert result == {} def test_parse_post_json_already_dict(self): data = {"post": {"id": "m1"}} result = parse_post_json(data) assert result == {"id": "m1"} def test_parse_post_json_none(self): data = {"post": None} result = parse_post_json(data) assert result == {} def test_parse_channel_json_valid(self): data = {"channel": '{"type":"D","display_name":"tester"}'} result = parse_channel_json(data) assert result == {"type": "D", "display_name": "tester"} def test_parse_channel_json_invalid(self): data = {"channel": "bad json!!!"} result = parse_channel_json(data) assert result == {} def test_parse_channel_json_missing(self): data = {} result = parse_channel_json(data) assert result == {} class TestExtractMentions: def test_single_mention(self): result = extract_mentions("hello @user1 how are you") assert result == ["user1"] def test_multiple_mentions(self): result = extract_mentions("@alice @bob @charlie test") assert result == ["alice", "bob", "charlie"] def test_mention_with_dots(self): result = extract_mentions("@user.test here") assert result == ["user.test"] def test_mention_with_hyphen(self): result = extract_mentions("@user-name test") assert result == ["user-name"] def test_no_mention(self): result = extract_mentions("hello world") assert result == [] def test_empty_text(self): result = extract_mentions("") assert result == [] def test_none_text(self): result = extract_mentions(None) assert result == [] def test_email_not_mention(self): result = extract_mentions("test@example.com") assert result == ["example.com"] class TestCheckBotMentioned: def test_bot_mentioned_case_insensitive(self): assert check_bot_mentioned("hello @FORCEPILOT test", "forcepilot") is True def test_bot_not_mentioned(self): assert check_bot_mentioned("hello world", "forcepilot") is False def test_empty_text(self): assert check_bot_mentioned("", "bot") is False def test_empty_bot_username(self): assert check_bot_mentioned("@bot hello", "") is False def test_none_text(self): assert check_bot_mentioned(None, "bot") is False def test_exact_match_in_longer_message(self): text = "some text @ForcePilot can you help?" assert check_bot_mentioned(text, "ForcePilot") is True class TestExtractUrls: def test_single_url(self): result = extract_urls("check https://example.com/page") assert result == ["https://example.com/page"] def test_multiple_urls(self): result = extract_urls("a https://a.com b https://b.com/path?q=1") assert result == ["https://a.com", "https://b.com/path?q=1"] def test_no_url(self): result = extract_urls("hello world") assert result == [] def test_empty_text(self): result = extract_urls("") assert result == [] def test_none_text(self): result = extract_urls(None) assert result == [] def test_http_url(self): result = extract_urls("see http://example.com") assert result == ["http://example.com"] class TestDetectMessageType: def test_text_message(self): assert detect_message_type({}) == MessageType.TEXT def test_file_message(self): assert detect_message_type({"file_ids": ["f1"]}) == MessageType.FILE def test_empty_file_ids(self): assert detect_message_type({"file_ids": []}) == MessageType.TEXT class TestExtractAttachments: def test_with_file_ids(self): result = extract_attachments({"file_ids": ["f1", "f2"]}) assert len(result) == 2 assert result[0].file_id == "f1" assert result[0].type == "file" assert result[1].file_id == "f2" def test_no_file_ids(self): result = extract_attachments({}) assert result == [] def test_empty_file_ids(self): result = extract_attachments({"file_ids": []}) assert result == [] # ============================================================ # accounts.py # ============================================================ class TestMattermostAccount: def test_default_values(self): a = MattermostAccount(account_id="test1") assert a.account_id == "test1" assert a.server_url == "" assert a.bot_token == "" assert a.nick == "ForcePilot" assert a.username == "" assert a.channels == [] assert a.enabled is True def test_configured_true(self): a = MattermostAccount(account_id="t", server_url="https://x.com", bot_token="tok") assert a.configured is True def test_configured_false(self): a = MattermostAccount(account_id="t") assert a.configured is False def test_configured_missing_token(self): a = MattermostAccount(account_id="t", server_url="https://x.com") assert a.configured is False def test_from_config_entry_full(self): entry = { "server_url": "https://mm.example.com", "bot_token": "secret123", "nick": "MyBot", "username": "mybot", "realname": "My Bot", "channels": ["ch1", "ch2"], "enabled": False, } result = MattermostAccount.from_config_entry("acc1", entry) assert result.account_id == "acc1" assert result.server_url == "https://mm.example.com" assert result.bot_token == "secret123" assert result.nick == "MyBot" assert result.username == "mybot" assert result.realname == "My Bot" assert result.channels == ["ch1", "ch2"] assert result.enabled is False def test_from_config_entry_minimal(self): result = MattermostAccount.from_config_entry("acc2", {}) assert result.account_id == "acc2" assert result.server_url == "" assert result.bot_token == "" assert result.configured is False def test_from_config_entry_legacy_fields(self): entry = {"host": "https://old.example.com", "token": "old_token"} result = MattermostAccount.from_config_entry("acc3", entry) assert result.server_url == "https://old.example.com" assert result.bot_token == "old_token" class TestMultiAccountConfig: def test_from_config_with_accounts(self): config = { "accounts": { "a1": {"server_url": "https://s1.com", "bot_token": "t1"}, "a2": {"server_url": "https://s2.com", "bot_token": "t2"}, }, "default_account_id": "a1", } mac = MultiAccountConfig.from_config(config) assert len(mac.accounts) == 2 assert mac.default_account_id == "a1" assert mac.accounts["a1"].server_url == "https://s1.com" def test_from_config_without_accounts(self): config = {"server_url": "https://solo.com", "bot_token": "t"} mac = MultiAccountConfig.from_config(config) assert len(mac.accounts) == 1 assert mac.default_account_id == DEFAULT_ACCOUNT_ID assert mac.accounts[DEFAULT_ACCOUNT_ID].server_url == "https://solo.com" def test_from_config_empty(self): mac = MultiAccountConfig.from_config({}) assert len(mac.accounts) == 1 assert mac.accounts[DEFAULT_ACCOUNT_ID].server_url == "" def test_resolve_account_explicit(self): config = { "accounts": { "a1": {"server_url": "https://s1.com", "bot_token": "t1"}, "a2": {"server_url": "https://s2.com", "bot_token": "t2"}, }, } mac = MultiAccountConfig.from_config(config) result = mac.resolve_account("a2") assert result.account_id == "a2" def test_resolve_account_fallback_default(self): config = {"accounts": {"a1": {"server_url": "https://s1.com", "bot_token": "t1"}}} mac = MultiAccountConfig.from_config(config) result = mac.resolve_account("nonexistent") assert result.account_id == "a1" def test_resolve_account_not_found(self): mac = MultiAccountConfig() result = mac.resolve_account() assert result is None def test_list_account_ids(self): config = {"accounts": {"a1": {"server_url": "x", "bot_token": "x"}}} mac = MultiAccountConfig.from_config(config) assert mac.list_account_ids() == ["a1"] def test_list_configured_accounts(self): config = { "accounts": { "a1": {"server_url": "https://s1.com", "bot_token": "t1"}, "a2": {"server_url": "", "bot_token": ""}, "a3": {"server_url": "https://s3.com", "bot_token": "t3", "enabled": False}, }, } mac = MultiAccountConfig.from_config(config) configured = mac.list_configured_accounts() assert len(configured) == 1 assert configured[0].account_id == "a1" def test_default_account_id_auto_detect(self): config = {"accounts": {"b1": {"server_url": "x", "bot_token": "x"}}} mac = MultiAccountConfig.from_config(config) assert mac.default_account_id == "b1" class TestMergeConfig: def test_basic_merge(self): top = {"server_url": "https://top.com", "bot_token": "top-token"} account = {"bot_token": "acc-token"} result = merge_config(top, account) assert result["server_url"] == "https://top.com" assert result["bot_token"] == "acc-token" def test_commands_merge(self): top = {"commands": {"cmd1": "desc1"}} account = {"commands": {"cmd2": "desc2"}} result = merge_config(top, account) assert result["commands"] == {"cmd1": "desc1", "cmd2": "desc2"} def test_interactions_merge(self): top = {"interactions": {"int1": "d1"}} account = {"interactions": {"int2": "d2"}} result = merge_config(top, account) assert result["interactions"] == {"int1": "d1", "int2": "d2"} def test_merge_with_empty_account(self): top = {"server_url": "x", "bot_token": "t", "commands": {"c1": "d1"}} result = merge_config(top, {}) assert result["server_url"] == "x" assert result["commands"] == {"c1": "d1"} # ============================================================ # polls.py # ============================================================ class TestBuildPollProps: def test_basic_poll(self): result = build_poll_props("What to eat?", ["Pizza", "Burger", "Salad"], poll_id="p1") assert len(result["attachments"]) == 1 attachment = result["attachments"][0] assert len(attachment["actions"]) == 1 assert attachment["actions"][0]["type"] == "select" assert len(attachment["actions"][0]["options"]) == 3 assert result["props"]["poll_id"] == "p1" assert result["props"]["type"] == "poll" def test_poll_with_fewer_than_2_options(self): result = build_poll_props("Single?", ["Only"]) opts = result["attachments"][0]["actions"][0]["options"] assert len(opts) >= 2 def test_poll_with_more_than_5_options(self): result = build_poll_props("Many?", ["A", "B", "C", "D", "E", "F"]) opts = result["attachments"][0]["actions"][0]["options"] assert len(opts) == 5 def test_empty_options(self): result = build_poll_props("What?", []) opts = result["attachments"][0]["actions"][0]["options"] assert len(opts) == 1 def test_poll_without_id(self): result = build_poll_props("Q?", ["A", "B"]) assert result["props"]["poll_id"] == "" class TestBuildYesNoPoll: def test_basic(self): result = build_yes_no_poll_props("Approve?", poll_id="p2") assert len(result["attachments"]) == 1 actions = result["attachments"][0]["actions"] assert len(actions) == 2 assert actions[0]["text"] == "✅ 是" assert actions[1]["text"] == "❌ 否" assert result["props"]["type"] == "yes_no_poll" class TestParsePollAction: def test_full_context(self): context = {"action": "poll_vote", "poll_id": "p1", "vote": "yes"} result = parse_poll_action(context) assert result["action"] == "poll_vote" assert result["poll_id"] == "p1" assert result["vote"] == "yes" def test_empty_context(self): result = parse_poll_action({}) assert result["action"] == "" assert result["poll_id"] == "" assert result["vote"] == "" # ============================================================ # cache.py # ============================================================ class TestTTLCache: def test_set_and_get(self): cache = TTLCache(ttl_s=60) cache.set("key1", "value1") assert cache.get("key1") == "value1" def test_get_missing(self): cache = TTLCache() assert cache.get("missing") is None def test_delete(self): cache = TTLCache() cache.set("key1", "val1") assert cache.delete("key1") is True assert cache.get("key1") is None def test_delete_missing(self): cache = TTLCache() assert cache.delete("missing") is False def test_ttl_expiry(self): cache = TTLCache(ttl_s=0) cache.set("key1", "val1") assert cache.get("key1") is None def test_clear(self): cache = TTLCache() cache.set("a", 1) cache.set("b", 2) cache.clear() assert cache.get("a") is None assert cache.get("b") is None def test_max_size_eviction(self): cache = TTLCache(ttl_s=600, max_size=3) for i in range(5): cache.set(str(i), i) # Oldest entries evicted, should keep 3 count = sum(1 for i in range(5) if cache.get(str(i)) is not None) assert count <= 3 def test_value_types(self): cache = TTLCache() cache.set("int", 42) cache.set("dict", {"a": 1}) cache.set("list", [1, 2, 3]) assert cache.get("int") == 42 assert cache.get("dict") == {"a": 1} assert cache.get("list") == [1, 2, 3] class TestLRUCache: def test_set_and_get(self): cache = LRUCache(max_size=5) cache.set("k1", "v1") assert cache.get("k1") == "v1" def test_get_updates_lru_order(self): cache = LRUCache(max_size=5) cache.set("k1", "v1") cache.set("k2", "v2") cache.get("k1") # makes k1 most-recent assert cache.size() == 2 def test_max_size_eviction(self): cache = LRUCache(max_size=2) cache.set("a", 1) cache.set("b", 2) cache.set("c", 3) assert cache.size() <= 2 assert cache.get("a") is None # 'a' should be evicted def test_ttl_expiry(self): cache = LRUCache(ttl_s=0, max_size=10) cache.set("k1", "v1") assert cache.get("k1") is None def test_clear(self): cache = LRUCache() cache.set("a", 1) cache.clear() assert cache.size() == 0 def test_size(self): cache = LRUCache() assert cache.size() == 0 cache.set("a", 1) cache.set("b", 2) assert cache.size() == 2 class TestMattermostChannelCache: def test_initialization(self): mcc = MattermostChannelCache() assert mcc.bot_user is not None assert mcc.user_by_name is not None assert mcc.channel_by_name is not None assert mcc.dm_channel is not None def test_clear(self): mcc = MattermostChannelCache() mcc.user_by_name.set("user1", "id1") mcc.clear() assert mcc.user_by_name.get("user1") is None def test_stats(self): mcc = MattermostChannelCache() stats = mcc.stats() assert "bot_user" in stats assert "user_by_name" in stats assert "dm_channel" in stats class TestSentMessageCache: def test_record_and_get(self): cache = SentMessageCache(ttl_s=600) cache.record("msg1", "chat1") entry = cache.get("msg1") assert entry["msg_id"] == "msg1" assert entry["chat_id"] == "chat1" def test_get_missing(self): cache = SentMessageCache() assert cache.get("missing") is None def test_get_thread_id(self): cache = SentMessageCache() cache.record("msg1", "chat1", thread_id="thread1") assert cache.get_thread_id("msg1") == "thread1" def test_get_thread_id_missing(self): cache = SentMessageCache() assert cache.get_thread_id("missing") is None def test_get_thread_id_none(self): cache = SentMessageCache() cache.record("msg1", "chat1") assert cache.get_thread_id("msg1") is None def test_record_all_fields(self): cache = SentMessageCache() cache.record("msg1", "chat1", channel_id="ch1", thread_id="t1", user_id="u1") entry = cache.get("msg1") assert entry["channel_id"] == "ch1" assert entry["thread_id"] == "t1" assert entry["user_id"] == "u1" def test_ttl_expiry(self): cache = SentMessageCache(ttl_s=0) cache.record("msg1", "chat1") assert cache.get("msg1") is None def test_max_size_eviction(self): cache = SentMessageCache(max_size=2, ttl_s=600) cache.record("a", "ca") cache.record("b", "cb") cache.record("c", "cc") assert cache.size() <= 2 def test_clear(self): cache = SentMessageCache() cache.record("a", "ca") cache.clear() assert cache.size() == 0 def test_size(self): cache = SentMessageCache() assert cache.size() == 0 cache.record("a", "ca") assert cache.size() == 1 # ============================================================ # debounce.py # ============================================================ class TestInboundDebouncer: def test_first_message_accepted(self): debouncer = InboundDebouncer(ttl_ms=100) assert debouncer.should_process("ch1", "", "hello") is True def test_duplicate_rejected(self): debouncer = InboundDebouncer(ttl_ms=5000) assert debouncer.should_process("ch1", "", "hello") is True assert debouncer.should_process("ch1", "", "hello") is False def test_different_content_accepted(self): debouncer = InboundDebouncer(ttl_ms=5000) assert debouncer.should_process("ch1", "", "hello") is True assert debouncer.should_process("ch1", "", "world") is True def test_different_channel_accepted(self): debouncer = InboundDebouncer(ttl_ms=5000) assert debouncer.should_process("ch1", "", "hello") is True assert debouncer.should_process("ch2", "", "hello") is True def test_different_thread_accepted(self): debouncer = InboundDebouncer(ttl_ms=5000) assert debouncer.should_process("ch1", "t1", "hello") is True assert debouncer.should_process("ch1", "t2", "hello") is True def test_content_truncation(self): debouncer = InboundDebouncer(ttl_ms=5000) long_msg = "x" * 200 assert debouncer.should_process("ch1", "", long_msg) is True assert debouncer.should_process("ch1", "", long_msg) is False def test_prune_triggers_on_max(self): debouncer = InboundDebouncer(ttl_ms=0, max_entries=2) debouncer.should_process("a", "", "1") debouncer.should_process("b", "", "2") debouncer.should_process("c", "", "3") assert len(debouncer._entries) <= 2 def test_clear(self): debouncer = InboundDebouncer(ttl_ms=5000) debouncer.should_process("ch1", "", "hello") debouncer.clear() assert len(debouncer._entries) == 0 def test_default_values(self): debouncer = InboundDebouncer() assert debouncer._ttl_ms == 1000 assert debouncer._max_entries == 2000 # ============================================================ # security.py # ============================================================ class TestNormalizeAllowEntry: def test_pure_user_id(self): assert normalize_allow_entry("abc123") == "abc123" def test_at_username(self): assert normalize_allow_entry("@bob") == "@bob" def test_user_prefix(self): assert normalize_allow_entry("user:xyz") == "xyz" def test_mattermost_prefix(self): assert normalize_allow_entry("mattermost:abc") == "abc" def test_channel_prefix_stripped(self): assert normalize_allow_entry("channel:xyz") == "xyz" def test_empty_string(self): assert normalize_allow_entry("") == "" def test_whitespace_only(self): assert normalize_allow_entry(" ") == "" def test_none(self): assert normalize_allow_entry(None) == "" def test_strip_whitespace(self): assert normalize_allow_entry(" abc ") == "abc" class TestIsAllowEntryMatch: def test_exact_id_match(self): assert is_allow_entry_match("abc123", "abc123") is True def test_case_insensitive_id_match(self): assert is_allow_entry_match("ABC123", "abc123") is True def test_id_no_match(self): assert is_allow_entry_match("abc123", "xyz789") is False def test_at_username_match(self): assert is_allow_entry_match("@bob", "any_id", "bob") is True def test_at_username_case_insensitive(self): assert is_allow_entry_match("@Bob", "any_id", "bob") is True def test_at_username_no_match(self): assert is_allow_entry_match("@alice", "any_id", "bob") is False def test_empty_entry(self): assert is_allow_entry_match("", "abc123") is False class TestMattermostSecurityConfig: def test_defaults(self): cfg = MattermostSecurityConfig() assert cfg.dm_policy == DM_POLICY_PAIRING assert cfg.group_policy == GROUP_POLICY_ALLOWLIST assert cfg.allow_from == frozenset() assert cfg.group_allow_from == frozenset() def test_from_config_full(self): config = { "dm_policy": "open", "group_policy": "disabled", "allow_from": ["@bob", "user:alice", "charlie_id"], "group_allow_from": ["group1"], "dangerously_allow_name_matching": True, } cfg = MattermostSecurityConfig.from_config(config) assert cfg.dm_policy == DM_POLICY_OPEN assert cfg.group_policy == GROUP_POLICY_DISABLED assert "@bob" in cfg.allow_from assert "alice" in cfg.allow_from # user: prefix stripped assert "charlie_id" in cfg.allow_from assert cfg.dangerously_allow_name_matching is True def test_from_config_invalid_policy_defaults(self): config = {"dm_policy": "nonsense", "group_policy": "nonsense"} cfg = MattermostSecurityConfig.from_config(config) assert cfg.dm_policy == DM_POLICY_PAIRING assert cfg.group_policy == GROUP_POLICY_ALLOWLIST class TestMattermostSecurity: def test_dm_policy_open_allows(self): sec = MattermostSecurity({"dm_policy": "open"}) result = sec.check_dm("u1") assert result.allowed is True def test_dm_policy_disabled_blocks(self): sec = MattermostSecurity({"dm_policy": "disabled"}) result = sec.check_dm("u1") assert result.allowed is False def test_dm_policy_pairing_no_manager_blocks(self): sec = MattermostSecurity({"dm_policy": "pairing"}) result = sec.check_dm("u1") assert result.allowed is False def test_dm_policy_pairing_with_manager(self): pm = MattermostPairingManager() pm.check_or_request("u1") pm.approve("u1", pm._pending["u1"].code) sec = MattermostSecurity({"dm_policy": "pairing"}, pairing_manager=pm) result = sec.check_dm("u1") assert result.allowed is True def test_group_policy_open_allows(self): sec = MattermostSecurity({"group_policy": "open"}) result = sec.check_group("u1") assert result.allowed is True def test_group_policy_disabled_blocks(self): sec = MattermostSecurity({"group_policy": "disabled"}) result = sec.check_group("u1") assert result.allowed is False def test_group_allowlist_match(self): sec = MattermostSecurity({"group_policy": "allowlist", "group_allow_from": ["u1"]}) result = sec.check_group("u1") assert result.allowed is True def test_group_allowlist_no_match(self): sec = MattermostSecurity({"group_policy": "allowlist", "group_allow_from": ["u1"]}) result = sec.check_group("u2") assert result.allowed is False def test_check_inbound_direct(self): sec = MattermostSecurity({"dm_policy": "open"}) result = sec.check_inbound("direct", "u1") assert result.allowed is True def test_check_inbound_group(self): sec = MattermostSecurity({"group_policy": "open"}) result = sec.check_inbound("group", "u1") assert result.allowed is True def test_check_inbound_unknown_chat_type(self): sec = MattermostSecurity() result = sec.check_inbound("invalid_type", "u1") assert result.allowed is False def test_check_dm_allowlist_with_name_matching(self): sec = MattermostSecurity( { "dm_policy": "allowlist", "allow_from": ["@bob"], "dangerously_allow_name_matching": True, } ) result = sec.check_dm("any_id", "bob") assert result.allowed is True def test_reload_allow_from(self): sec = MattermostSecurity({"allow_from": ["u1"]}) sec.reload_config("allow_from", ["u1", "u2", "@alice"]) assert len(sec._config.allow_from) == 3 def test_reload_dm_policy(self): sec = MattermostSecurity({"dm_policy": "open"}) sec.reload_config("dm_policy", "disabled") assert sec._config.dm_policy == DM_POLICY_DISABLED def test_reload_group_policy_invalid_ignores(self): sec = MattermostSecurity({"group_policy": "open"}) sec.reload_config("group_policy", "bad_value") assert sec._config.group_policy == GROUP_POLICY_OPEN def test_reload_dangerously_allow_name(self): sec = MattermostSecurity() sec.reload_config("dangerously_allow_name_matching", True) assert sec._config.dangerously_allow_name_matching is True def test_authorize_command_invocation_allowed(self): sec = MattermostSecurity({"allow_from": ["admin1"]}) assert sec.authorize_command_invocation("admin1", "restart") is True def test_authorize_command_invocation_denied(self): sec = MattermostSecurity({"allow_from": ["admin1"]}) assert sec.authorize_command_invocation("not_admin", "restart") is False def test_authorize_command_non_managed(self): sec = MattermostSecurity() assert sec.authorize_command_invocation("anyone", "help") is True def test_resolve_group_require_mention(self): sec = MattermostSecurity({"groups": {"ch1": {"requireMention": False}}}) assert sec.resolve_group_require_mention("ch1") is False assert sec.resolve_group_require_mention("unknown") is None def test_audit_event(self): sec = MattermostSecurity() sec.audit_event("dm_blocked", "u1", "test detail", "ch1") # ============================================================ # secret.py # ============================================================ class TestSecretDescriptor: def test_defaults(self): sd = SecretDescriptor(name="test", description="Test secret") assert sd.name == "test" assert sd.description == "Test secret" assert sd.required is True assert sd.env_var == "" assert sd.config_key == "" class TestGetSecretContract: def test_returns_list(self): result = get_secret_contract() assert isinstance(result, list) assert len(result) >= 3 names = {s.name for s in result} assert "bot_token" in names class TestResolveSecret: def test_from_config(self): config = {"bot_token": "from-config"} result = resolve_secret("bot_token", config) assert result == "from-config" def test_from_env(self, monkeypatch): monkeypatch.setenv("MATTERMOST_BOT_TOKEN", "from-env") result = resolve_secret("bot_token", {}) assert result == "from-env" def test_config_takes_precedence(self, monkeypatch): monkeypatch.setenv("MATTERMOST_BOT_TOKEN", "from-env") result = resolve_secret("bot_token", {"bot_token": "from-config"}) assert result == "from-config" def test_unknown_secret(self): result = resolve_secret("nonexistent", {}) assert result is None def test_no_value_returns_none(self): result = resolve_secret("bot_token", {}) assert result is None class TestListConfiguredSecrets: def test_all_missing(self): result = list_configured_secrets({}) for v in result.values(): assert v == "missing" def test_from_config(self): result = list_configured_secrets({"bot_token": "x"}) assert result["bot_token"] == "config" def test_from_env(self, monkeypatch): monkeypatch.setenv("MATTERMOST_BOT_TOKEN", "env-token") result = list_configured_secrets({}) assert result["bot_token"] == "env" class TestBuildSecretTargetRegistryEntries: def test_entries(self): entries = build_secret_target_registry_entries() assert len(entries) >= 3 for entry in entries: assert "channel_id" in entry assert "secret_name" in entry assert entry["channel_id"] == "mattermost" class TestCollectRuntimeConfigAssignments: def test_bot_token_from_config(self): config = {"bot_token": "test-token"} assignments = collect_runtime_config_assignments(config) bt = [a for a in assignments if a["secret_name"] == "bot_token"] assert len(bt) == 1 assert bt[0]["source"] == "config" # ============================================================ # slash.py # ============================================================ class TestSlashCommand: def test_defaults(self): cmd = SlashCommand(command="/test", description="A test command") assert cmd.command == "/test" assert cmd.description == "A test command" assert cmd.auto_complete is True assert cmd.hint == "" assert cmd.auto_complete_desc == "" assert cmd.auto_complete_hint == "" class TestDefaultCommands: def test_default_commands_present(self): triggers = {c.command for c in DEFAULT_COMMANDS} assert "/forcepilot" in triggers assert "/fp" in triggers assert "/model" in triggers assert "/clear" in triggers assert "/help" in triggers class TestSlashCommandConfig: def test_default_uses_default_commands(self): cfg = SlashCommandConfig() assert len(cfg.commands) == len(DEFAULT_COMMANDS) def test_from_config_with_custom_dict_commands(self): config = { "commands": [ {"command": "/mycmd", "description": "My command", "auto_complete": False}, ], "auto_register": False, } cfg = SlashCommandConfig.from_config(config) assert len(cfg.commands) == 1 assert cfg.commands[0].command == "/mycmd" assert cfg.commands[0].auto_complete is False def test_from_config_with_string_commands(self): config = {"commands": ["/cmd1", "/cmd2"]} cfg = SlashCommandConfig.from_config(config) assert len(cfg.commands) == 2 assert cfg.commands[0].command == "/cmd1" def test_from_config_empty_commands_uses_defaults(self): cfg = SlashCommandConfig.from_config({}) assert len(cfg.commands) == len(DEFAULT_COMMANDS) class TestBuildCommandPayload: def test_basic(self): cmd = SlashCommand(command="/test", description="Test desc", hint="[args]") payload = build_command_payload(cmd, "team1", "https://cb.example.com") assert payload["team_id"] == "team1" assert payload["trigger"] == "test" assert payload["url"] == "https://cb.example.com" assert payload["hint"] == "[args]" def test_without_hint(self): cmd = SlashCommand(command="/simple", description="Simple") payload = build_command_payload(cmd, "t1", "") assert "hint" not in payload class TestGetSupportedCommands: def test_with_config(self): cfg = SlashCommandConfig(commands=[SlashCommand(command="/custom", description="Custom")]) result = get_supported_commands(cfg) assert len(result) == 1 assert result[0].command == "/custom" def test_without_config(self): result = get_supported_commands() assert len(result) == len(DEFAULT_COMMANDS) class TestBuildSkillCommands: def test_multiple_skills(self): skills = ["skill1", "skill2"] cmds = build_skill_commands(skills) assert len(cmds) == 2 assert cmds[0].command == "/oc_skill1" assert cmds[1].command == "/oc_skill2" def test_empty_skills(self): cmds = build_skill_commands([]) assert cmds == [] class TestManagedSlashCommand: def test_defaults(self): cmd = ManagedSlashCommand(trigger="test", description="Test") assert cmd.trigger == "test" assert cmd.managed is True assert cmd.method == "POST" # ============================================================ # approval.py # ============================================================ class TestApprovalRequest: def test_expired_check(self): req = ApprovalRequest( request_id="r1", action="restart", description="Restarting...", params={}, created_at=time.monotonic() - APPROVAL_TIMEOUT_S - 1, ) assert req.expired is True def test_not_expired(self): req = ApprovalRequest( request_id="r1", action="restart", description="Restarting...", params={}, created_at=time.monotonic(), ) assert req.expired is False class TestApprovalConfig: def test_defaults(self): cfg = ApprovalConfig() assert cfg.enabled is False assert cfg.required_for == ["restart", "sudo", "exec", "delete"] def test_from_config(self): config = {"approval_enabled": True, "approval_timeout_s": 120, "approval_secret": "secret-key"} cfg = ApprovalConfig.from_config(config) assert cfg.enabled is True assert cfg.approval_timeout_s == 120 assert cfg.approval_secret == "secret-key" class TestApprovalManager: def test_enabled_property(self): mgr = ApprovalManager({"approval_enabled": True}) assert mgr.enabled is True def test_requires_approval_when_disabled(self): mgr = ApprovalManager() assert mgr.requires_approval("restart") is False def test_requires_approval_when_enabled(self): mgr = ApprovalManager({"approval_enabled": True}) assert mgr.requires_approval("restart") is True assert mgr.requires_approval("help") is False def test_create_and_approve_request(self): mgr = ApprovalManager() req = mgr.create_request("restart", "Restart the bot", {}, from_user_id="u1") assert req.status == "pending" assert mgr.approve(req.request_id) is True assert req.status == "approved" def test_create_and_deny_request(self): mgr = ApprovalManager() req = mgr.create_request("restart", "Restart", {}) assert mgr.deny(req.request_id) is True assert req.status == "denied" def test_approve_nonexistent(self): mgr = ApprovalManager() assert mgr.approve("nonexistent") is False def test_get_request_found(self): mgr = ApprovalManager() req = mgr.create_request("exec", "Run command", {"cmd": "ls"}) found = mgr.get_request(req.request_id) assert found.request_id == req.request_id def test_get_request_not_found(self): mgr = ApprovalManager() assert mgr.get_request("nonexistent") is None def test_build_approval_message(self): mgr = ApprovalManager() req = mgr.create_request("exec", "Run ls", {"cmd": "ls"}, from_user_id="u1") msg = mgr.build_approval_message(req) assert "需要审批" in msg or "approve" in msg.casefold() assert req.request_id in msg def test_verify_hmac_valid(self): import hmac as hmac_module mgr = ApprovalManager({"approval_secret": "my-secret"}) payload = b"test-payload" expected = hmac_module.new(b"my-secret", payload, hashlib.sha256).hexdigest() assert mgr.verify_hmac(payload, expected) is True def test_verify_hmac_invalid(self): mgr = ApprovalManager({"approval_secret": "my-secret"}) assert mgr.verify_hmac(b"test", "bad-signature") is False def test_verify_hmac_missing_secret(self): mgr = ApprovalManager() assert mgr.verify_hmac(b"test", "sig") is False def test_approve_expired_request(self): mgr = ApprovalManager() req = mgr.create_request("restart", "Old request", {}) req.created_at = time.monotonic() - APPROVAL_TIMEOUT_S - 1 assert mgr.approve(req.request_id) is False # ============================================================ # config_writes.py # ============================================================ class TestConfigWritesManager: def test_disabled_by_default(self): mgr = ConfigWritesManager() assert mgr.is_config_writes_enabled() is False def test_write_disabled_returns_false(self): mgr = ConfigWritesManager() assert mgr.write_config("chatmode", "oncall") is False def test_write_enabled(self): mgr = ConfigWritesManager({"config_writes": True}) assert mgr.is_config_writes_enabled() is True def test_write_allowed_key(self): mgr = ConfigWritesManager({"config_writes": True}) assert mgr.write_config("chatmode", "oncall") is True def test_write_disallowed_key(self): mgr = ConfigWritesManager({"config_writes": True}) assert mgr.write_config("server_url", "evil") is False def test_observer_notification(self): mgr = ConfigWritesManager({"config_writes": True}) notified = [] def observer(key, value): notified.append((key, value)) mgr.register_observer(observer) mgr.write_config("chatmode", "oncall") assert len(notified) == 1 assert notified[0] == ("chatmode", "oncall") def test_observer_error_handling(self): mgr = ConfigWritesManager({"config_writes": True}) def bad_observer(key, value): raise RuntimeError("observer error") mgr.register_observer(bad_observer) assert mgr.write_config("chatmode", "oncall") is True # still succeeds def test_add_to_allowlist(self): mgr = ConfigWritesManager({"config_writes": True, "allow_from": ["u1"]}) assert mgr.add_to_allowlist("u2") is True assert "u2" in mgr._config["allow_from"] def test_add_duplicate_to_allowlist(self): mgr = ConfigWritesManager({"config_writes": True, "allow_from": ["u1"]}) mgr.add_to_allowlist("u1") assert mgr._config["allow_from"].count("u1") == 1 def test_add_to_allowlist_disabled(self): mgr = ConfigWritesManager({"allow_from": ["u1"]}) assert mgr.add_to_allowlist("u2") is False def test_remove_from_allowlist(self): mgr = ConfigWritesManager({"config_writes": True, "allow_from": ["u1", "u2"]}) assert mgr.remove_from_allowlist("u1") is True assert "u1" not in mgr._config["allow_from"] def test_remove_from_allowlist_disabled(self): mgr = ConfigWritesManager({"allow_from": ["u1"]}) assert mgr.remove_from_allowlist("u1") is False def test_get_config(self): mgr = ConfigWritesManager({"config_writes": True, "chatmode": "onmessage", "extra": "not_allowed"}) cfg = mgr.get_config() assert "chatmode" in cfg assert "extra" not in cfg # ============================================================ # pairing.py # ============================================================ class TestPairingRequest: def test_expired(self): req = PairingRequest(user_id="u1", code="ABC123", created_at=time.monotonic() - PAIRING_REQUEST_TTL_S - 1) assert req.expired is True def test_not_expired(self): req = PairingRequest(user_id="u1", code="ABC123") assert req.expired is False class TestMattermostPairingManager: def test_check_or_request_new_user(self): pm = MattermostPairingManager() result = pm.check_or_request("u1") assert result.needs_pairing is True assert len(result.code) == PAIRING_CODE_LENGTH def test_check_or_request_twice_same_code(self): pm = MattermostPairingManager() r1 = pm.check_or_request("u1") r2 = pm.check_or_request("u1") assert r2.needs_pairing is True assert r2.code == r1.code def test_approve_success(self): pm = MattermostPairingManager() result = pm.check_or_request("u1") assert pm.approve("u1", result.code) is True def test_approve_wrong_code(self): pm = MattermostPairingManager() pm.check_or_request("u1") assert pm.approve("u1", "WRONG") is False def test_approve_nonexistent_user(self): pm = MattermostPairingManager() assert pm.approve("nonexistent", "ABC123") is False def test_deny(self): pm = MattermostPairingManager() pm.check_or_request("u1") assert pm.deny("u1") is True def test_is_approved_after_approve(self): pm = MattermostPairingManager() result = pm.check_or_request("u1") pm.approve("u1", result.code) assert pm.is_approved("u1") is True def test_is_approved_false_for_new(self): pm = MattermostPairingManager() assert pm.is_approved("u1") is False def test_list_pending(self): pm = MattermostPairingManager() pm.check_or_request("u1") pm.check_or_request("u2") pending = pm.list_pending() assert len(pending) == 2 def test_list_approved(self): pm = MattermostPairingManager() r = pm.check_or_request("u1") pm.approve("u1", r.code) approved = pm.list_approved() assert len(approved) == 1 def test_cleanup_expired(self): pm = MattermostPairingManager() pm.check_or_request("u1") pm._pending["u1"].created_at = time.monotonic() - PAIRING_REQUEST_TTL_S - 1 pm._cleanup_expired() assert "u1" not in pm._pending class TestBuildPairingNotification: def test_includes_user_id_and_code(self): msg = build_pairing_notification("user123", "ABC123") assert "user123" in msg assert "ABC123" in msg # ============================================================ # send.py # ============================================================ class TestBuildPostOptions: def test_basic(self): from yuxi.channels.models import ChannelIdentity, ChannelResponse, ChannelType identity = ChannelIdentity( channel_id="mm", channel_type=ChannelType.MATTERMOST, channel_user_id="u1", channel_chat_id="ch1", ) response = ChannelResponse(identity=identity, content="hello") opts = build_post_options(response) assert opts["channel_id"] == "ch1" assert opts["message"] == "hello" assert opts["root_id"] == "" def test_with_reply_to(self): from yuxi.channels.models import ChannelIdentity, ChannelResponse, ChannelType identity = ChannelIdentity( channel_id="mm", channel_type=ChannelType.MATTERMOST, channel_user_id="u1", channel_chat_id="ch1", ) response = ChannelResponse(identity=identity, content="reply", reply_to_message_id="root1") opts = build_post_options(response) assert opts["root_id"] == "root1" def test_with_props_and_files(self): from yuxi.channels.models import ChannelIdentity, ChannelResponse, ChannelType identity = ChannelIdentity( channel_id="mm", channel_type=ChannelType.MATTERMOST, channel_user_id="u1", channel_chat_id="ch1", ) response = ChannelResponse( identity=identity, content="multi", metadata={"props": {"k": "v"}, "file_ids": ["f1", "f2"]}, ) opts = build_post_options(response) assert opts["props"] == {"k": "v"} assert opts["file_ids"] == ["f1", "f2"] class TestBuildPatchOptions: def test_sets_message(self): opts = build_patch_options("updated message") assert opts["message"] == "updated message" class TestChunkTextForOutbound: def test_short_text_no_chunking(self): result = chunk_text_for_outbound("hello world", 100) assert result == ["hello world"] def test_text_at_limit_no_chunking(self): text = "x" * 100 result = chunk_text_for_outbound(text, 100) assert len(result) == 1 assert result[0] == text def test_paragraph_split(self): text = "para1\n\npara2\n\npara3" result = chunk_text_for_outbound(text, 10) assert len(result) > 1 def test_single_long_paragraph(self): text = "abcdefghij\nabcdefghij\nabcdefghij" result = chunk_text_for_outbound(text, 12) assert len(result) >= 1 def test_very_long_single_line(self): text = "x" * 500 result = chunk_text_for_outbound(text, 100) assert len(result) >= 5 # all chunks <= limit for chunk in result: assert len(chunk) <= 100 def test_always_returns_at_least_one_chunk(self): result = chunk_text_for_outbound("test", 1) assert len(result) >= 1 def test_empty_text(self): result = chunk_text_for_outbound("", 100) assert result == [""] class TestSplitHard: def test_exact_multiple(self): result = _split_hard("abcdef", 3) assert result == ["abc", "def"] def test_partial_trailing(self): result = _split_hard("abcde", 3) assert result == ["abc", "de"] def test_single_chunk(self): result = _split_hard("ab", 10) assert result == ["ab"] # ============================================================ # reply.py # ============================================================ class TestReplyConfig: def test_defaults(self): cfg = ReplyConfig() assert cfg.mode == REPLY_OFF assert cfg.thread_only is False def test_from_config(self): cfg = ReplyConfig.from_config({"reply_to_mode": "all", "reply_thread_only": True}) assert cfg.mode == REPLY_ALL assert cfg.thread_only is True def test_from_config_invalid_mode_falls_back(self): cfg = ReplyConfig.from_config({"reply_to_mode": "invalid"}) assert cfg.mode == REPLY_OFF class TestReplyManager: def test_default_mode_is_off(self): mgr = ReplyManager() assert mgr.mode == REPLY_OFF def test_should_reply_off(self): mgr = ReplyManager({"reply_to_mode": "off"}) assert mgr.should_reply("group") is False def test_should_reply_all(self): mgr = ReplyManager({"reply_to_mode": "all"}) assert mgr.should_reply("group") is True assert mgr.should_reply("direct") is False def test_should_reply_batched(self): mgr = ReplyManager({"reply_to_mode": "batched"}) assert mgr.should_reply("group") is True def test_resolve_reply_target_off(self): mgr = ReplyManager({"reply_to_mode": "off"}) assert mgr.resolve_reply_target("ch1", "root1") is None def test_resolve_reply_target_all_with_root(self): mgr = ReplyManager({"reply_to_mode": "all"}) assert mgr.resolve_reply_target("ch1", "root1") == "root1" def test_resolve_reply_target_all_without_root(self): mgr = ReplyManager({"reply_to_mode": "all"}) assert mgr.resolve_reply_target("ch1", None) is None # ============================================================ # target_resolution.py # ============================================================ class TestParseTarget: def test_channel_prefix(self): result = parse_target("channel:abc123") assert result.target_type == "channel" assert result.target_id == "abc123" def test_hash_channel(self): result = parse_target("#general") assert result.target_type == "channel" assert result.target_id == "general" def test_channel_name_prefix(self): result = parse_target("channel-name:general") assert result.target_type == "channel" assert result.target_id == "general" def test_user_prefix(self): result = parse_target("user:abc123") assert result.target_type == "user" assert result.target_id == "abc123" def test_at_username(self): result = parse_target("@bob") assert result.target_type == "user" assert result.target_id == "bob" def test_nick_host_user(self): result = parse_target("alice!~alice@server.com") assert result.target_type == "user" assert result.target_id == "alice" def test_mattermost_prefix(self): result = parse_target("mattermost:user:abc123") assert result.target_type == "user" assert result.target_id == "abc123" def test_dm_channel_id(self): dm_id = "abcdefghij0123456789abcdef__abcdefghij0123456789abcdef" result = parse_target(dm_id) assert result.target_type == "channel" def test_26_char_alnum_channel(self): result = parse_target("a" * 26) assert result.target_type == "channel" def test_empty_input(self): result = parse_target("") assert result.target_type == "unknown" def test_none_input(self): result = parse_target(None) assert result.target_type == "unknown" def test_unknown(self): result = parse_target("random-text") assert result.target_type == "unknown" def test_whitespace_only(self): result = parse_target(" ") assert result.target_type == "unknown" def test_is_channel_property(self): result = parse_target("channel:ch1") assert result.is_channel is True assert result.is_user is False def test_is_user_property(self): result = parse_target("user:u1") assert result.is_channel is False assert result.is_user is True class TestLooksLikeMattermostTargetId: def test_26_char_alnum(self): assert looks_like_mattermost_target_id("a" * 26) is True def test_dm_channel_id(self): dm = "abcdefghij0123456789abcdef__abcdefghij0123456789abcdef" assert looks_like_mattermost_target_id(dm) is True def test_channel_prefix(self): assert looks_like_mattermost_target_id("channel:abc") is True def test_user_prefix(self): assert looks_like_mattermost_target_id("user:abc") is True def test_channel_name(self): assert looks_like_mattermost_target_id("channel-name:general") is True def test_random_string(self): assert looks_like_mattermost_target_id("random") is False def test_empty(self): assert looks_like_mattermost_target_id("") is False assert looks_like_mattermost_target_id(None) is False class TestResolveChannelTarget: def test_channel_prefix(self): result = resolve_channel_target("channel:ch1") assert result == "ch1" def test_with_known_channels(self): result = resolve_channel_target("general", {"ch1": "general", "ch2": "random"}) assert result == "ch1" def test_channel_name_format(self): result = resolve_channel_target("channel-name:general", {}) assert result == "general" def test_no_match(self): result = resolve_channel_target("unknown", {}) assert result == "unknown" def test_empty(self): assert resolve_channel_target("") is None class TestResolveUserTarget: def test_user_prefix(self): result = resolve_user_target("user:u1") assert result == "u1" def test_with_known_users(self): result = resolve_user_target("@bob", {"user1": "bob", "user2": "alice"}) assert result == "user1" def test_empty(self): assert resolve_user_target("") is None # ============================================================ # monitor.py # ============================================================ class TestMentionGateConfig: def test_defaults(self): cfg = MentionGateConfig() assert cfg.chat_mode == MODE_ONMESSAGE assert cfg.require_mention is True assert len(cfg.onchar_prefixes) > 0 def test_from_config_with_undefined_chatmode(self): cfg = MentionGateConfig.from_config({"chatmode": "undefined"}) assert cfg.chat_mode == MODE_ONMESSAGE def test_from_config_custom_prefixes(self): cfg = MentionGateConfig.from_config({"onchar_prefixes": ["$", "%"]}) assert "$" in cfg.onchar_prefixes assert "%" in cfg.onchar_prefixes def test_from_config_empty_prefixes(self): cfg = MentionGateConfig.from_config({"onchar_prefixes": []}) assert len(cfg.onchar_prefixes) > 0 class TestMentionGate: def test_dm_onmessage_allows(self): mg = MentionGate() result = mg.check("direct", "hello", False) assert result.should_respond is True def test_dm_onchar_no_prefix_blocks(self): mg = MentionGate({"chatmode": "onchar"}) result = mg.check("direct", "hello", False) assert result.should_respond is False def test_dm_onchar_with_prefix_allows(self): mg = MentionGate({"chatmode": "onchar", "onchar_prefixes": [">"]}) result = mg.check("direct", ">help", False) assert result.should_respond is True def test_group_onmessage_without_mention_blocks(self): mg = MentionGate({"require_mention": True}) result = mg.check("group", "hello", False) assert result.should_respond is False def test_group_onmessage_with_mention_allows(self): mg = MentionGate({"require_mention": True}) result = mg.check("group", "@bot hello", True) assert result.should_respond is True def test_group_oncall_blocks(self): mg = MentionGate({"chatmode": "oncall"}) result = mg.check("group", "hello", True) assert result.should_respond is False def test_group_onchar_without_prefix_blocks(self): mg = MentionGate({"chatmode": "onchar", "onchar_prefixes": [">"]}) result = mg.check("group", "hello", True) assert result.should_respond is False def test_group_onchar_with_prefix_allows(self): mg = MentionGate({"chatmode": "onchar", "onchar_prefixes": [">"]}) result = mg.check("group", "> hello", False) assert result.should_respond is True def test_reload_chatmode(self): mg = MentionGate({"chatmode": "onmessage"}) mg.reload_config("chatmode", "onchar") assert mg.chat_mode == MODE_ONCHAR def test_reload_invalid_chatmode_ignored(self): mg = MentionGate({"chatmode": "onmessage"}) mg.reload_config("chatmode", "invalid") assert mg.chat_mode == MODE_ONMESSAGE def test_reload_require_mention(self): mg = MentionGate({"require_mention": True}) mg.reload_config("require_mention", False) assert mg.require_mention is False def test_reload_onchar_prefixes(self): mg = MentionGate({"onchar_prefixes": [">"]}) mg.reload_config("onchar_prefixes", ["$", "%"]) assert "$" in mg._config.onchar_prefixes class TestIsControlCommand: def test_restart(self): assert is_control_command("/restart") is True def test_sudo(self): assert is_control_command("/sudo rm -rf /") is True def test_exec(self): assert is_control_command("/exec cmd") is True def test_delete(self): assert is_control_command("/delete something") is True def test_config(self): assert is_control_command("/config set key=val") is True def test_approve(self): assert is_control_command("/approve req123") is True def test_deny(self): assert is_control_command("/deny req123") is True def test_help_not_control(self): assert is_control_command("/help") is False def test_whitespace_padded(self): assert is_control_command(" /restart ") is True def test_empty(self): assert is_control_command("") is False class TestResolveControlCommandGate: def test_control_command_in_dm(self): assert resolve_control_command_gate("/restart", "direct") is True def test_control_command_in_group_mentioned(self): assert resolve_control_command_gate("/restart", "group", True) is True def test_control_command_in_group_not_mentioned(self): assert resolve_control_command_gate("/restart", "group", False) is False def test_non_control_command(self): assert resolve_control_command_gate("/help", "direct") is False # ============================================================ # model_picker.py # ============================================================ class TestModelOption: def test_button_id_deterministic(self): m = ModelOption("gpt-4o", "GPT-4o", "openai") bid = m.button_id assert len(bid) == 8 def test_button_id_same_for_same_model(self): m1 = ModelOption("gpt-4o", "GPT-4o", "openai") m2 = ModelOption("gpt-4o", "GPT-4o", "openai") assert m1.button_id == m2.button_id def test_button_id_different_for_different_model(self): m1 = ModelOption("gpt-4o", "GPT-4o") m2 = ModelOption("gpt-4o-mini", "GPT-4o Mini") assert m1.button_id != m2.button_id class TestGetDefaultModelOptions: def test_returns_list(self): models = get_default_model_options() assert len(models) >= 4 for m in models: assert m.model_id assert m.display_name class TestGetAvailableProviders: def test_default_models(self): providers = get_available_providers() assert "openai" in providers def test_empty_models(self): providers = get_available_providers([]) assert providers == [] def test_deduplicates(self): models = [ ModelOption("a", "A", "p1"), ModelOption("b", "B", "p1"), ModelOption("c", "C", "p2"), ] providers = get_available_providers(models) assert providers == ["p1", "p2"] class TestBuildModelPickerActions: def test_provider_state(self): actions = build_model_picker_actions(state=PickerState.PROVIDERS) # should have provider buttons + view all assert len(actions) >= 2 def test_list_state_single_page(self): models = get_default_model_options() actions = build_model_picker_actions(models, state=PickerState.LIST, page=0) # should have model buttons + back button assert any(a["id"] == "back_to_providers" for a in actions) def test_list_state_with_pagination(self): many_models = [ModelOption(f"m{i}", f"M{i}", "p") for i in range(MODEL_PICKER_PAGE_SIZE + 2)] actions = build_model_picker_actions(many_models, state=PickerState.LIST, page=0) assert any(a["id"] == "next_page" for a in actions) assert not any(a["id"] == "prev_page" for a in actions) def test_prev_page_present_on_later_page(self): many_models = [ModelOption(f"m{i}", f"M{i}", "p") for i in range(MODEL_PICKER_PAGE_SIZE + 2)] actions = build_model_picker_actions(many_models, state=PickerState.LIST, page=1) assert any(a["id"] == "prev_page" for a in actions) assert not any(a["id"] == "next_page" for a in actions) class TestBuildModelPickerAttachment: def test_provider_state(self): attachment = build_model_picker_attachment(state=PickerState.PROVIDERS) assert "模型选择" in attachment["title"] assert len(attachment["actions"]) > 0 def test_list_state(self): attachment = build_model_picker_attachment(state=PickerState.LIST, page=0) assert "模型" in attachment["title"] def test_with_provider_filter(self): attachment = build_model_picker_attachment(provider_filter="openai") assert "模型" in attachment["title"] def test_current_model_marker(self): attachment = build_model_picker_attachment( current_model="gpt-4o-mini", state=PickerState.LIST, page=0, ) assert "Mini" in attachment["text"] assert "✅" in attachment["text"] class TestBuildModelSelectAttachment: def test_openai(self): attachment = build_model_select_attachment("openai") assert "模型" in attachment["title"] assert len(attachment["actions"]) > 0 # ============================================================ # session.py # ============================================================ class TestResolveChatId: def test_direct_message(self): post = {"channel_id": "ch1", "user_id": "u1"} channel_data = {"type": "D"} result = resolve_chat_id(post, channel_data) assert result == "dm_u1" def test_group_channel(self): post = {"channel_id": "ch1", "user_id": "u1"} channel_data = {"type": "O"} result = resolve_chat_id(post, channel_data) assert result == "channel_ch1" def test_thread_message(self): post = {"channel_id": "ch1", "user_id": "u1", "root_id": "root1"} channel_data = {"type": "O"} result = resolve_chat_id(post, channel_data) assert result == "channel_ch1:thread_root1" class TestResolveChatType: def test_direct(self): post = {"channel_id": "ch1"} channel_data = {"type": "D"} result = resolve_chat_type(post, channel_data) assert result == ChatType.DIRECT def test_group(self): post = {"channel_id": "ch1"} channel_data = {"type": "O"} result = resolve_chat_type(post, channel_data) assert result == ChatType.GROUP def test_thread(self): post = {"channel_id": "ch1", "root_id": "r1"} channel_data = {"type": "D"} result = resolve_chat_type(post, channel_data) assert result == ChatType.THREAD class TestBuildThreadId: def test_new_style_without_root(self): result = build_thread_id("ch1", None) assert result == "channel_ch1" def test_new_style_with_root(self): result = build_thread_id("ch1", "root1") assert result == "channel_ch1:thread_root1" def test_old_style_direct(self): result = build_thread_id(ChatType.DIRECT, user_id="u1", agent_id="main") assert "dm" in result assert "u1" in result def test_old_style_group(self): result = build_thread_id(ChatType.GROUP, channel_id="ch1", agent_id="main") assert "ch1" in result def test_old_style_thread(self): result = build_thread_id(ChatType.THREAD, channel_id="ch1", agent_id="main") assert "ch1" in result # ============================================================ # directory.py # ============================================================ class TestDirectoryPeer: def test_display_name_fallback_to_username(self): peer = DirectoryPeer(user_id="u1", username="bob") assert peer.display_name == "bob" def test_display_name_prefers_nickname(self): peer = DirectoryPeer(user_id="u1", username="bob", nickname="Bobby") assert peer.display_name == "Bobby" def test_opaque_id(self): peer = DirectoryPeer(user_id="u1", username="b") assert peer.opaque_id == "user:u1" class TestDirectoryGroup: def test_defaults(self): group = DirectoryGroup(channel_id="ch1", name="general") assert group.group_type == "O" assert group.member_count == 0 def test_opaque_id(self): group = DirectoryGroup(channel_id="ch1", name="g") assert group.opaque_id == "channel:ch1" class TestDirectorySnapshot: def test_peer_count(self): snap = DirectorySnapshot() assert snap.peer_count == 0 def test_group_count(self): snap = DirectorySnapshot(groups=[DirectoryGroup("ch1", "g")]) assert snap.group_count == 1 def test_find_peer(self): p = DirectoryPeer("u1", "bob") snap = DirectorySnapshot(peers=[p]) assert snap.find_peer("u1") == p assert snap.find_peer("u2") is None def test_find_group(self): g = DirectoryGroup("ch1", "general") snap = DirectorySnapshot(groups=[g]) assert snap.find_group("ch1") == g assert snap.find_group("ch2") is None def test_list_peers_by_name(self): p1 = DirectoryPeer("u1", "bob", "Bobby") p2 = DirectoryPeer("u2", "alice") snap = DirectorySnapshot(peers=[p1, p2]) results = snap.list_peers_by_name("bob") assert len(results) == 1 assert results[0].user_id == "u1" def test_list_groups_by_name(self): g1 = DirectoryGroup("ch1", "general", "General") g2 = DirectoryGroup("ch2", "random") snap = DirectorySnapshot(groups=[g1, g2]) results = snap.list_groups_by_name("gen") assert len(results) == 1 assert results[0].channel_id == "ch1" class TestBuildPeerFromUserData: def test_full_data(self): user = { "id": "u1", "username": "bob", "nickname": "Bobby", "email": "bob@test.com", "first_name": "Bob", "last_name": "Smith", } peer = build_peer_from_user_data(user) assert peer.user_id == "u1" assert peer.username == "bob" assert peer.nickname == "Bobby" assert peer.email == "bob@test.com" assert peer.first_name == "Bob" assert peer.last_name == "Smith" def test_empty_data(self): peer = build_peer_from_user_data({}) assert peer.user_id == "" class TestBuildGroupFromChannelData: def test_full_data(self): ch = {"id": "ch1", "name": "general", "display_name": "General", "type": "P"} group = build_group_from_channel_data(ch) assert group.channel_id == "ch1" assert group.name == "general" assert group.group_type == "P" def test_empty_data(self): group = build_group_from_channel_data({}) assert group.channel_id == ""