mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-05-03 02:11:48 +00:00
e21898ea98
Proves token A's detected capabilities do not leak to token B after the fix in the preceding commit. Before the fix this test would have seen both tokens return token A's cached value.
1119 lines
48 KiB
Python
1119 lines
48 KiB
Python
"""Tests for the Discord server introspection and management tool."""
|
|
|
|
import json
|
|
import os
|
|
import urllib.error
|
|
from io import BytesIO
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from tools.discord_tool import (
|
|
DiscordAPIError,
|
|
_ACTIONS,
|
|
_ADMIN_ACTIONS,
|
|
_CORE_ACTIONS,
|
|
_available_actions,
|
|
_build_schema,
|
|
_channel_type_name,
|
|
_detect_capabilities,
|
|
_discord_request,
|
|
_enrich_403,
|
|
_get_bot_token,
|
|
_load_allowed_actions_config,
|
|
_reset_capability_cache,
|
|
check_discord_tool_requirements,
|
|
discord_admin_handler,
|
|
discord_core,
|
|
get_dynamic_schema,
|
|
get_dynamic_schema_admin,
|
|
get_dynamic_schema_core,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _mock_urlopen(response_data, status=200):
|
|
"""Create a mock for urllib.request.urlopen."""
|
|
mock_resp = MagicMock()
|
|
mock_resp.status = status
|
|
mock_resp.read.return_value = json.dumps(response_data).encode("utf-8")
|
|
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
|
|
mock_resp.__exit__ = MagicMock(return_value=False)
|
|
return mock_resp
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Token / check_fn
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCheckRequirements:
|
|
def test_no_token(self, monkeypatch):
|
|
monkeypatch.delenv("DISCORD_BOT_TOKEN", raising=False)
|
|
assert check_discord_tool_requirements() is False
|
|
|
|
def test_empty_token(self, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "")
|
|
assert check_discord_tool_requirements() is False
|
|
|
|
def test_valid_token(self, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token-123")
|
|
assert check_discord_tool_requirements() is True
|
|
|
|
def test_get_bot_token(self, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", " my-token ")
|
|
assert _get_bot_token() == "my-token"
|
|
|
|
def test_get_bot_token_missing(self, monkeypatch):
|
|
monkeypatch.delenv("DISCORD_BOT_TOKEN", raising=False)
|
|
assert _get_bot_token() is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Channel type names
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestChannelTypeNames:
|
|
def test_known_types(self):
|
|
assert _channel_type_name(0) == "text"
|
|
assert _channel_type_name(2) == "voice"
|
|
assert _channel_type_name(4) == "category"
|
|
assert _channel_type_name(5) == "announcement"
|
|
assert _channel_type_name(13) == "stage"
|
|
assert _channel_type_name(15) == "forum"
|
|
|
|
def test_unknown_type(self):
|
|
assert _channel_type_name(99) == "unknown(99)"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Discord API request helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDiscordRequest:
|
|
@patch("tools.discord_tool.urllib.request.urlopen")
|
|
def test_get_request(self, mock_urlopen_fn):
|
|
mock_urlopen_fn.return_value = _mock_urlopen({"ok": True})
|
|
result = _discord_request("GET", "/test", "token123")
|
|
assert result == {"ok": True}
|
|
|
|
# Verify the request was constructed correctly
|
|
call_args = mock_urlopen_fn.call_args
|
|
req = call_args[0][0]
|
|
assert "https://discord.com/api/v10/test" in req.full_url
|
|
assert req.get_header("Authorization") == "Bot token123"
|
|
assert req.get_method() == "GET"
|
|
|
|
@patch("tools.discord_tool.urllib.request.urlopen")
|
|
def test_get_with_params(self, mock_urlopen_fn):
|
|
mock_urlopen_fn.return_value = _mock_urlopen({"ok": True})
|
|
_discord_request("GET", "/test", "tok", params={"foo": "bar"})
|
|
req = mock_urlopen_fn.call_args[0][0]
|
|
assert "foo=bar" in req.full_url
|
|
|
|
@patch("tools.discord_tool.urllib.request.urlopen")
|
|
def test_post_with_body(self, mock_urlopen_fn):
|
|
mock_urlopen_fn.return_value = _mock_urlopen({"id": "123"})
|
|
result = _discord_request("POST", "/channels", "tok", body={"name": "test"})
|
|
assert result == {"id": "123"}
|
|
req = mock_urlopen_fn.call_args[0][0]
|
|
assert req.data == json.dumps({"name": "test"}).encode("utf-8")
|
|
|
|
@patch("tools.discord_tool.urllib.request.urlopen")
|
|
def test_204_returns_none(self, mock_urlopen_fn):
|
|
mock_resp = _mock_urlopen({}, status=204)
|
|
mock_urlopen_fn.return_value = mock_resp
|
|
result = _discord_request("PUT", "/pins/1", "tok")
|
|
assert result is None
|
|
|
|
@patch("tools.discord_tool.urllib.request.urlopen")
|
|
def test_http_error(self, mock_urlopen_fn):
|
|
error_body = json.dumps({"message": "Missing Access"}).encode()
|
|
http_error = urllib.error.HTTPError(
|
|
url="https://discord.com/api/v10/test",
|
|
code=403,
|
|
msg="Forbidden",
|
|
hdrs={},
|
|
fp=BytesIO(error_body),
|
|
)
|
|
mock_urlopen_fn.side_effect = http_error
|
|
with pytest.raises(DiscordAPIError) as exc_info:
|
|
_discord_request("GET", "/test", "tok")
|
|
assert exc_info.value.status == 403
|
|
assert "Missing Access" in exc_info.value.body
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main handler: validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDiscordServerValidation:
|
|
def test_no_token(self, monkeypatch):
|
|
monkeypatch.delenv("DISCORD_BOT_TOKEN", raising=False)
|
|
result = json.loads(discord_admin_handler(action="list_guilds"))
|
|
assert "error" in result
|
|
assert "DISCORD_BOT_TOKEN" in result["error"]
|
|
|
|
def test_unknown_action(self, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
result = json.loads(discord_core(action="bad_action"))
|
|
assert "error" in result
|
|
assert "Unknown action" in result["error"]
|
|
assert "available_actions" in result
|
|
|
|
def test_missing_required_guild_id(self, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
result = json.loads(discord_admin_handler(action="list_channels"))
|
|
assert "error" in result
|
|
assert "guild_id" in result["error"]
|
|
|
|
def test_missing_required_channel_id(self, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
result = json.loads(discord_core(action="fetch_messages"))
|
|
assert "error" in result
|
|
assert "channel_id" in result["error"]
|
|
|
|
def test_missing_multiple_params(self, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
result = json.loads(discord_admin_handler(action="add_role"))
|
|
assert "error" in result
|
|
assert "guild_id" in result["error"]
|
|
assert "user_id" in result["error"]
|
|
assert "role_id" in result["error"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action: list_guilds
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestListGuilds:
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_list_guilds(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = [
|
|
{"id": "111", "name": "Test Server", "icon": "abc", "owner": True, "permissions": "123"},
|
|
{"id": "222", "name": "Other Server", "icon": None, "owner": False, "permissions": "456"},
|
|
]
|
|
result = json.loads(discord_admin_handler(action="list_guilds"))
|
|
assert result["count"] == 2
|
|
assert result["guilds"][0]["name"] == "Test Server"
|
|
assert result["guilds"][1]["id"] == "222"
|
|
mock_req.assert_called_once_with("GET", "/users/@me/guilds", "test-token")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action: server_info
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestServerInfo:
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_server_info(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = {
|
|
"id": "111",
|
|
"name": "My Server",
|
|
"description": "A cool server",
|
|
"icon": "icon_hash",
|
|
"owner_id": "999",
|
|
"approximate_member_count": 42,
|
|
"approximate_presence_count": 10,
|
|
"features": ["COMMUNITY"],
|
|
"premium_tier": 2,
|
|
"premium_subscription_count": 5,
|
|
"verification_level": 1,
|
|
}
|
|
result = json.loads(discord_admin_handler(action="server_info", guild_id="111"))
|
|
assert result["name"] == "My Server"
|
|
assert result["member_count"] == 42
|
|
assert result["online_count"] == 10
|
|
mock_req.assert_called_once_with(
|
|
"GET", "/guilds/111", "test-token", params={"with_counts": "true"}
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action: list_channels
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestListChannels:
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_list_channels_organized(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = [
|
|
{"id": "10", "name": "General", "type": 4, "position": 0, "parent_id": None},
|
|
{"id": "11", "name": "chat", "type": 0, "position": 0, "parent_id": "10", "topic": "Main chat", "nsfw": False},
|
|
{"id": "12", "name": "voice", "type": 2, "position": 1, "parent_id": "10", "topic": None, "nsfw": False},
|
|
{"id": "13", "name": "no-category", "type": 0, "position": 0, "parent_id": None, "topic": None, "nsfw": False},
|
|
]
|
|
result = json.loads(discord_admin_handler(action="list_channels", guild_id="111"))
|
|
assert result["total_channels"] == 3 # excludes the category itself
|
|
groups = result["channel_groups"]
|
|
# Uncategorized first
|
|
assert groups[0]["category"] is None
|
|
assert len(groups[0]["channels"]) == 1
|
|
assert groups[0]["channels"][0]["name"] == "no-category"
|
|
# Then the category
|
|
assert groups[1]["category"]["name"] == "General"
|
|
assert len(groups[1]["channels"]) == 2
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_empty_guild(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = []
|
|
result = json.loads(discord_admin_handler(action="list_channels", guild_id="111"))
|
|
assert result["total_channels"] == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action: channel_info
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestChannelInfo:
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_channel_info(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = {
|
|
"id": "11", "name": "general", "type": 0, "guild_id": "111",
|
|
"topic": "Welcome!", "nsfw": False, "position": 0,
|
|
"parent_id": "10", "rate_limit_per_user": 0, "last_message_id": "999",
|
|
}
|
|
result = json.loads(discord_admin_handler(action="channel_info", channel_id="11"))
|
|
assert result["name"] == "general"
|
|
assert result["type"] == "text"
|
|
assert result["guild_id"] == "111"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action: list_roles
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestListRoles:
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_list_roles_sorted(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = [
|
|
{"id": "1", "name": "@everyone", "position": 0, "color": 0, "mentionable": False, "managed": False, "hoist": False},
|
|
{"id": "2", "name": "Admin", "position": 2, "color": 16711680, "mentionable": True, "managed": False, "hoist": True},
|
|
{"id": "3", "name": "Mod", "position": 1, "color": 255, "mentionable": True, "managed": False, "hoist": True},
|
|
]
|
|
result = json.loads(discord_admin_handler(action="list_roles", guild_id="111"))
|
|
assert result["count"] == 3
|
|
# Should be sorted by position descending
|
|
assert result["roles"][0]["name"] == "Admin"
|
|
assert result["roles"][0]["color"] == "#ff0000"
|
|
assert result["roles"][1]["name"] == "Mod"
|
|
assert result["roles"][2]["name"] == "@everyone"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action: member_info
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMemberInfo:
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_member_info(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = {
|
|
"user": {"id": "42", "username": "testuser", "global_name": "Test User", "avatar": "abc", "bot": False},
|
|
"nick": "Testy",
|
|
"roles": ["2", "3"],
|
|
"joined_at": "2024-01-01T00:00:00Z",
|
|
"premium_since": None,
|
|
}
|
|
result = json.loads(discord_admin_handler(action="member_info", guild_id="111", user_id="42"))
|
|
assert result["username"] == "testuser"
|
|
assert result["nickname"] == "Testy"
|
|
assert result["roles"] == ["2", "3"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action: search_members
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSearchMembers:
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_search_members(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = [
|
|
{"user": {"id": "42", "username": "testuser", "global_name": "Test", "bot": False}, "nick": None, "roles": []},
|
|
]
|
|
result = json.loads(discord_core(action="search_members", guild_id="111", query="test"))
|
|
assert result["count"] == 1
|
|
assert result["members"][0]["username"] == "testuser"
|
|
mock_req.assert_called_once_with(
|
|
"GET", "/guilds/111/members/search", "test-token",
|
|
params={"query": "test", "limit": "50"},
|
|
)
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_search_members_limit_capped(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = []
|
|
discord_core(action="search_members", guild_id="111", query="x", limit=200)
|
|
call_params = mock_req.call_args[1]["params"]
|
|
assert call_params["limit"] == "100" # Capped at 100
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action: fetch_messages
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFetchMessages:
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_fetch_messages(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = [
|
|
{
|
|
"id": "1001",
|
|
"content": "Hello world",
|
|
"author": {"id": "42", "username": "user1", "global_name": "User One", "bot": False},
|
|
"timestamp": "2024-01-01T12:00:00Z",
|
|
"edited_timestamp": None,
|
|
"attachments": [],
|
|
"pinned": False,
|
|
},
|
|
]
|
|
result = json.loads(discord_core(action="fetch_messages", channel_id="11"))
|
|
assert result["count"] == 1
|
|
assert result["messages"][0]["content"] == "Hello world"
|
|
assert result["messages"][0]["author"]["username"] == "user1"
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_fetch_messages_with_pagination(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = []
|
|
discord_core(action="fetch_messages", channel_id="11", before="999", limit=10)
|
|
call_params = mock_req.call_args[1]["params"]
|
|
assert call_params["before"] == "999"
|
|
assert call_params["limit"] == "10"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action: list_pins
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestListPins:
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_list_pins(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = [
|
|
{"id": "500", "content": "Important announcement", "author": {"username": "admin"}, "timestamp": "2024-01-01T00:00:00Z"},
|
|
]
|
|
result = json.loads(discord_admin_handler(action="list_pins", channel_id="11"))
|
|
assert result["count"] == 1
|
|
assert result["pinned_messages"][0]["content"] == "Important announcement"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Actions: pin_message / unpin_message
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPinUnpin:
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_pin_message(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = None # 204
|
|
result = json.loads(discord_admin_handler(action="pin_message", channel_id="11", message_id="500"))
|
|
assert result["success"] is True
|
|
mock_req.assert_called_once_with("PUT", "/channels/11/pins/500", "test-token")
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_unpin_message(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = None
|
|
result = json.loads(discord_admin_handler(action="unpin_message", channel_id="11", message_id="500"))
|
|
assert result["success"] is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action: create_thread
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCreateThread:
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_create_standalone_thread(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = {"id": "800", "name": "New Thread"}
|
|
result = json.loads(discord_core(action="create_thread", channel_id="11", name="New Thread"))
|
|
assert result["success"] is True
|
|
assert result["thread_id"] == "800"
|
|
# Verify the API call
|
|
mock_req.assert_called_once_with(
|
|
"POST", "/channels/11/threads", "test-token",
|
|
body={"name": "New Thread", "auto_archive_duration": 1440, "type": 11},
|
|
)
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_create_thread_from_message(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = {"id": "801", "name": "Discussion"}
|
|
result = json.loads(discord_core(
|
|
action="create_thread", channel_id="11", name="Discussion", message_id="1001",
|
|
))
|
|
assert result["success"] is True
|
|
mock_req.assert_called_once_with(
|
|
"POST", "/channels/11/messages/1001/threads", "test-token",
|
|
body={"name": "Discussion", "auto_archive_duration": 1440},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Actions: add_role / remove_role
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRoleManagement:
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_add_role(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = None
|
|
result = json.loads(discord_admin_handler(
|
|
action="add_role", guild_id="111", user_id="42", role_id="2",
|
|
))
|
|
assert result["success"] is True
|
|
mock_req.assert_called_once_with(
|
|
"PUT", "/guilds/111/members/42/roles/2", "test-token",
|
|
)
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_remove_role(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.return_value = None
|
|
result = json.loads(discord_admin_handler(
|
|
action="remove_role", guild_id="111", user_id="42", role_id="2",
|
|
))
|
|
assert result["success"] is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Error handling
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestErrorHandling:
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_api_error_handled(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.side_effect = DiscordAPIError(403, '{"message": "Missing Access"}')
|
|
result = json.loads(discord_admin_handler(action="list_guilds"))
|
|
assert "error" in result
|
|
assert "403" in result["error"]
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_unexpected_error_handled_admin(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.side_effect = RuntimeError("something broke")
|
|
result = json.loads(discord_admin_handler(action="list_guilds"))
|
|
assert "error" in result
|
|
assert "something broke" in result["error"]
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_unexpected_error_handled_core(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
|
|
mock_req.side_effect = RuntimeError("something broke")
|
|
result = json.loads(discord_core(action="fetch_messages", channel_id="11"))
|
|
assert "error" in result
|
|
assert "something broke" in result["error"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRegistration:
|
|
def test_core_tool_registered(self):
|
|
from tools.registry import registry
|
|
entry = registry._tools.get("discord")
|
|
assert entry is not None
|
|
assert entry.schema["name"] == "discord"
|
|
assert entry.toolset == "discord"
|
|
assert entry.check_fn is not None
|
|
assert entry.requires_env == ["DISCORD_BOT_TOKEN"]
|
|
|
|
def test_admin_tool_registered(self):
|
|
from tools.registry import registry
|
|
entry = registry._tools.get("discord_admin")
|
|
assert entry is not None
|
|
assert entry.schema["name"] == "discord_admin"
|
|
assert entry.toolset == "discord_admin"
|
|
assert entry.check_fn is not None
|
|
assert entry.requires_env == ["DISCORD_BOT_TOKEN"]
|
|
|
|
def test_core_schema_actions(self):
|
|
"""Core static schema should list only core actions."""
|
|
from tools.registry import registry
|
|
entry = registry._tools["discord"]
|
|
actions = set(entry.schema["parameters"]["properties"]["action"]["enum"])
|
|
assert actions == {"fetch_messages", "search_members", "create_thread"}
|
|
|
|
def test_admin_schema_actions(self):
|
|
"""Admin static schema should list only admin actions."""
|
|
from tools.registry import registry
|
|
entry = registry._tools["discord_admin"]
|
|
actions = set(entry.schema["parameters"]["properties"]["action"]["enum"])
|
|
expected_admin = set(_ACTIONS.keys()) - {"fetch_messages", "search_members", "create_thread"}
|
|
assert actions == expected_admin
|
|
|
|
def test_all_actions_covered(self):
|
|
"""Core + admin actions should cover all known actions."""
|
|
assert set(_CORE_ACTIONS.keys()) | set(_ADMIN_ACTIONS.keys()) == set(_ACTIONS.keys())
|
|
assert set(_CORE_ACTIONS.keys()) & set(_ADMIN_ACTIONS.keys()) == set()
|
|
|
|
def test_schema_parameter_bounds(self):
|
|
from tools.registry import registry
|
|
entry = registry._tools["discord"]
|
|
props = entry.schema["parameters"]["properties"]
|
|
assert props["limit"]["minimum"] == 1
|
|
assert props["limit"]["maximum"] == 100
|
|
assert props["auto_archive_duration"]["enum"] == [60, 1440, 4320, 10080]
|
|
|
|
def test_core_schema_description(self):
|
|
"""Core schema description should mention core actions."""
|
|
from tools.registry import registry
|
|
entry = registry._tools["discord"]
|
|
desc = entry.schema["description"]
|
|
assert "fetch_messages(channel_id)" in desc
|
|
assert "search_members(guild_id, query)" in desc
|
|
assert "create_thread(channel_id, name)" in desc
|
|
# Admin actions should NOT be in core description
|
|
assert "list_guilds()" not in desc
|
|
assert "add_role(" not in desc
|
|
|
|
def test_admin_schema_description(self):
|
|
"""Admin schema description should mention admin actions."""
|
|
from tools.registry import registry
|
|
entry = registry._tools["discord_admin"]
|
|
desc = entry.schema["description"]
|
|
assert "list_guilds()" in desc
|
|
assert "add_role(guild_id, user_id, role_id)" in desc
|
|
# Core actions should NOT be in admin description
|
|
assert "fetch_messages(" not in desc
|
|
assert "create_thread(" not in desc
|
|
|
|
def test_handler_callable(self):
|
|
from tools.registry import registry
|
|
entry = registry._tools["discord"]
|
|
assert callable(entry.handler)
|
|
entry_admin = registry._tools["discord_admin"]
|
|
assert callable(entry_admin.handler)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Toolset: discord / discord_admin only in hermes-discord
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestToolsetInclusion:
|
|
def test_discord_tools_in_hermes_discord_toolset(self):
|
|
from toolsets import TOOLSETS
|
|
assert "discord" in TOOLSETS["hermes-discord"]["tools"]
|
|
assert "discord_admin" in TOOLSETS["hermes-discord"]["tools"]
|
|
|
|
def test_discord_tools_not_in_core_tools(self):
|
|
from toolsets import _HERMES_CORE_TOOLS
|
|
assert "discord" not in _HERMES_CORE_TOOLS
|
|
assert "discord_admin" not in _HERMES_CORE_TOOLS
|
|
|
|
def test_discord_tools_not_in_other_toolsets(self):
|
|
from toolsets import TOOLSETS
|
|
for name, ts in TOOLSETS.items():
|
|
if name in ("hermes-discord", "hermes-gateway", "discord", "discord_admin"):
|
|
continue
|
|
tools = ts.get("tools", [])
|
|
assert "discord" not in tools or name == "discord", (
|
|
f"discord tool should not be in toolset '{name}'"
|
|
)
|
|
assert "discord_admin" not in tools or name == "discord_admin", (
|
|
f"discord_admin tool should not be in toolset '{name}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Capability detection (privileged intents)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCapabilityDetection:
|
|
def setup_method(self):
|
|
_reset_capability_cache()
|
|
|
|
def teardown_method(self):
|
|
_reset_capability_cache()
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_both_intents_enabled(self, mock_req):
|
|
# flags: GUILD_MEMBERS (1<<14) + MESSAGE_CONTENT (1<<18) = 278528
|
|
mock_req.return_value = {"flags": (1 << 14) | (1 << 18)}
|
|
caps = _detect_capabilities("tok")
|
|
assert caps["has_members_intent"] is True
|
|
assert caps["has_message_content"] is True
|
|
assert caps["detected"] is True
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_no_intents(self, mock_req):
|
|
mock_req.return_value = {"flags": 0}
|
|
caps = _detect_capabilities("tok")
|
|
assert caps["has_members_intent"] is False
|
|
assert caps["has_message_content"] is False
|
|
assert caps["detected"] is True
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_limited_intent_variants_counted(self, mock_req):
|
|
# GUILD_MEMBERS_LIMITED (1<<15), MESSAGE_CONTENT_LIMITED (1<<19)
|
|
mock_req.return_value = {"flags": (1 << 15) | (1 << 19)}
|
|
caps = _detect_capabilities("tok")
|
|
assert caps["has_members_intent"] is True
|
|
assert caps["has_message_content"] is True
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_only_members_intent(self, mock_req):
|
|
mock_req.return_value = {"flags": 1 << 14}
|
|
caps = _detect_capabilities("tok")
|
|
assert caps["has_members_intent"] is True
|
|
assert caps["has_message_content"] is False
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_detection_failure_is_permissive(self, mock_req):
|
|
"""If detection fails (network/401/revoked token), expose everything
|
|
and let runtime errors surface. Silent failure should never hide
|
|
actions the bot actually has."""
|
|
mock_req.side_effect = DiscordAPIError(401, "unauthorized")
|
|
caps = _detect_capabilities("tok")
|
|
assert caps["detected"] is False
|
|
assert caps["has_members_intent"] is True
|
|
assert caps["has_message_content"] is True
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_detection_is_cached(self, mock_req):
|
|
mock_req.return_value = {"flags": 0}
|
|
_detect_capabilities("tok")
|
|
_detect_capabilities("tok")
|
|
_detect_capabilities("tok")
|
|
assert mock_req.call_count == 1
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_force_refresh(self, mock_req):
|
|
mock_req.return_value = {"flags": 0}
|
|
_detect_capabilities("tok")
|
|
_detect_capabilities("tok", force=True)
|
|
assert mock_req.call_count == 2
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_cache_is_keyed_by_token(self, mock_req):
|
|
"""Regression: token A's capabilities must not leak to token B.
|
|
|
|
Before the fix, the cache was a single module-global dict. The first
|
|
call populated it and every subsequent call — regardless of token —
|
|
returned the same cached value, producing wrong schema gating for
|
|
rotated or multi-token deployments.
|
|
"""
|
|
def _per_token_flags(method, path, token, **_kwargs):
|
|
# token A: both intents; token B: neither.
|
|
if token == "tok_a":
|
|
return {"flags": (1 << 14) | (1 << 18)}
|
|
return {"flags": 0}
|
|
|
|
mock_req.side_effect = _per_token_flags
|
|
|
|
caps_a = _detect_capabilities("tok_a")
|
|
caps_b = _detect_capabilities("tok_b")
|
|
|
|
assert caps_a["has_members_intent"] is True
|
|
assert caps_a["has_message_content"] is True
|
|
assert caps_b["has_members_intent"] is False
|
|
assert caps_b["has_message_content"] is False
|
|
# Each token should hit the endpoint exactly once.
|
|
assert mock_req.call_count == 2
|
|
|
|
# Re-requesting either token serves from its own cache entry.
|
|
_detect_capabilities("tok_a")
|
|
_detect_capabilities("tok_b")
|
|
assert mock_req.call_count == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config allowlist
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestConfigAllowlist:
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_tools_logger(self):
|
|
"""Restore the ``tools`` logger level after cross-test pollution.
|
|
|
|
``AIAgent(quiet_mode=True)`` globally sets ``tools`` and
|
|
``tools.*`` children to ``ERROR`` (see run_agent.py quiet_mode
|
|
block). xdist workers are persistent, so a streaming test on the
|
|
same worker will silence WARNING-level logs from
|
|
``tools.discord_tool`` for every test that follows. Reset here so
|
|
``caplog`` can capture warnings regardless of worker history.
|
|
"""
|
|
import logging as _logging
|
|
_prev_tools = _logging.getLogger("tools").level
|
|
_prev_dt = _logging.getLogger("tools.discord_tool").level
|
|
_logging.getLogger("tools").setLevel(_logging.NOTSET)
|
|
_logging.getLogger("tools.discord_tool").setLevel(_logging.NOTSET)
|
|
try:
|
|
yield
|
|
finally:
|
|
_logging.getLogger("tools").setLevel(_prev_tools)
|
|
_logging.getLogger("tools.discord_tool").setLevel(_prev_dt)
|
|
|
|
def test_empty_string_returns_none(self, monkeypatch):
|
|
"""Empty config means no allowlist — all actions visible."""
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": ""}},
|
|
)
|
|
assert _load_allowed_actions_config() is None
|
|
|
|
def test_missing_key_returns_none(self, monkeypatch):
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {}},
|
|
)
|
|
assert _load_allowed_actions_config() is None
|
|
|
|
def test_comma_separated_string(self, monkeypatch):
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": "list_guilds,list_channels,fetch_messages"}},
|
|
)
|
|
result = _load_allowed_actions_config()
|
|
assert result == ["list_guilds", "list_channels", "fetch_messages"]
|
|
|
|
def test_yaml_list(self, monkeypatch):
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": ["list_guilds", "server_info"]}},
|
|
)
|
|
result = _load_allowed_actions_config()
|
|
assert result == ["list_guilds", "server_info"]
|
|
|
|
def test_unknown_names_dropped(self, monkeypatch, caplog):
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": "list_guilds,bogus_action,fetch_messages"}},
|
|
)
|
|
with caplog.at_level("WARNING"):
|
|
result = _load_allowed_actions_config()
|
|
assert result == ["list_guilds", "fetch_messages"]
|
|
assert "bogus_action" in caplog.text
|
|
|
|
def test_config_load_failure_is_permissive(self, monkeypatch):
|
|
"""If config can't be loaded at all, fall back to None (all allowed)."""
|
|
def bad_load():
|
|
raise RuntimeError("disk gone")
|
|
monkeypatch.setattr("hermes_cli.config.load_config", bad_load)
|
|
assert _load_allowed_actions_config() is None
|
|
|
|
def test_unexpected_type_ignored(self, monkeypatch, caplog):
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": {"unexpected": "dict"}}},
|
|
)
|
|
with caplog.at_level("WARNING"):
|
|
result = _load_allowed_actions_config()
|
|
assert result is None
|
|
assert "unexpected type" in caplog.text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action filtering combines intents + allowlist
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestAvailableActions:
|
|
def test_all_available_when_unrestricted(self):
|
|
caps = {"detected": True, "has_members_intent": True, "has_message_content": True}
|
|
assert _available_actions(caps, None) == list(_ACTIONS.keys())
|
|
|
|
def test_no_members_intent_hides_member_actions(self):
|
|
caps = {"detected": True, "has_members_intent": False, "has_message_content": True}
|
|
actions = _available_actions(caps, None)
|
|
assert "search_members" not in actions
|
|
assert "member_info" not in actions
|
|
# fetch_messages stays — MESSAGE_CONTENT affects content field but action works
|
|
assert "fetch_messages" in actions
|
|
|
|
def test_no_message_content_keeps_fetch_messages(self):
|
|
"""MESSAGE_CONTENT affects the content field, not the action.
|
|
Hiding fetch_messages would lose author/timestamp/attachments access."""
|
|
caps = {"detected": True, "has_members_intent": True, "has_message_content": False}
|
|
actions = _available_actions(caps, None)
|
|
assert "fetch_messages" in actions
|
|
assert "list_pins" in actions
|
|
|
|
def test_allowlist_intersects_with_intents(self):
|
|
"""Allowlist can only narrow — not re-enable intent-gated actions."""
|
|
caps = {"detected": True, "has_members_intent": False, "has_message_content": True}
|
|
allowlist = ["list_guilds", "search_members", "fetch_messages"]
|
|
actions = _available_actions(caps, allowlist)
|
|
# search_members gated by intent → stripped even though allowlisted
|
|
assert actions == ["list_guilds", "fetch_messages"]
|
|
|
|
def test_empty_allowlist_yields_empty(self):
|
|
caps = {"detected": True, "has_members_intent": True, "has_message_content": True}
|
|
assert _available_actions(caps, []) == []
|
|
|
|
def test_allowlist_preserves_canonical_order(self):
|
|
caps = {"detected": True, "has_members_intent": True, "has_message_content": True}
|
|
# Pass allowlist out of canonical order
|
|
allowlist = ["fetch_messages", "list_guilds", "server_info"]
|
|
assert _available_actions(caps, allowlist) == ["list_guilds", "server_info", "fetch_messages"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dynamic schema build (integration of intents + config)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDynamicSchema:
|
|
def setup_method(self):
|
|
_reset_capability_cache()
|
|
|
|
def teardown_method(self):
|
|
_reset_capability_cache()
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_no_token_returns_none(self, mock_req, monkeypatch):
|
|
monkeypatch.delenv("DISCORD_BOT_TOKEN", raising=False)
|
|
assert get_dynamic_schema_core() is None
|
|
assert get_dynamic_schema_admin() is None
|
|
mock_req.assert_not_called()
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_full_intents_core_schema(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": ""}},
|
|
)
|
|
mock_req.return_value = {"flags": (1 << 14) | (1 << 18)}
|
|
schema = get_dynamic_schema_core()
|
|
actions = set(schema["parameters"]["properties"]["action"]["enum"])
|
|
assert actions == set(_CORE_ACTIONS.keys())
|
|
assert schema["name"] == "discord"
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_full_intents_admin_schema(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": ""}},
|
|
)
|
|
mock_req.return_value = {"flags": (1 << 14) | (1 << 18)}
|
|
schema = get_dynamic_schema_admin()
|
|
actions = set(schema["parameters"]["properties"]["action"]["enum"])
|
|
assert actions == set(_ADMIN_ACTIONS.keys())
|
|
assert schema["name"] == "discord_admin"
|
|
# No content warning when MESSAGE_CONTENT is enabled
|
|
assert "MESSAGE_CONTENT" not in schema["description"]
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_no_members_intent_removes_member_actions_from_admin_schema(
|
|
self, mock_req, monkeypatch,
|
|
):
|
|
"""member_info is an admin action; it should be hidden when
|
|
GUILD_MEMBERS intent is missing."""
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": ""}},
|
|
)
|
|
mock_req.return_value = {"flags": 1 << 18} # only MESSAGE_CONTENT
|
|
schema = get_dynamic_schema_admin()
|
|
actions = schema["parameters"]["properties"]["action"]["enum"]
|
|
assert "member_info" not in actions
|
|
assert "member_info" not in schema["description"]
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_no_members_intent_hides_search_members_from_core(
|
|
self, mock_req, monkeypatch,
|
|
):
|
|
"""search_members is a core action gated by GUILD_MEMBERS intent."""
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": ""}},
|
|
)
|
|
mock_req.return_value = {"flags": 1 << 18} # only MESSAGE_CONTENT
|
|
schema = get_dynamic_schema_core()
|
|
actions = schema["parameters"]["properties"]["action"]["enum"]
|
|
assert "search_members" not in actions
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_no_message_content_adds_warning_note(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": ""}},
|
|
)
|
|
mock_req.return_value = {"flags": 1 << 14} # only GUILD_MEMBERS
|
|
schema = get_dynamic_schema_core()
|
|
assert "MESSAGE_CONTENT" in schema["description"]
|
|
# But fetch_messages is still available
|
|
actions = schema["parameters"]["properties"]["action"]["enum"]
|
|
assert "fetch_messages" in actions
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_config_allowlist_narrows_admin_schema(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": "list_guilds,list_channels"}},
|
|
)
|
|
mock_req.return_value = {"flags": (1 << 14) | (1 << 18)}
|
|
schema = get_dynamic_schema_admin()
|
|
actions = schema["parameters"]["properties"]["action"]["enum"]
|
|
assert actions == ["list_guilds", "list_channels"]
|
|
assert "list_guilds()" in schema["description"]
|
|
assert "add_role(" not in schema["description"]
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_empty_allowlist_with_valid_values_hides_tools(self, mock_req, monkeypatch):
|
|
"""If the allowlist resolves to zero valid actions (e.g. all names
|
|
were typos), get_dynamic_schema returns None so the tool is dropped."""
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": "typo_one,typo_two"}},
|
|
)
|
|
mock_req.return_value = {"flags": (1 << 14) | (1 << 18)}
|
|
assert get_dynamic_schema_core() is None
|
|
assert get_dynamic_schema_admin() is None
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_backward_compat_wrapper(self, mock_req, monkeypatch):
|
|
"""get_dynamic_schema() should delegate to get_dynamic_schema_core()."""
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": ""}},
|
|
)
|
|
mock_req.return_value = {"flags": (1 << 14) | (1 << 18)}
|
|
schema = get_dynamic_schema()
|
|
assert schema is not None
|
|
assert schema["name"] == "discord"
|
|
actions = set(schema["parameters"]["properties"]["action"]["enum"])
|
|
assert actions == set(_CORE_ACTIONS.keys())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Runtime allowlist enforcement (defense in depth — schema already filtered)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRuntimeAllowlistEnforcement:
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_denied_action_blocked_at_runtime(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": "list_guilds"}},
|
|
)
|
|
result = json.loads(discord_admin_handler(action="add_role", guild_id="1", user_id="2", role_id="3"))
|
|
assert "error" in result
|
|
assert "disabled by config" in result["error"]
|
|
mock_req.assert_not_called()
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_allowed_action_proceeds(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": "list_guilds"}},
|
|
)
|
|
mock_req.return_value = []
|
|
result = json.loads(discord_admin_handler(action="list_guilds"))
|
|
assert "guilds" in result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 403 enrichment
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class Test403Enrichment:
|
|
def test_enrich_known_action(self):
|
|
msg = _enrich_403("add_role", '{"message":"Missing Permissions"}')
|
|
assert "MANAGE_ROLES" in msg
|
|
assert "Missing Permissions" in msg # Raw body preserved
|
|
|
|
def test_enrich_unknown_action_includes_body(self):
|
|
msg = _enrich_403("some_new_action", '{"message":"weird"}')
|
|
assert "some_new_action" in msg
|
|
assert "weird" in msg
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_403_in_runtime_is_enriched(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": ""}},
|
|
)
|
|
mock_req.side_effect = DiscordAPIError(403, '{"message":"Missing Permissions"}')
|
|
result = json.loads(discord_admin_handler(
|
|
action="add_role", guild_id="1", user_id="2", role_id="3",
|
|
))
|
|
assert "error" in result
|
|
assert "MANAGE_ROLES" in result["error"]
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_non_403_errors_are_not_enriched(self, mock_req, monkeypatch):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": ""}},
|
|
)
|
|
mock_req.side_effect = DiscordAPIError(500, "server error")
|
|
result = json.loads(discord_admin_handler(action="list_guilds"))
|
|
assert "500" in result["error"]
|
|
assert "MANAGE_ROLES" not in result["error"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# model_tools integration — dynamic schema replaces static
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestModelToolsIntegration:
|
|
def setup_method(self):
|
|
_reset_capability_cache()
|
|
|
|
def teardown_method(self):
|
|
_reset_capability_cache()
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_discord_admin_schema_rebuilt_by_get_tool_definitions(
|
|
self, mock_req, monkeypatch,
|
|
):
|
|
"""When model_tools.get_tool_definitions runs with discord_admin
|
|
available, it should replace the static schema with the dynamic one."""
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": "list_guilds,server_info"}},
|
|
)
|
|
# Bot without GUILD_MEMBERS intent
|
|
mock_req.return_value = {"flags": 0}
|
|
|
|
from model_tools import get_tool_definitions
|
|
tools = get_tool_definitions(enabled_toolsets=["hermes-discord"], quiet_mode=True)
|
|
discord_admin_tool = next(
|
|
(t for t in tools if t.get("function", {}).get("name") == "discord_admin"),
|
|
None,
|
|
)
|
|
assert discord_admin_tool is not None, "discord_admin should be in the schema"
|
|
actions = discord_admin_tool["function"]["parameters"]["properties"]["action"]["enum"]
|
|
assert actions == ["list_guilds", "server_info"]
|
|
|
|
@patch("tools.discord_tool._discord_request")
|
|
def test_discord_tools_dropped_when_allowlist_empties_them(
|
|
self, mock_req, monkeypatch,
|
|
):
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "tok")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"discord": {"server_actions": "all_bogus_names"}},
|
|
)
|
|
mock_req.return_value = {"flags": 0}
|
|
|
|
from model_tools import get_tool_definitions
|
|
tools = get_tool_definitions(enabled_toolsets=["hermes-discord"], quiet_mode=True)
|
|
names = [t.get("function", {}).get("name") for t in tools]
|
|
assert "discord" not in names
|
|
assert "discord_admin" not in names
|
|
assert "discord_server" not in names
|