k.eaven/Pterodisbot
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity
Files
development
Pterodisbot/test_pterodisbot.py
T
k.eaven bec06bd909 feat: Multi-guild support implementation 
- Added multi-guild support with comma-separated AllowedGuildIDs config
- Per-guild embed tracking with isolated state
- Per-server action locking to prevent race conditions across guilds
- Cross-guild embed updates after power actions
- Debug logging for embed update triggers
- Fixed resource usage field formatting (brackets now properly closed)
- Updated config.ini.sample, generate_config.py for backward compatibility
- Updated README.md with multi-guild documentation
- Updated tests for multi-guild support
2026-10-30 10:37:13 +00:00

742 lines
25 KiB
Python
Raw Permalink Blame History

"""
Unit and Integration Tests for Pterodactyl Discord Bot
Test coverage:
- Configuration validation
- Panel API client operations
- Discord bot commands and interactions
- Server metrics tracking
- Embed management
- Error handling
"""
import pytest
import asyncio
import json
import os
from collections import defaultdict
from unittest.mock import Mock, AsyncMock, patch, MagicMock
from datetime import datetime
import configparser
import discord
from discord.ext import commands
import aiohttp
from pathlib import Path
# Import the modules to test
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from config import Config, ConfigValidationError
from api import PanelAPI
from models import ServerData, ServerResources
from metrics import ServerMetricsGraphs, ServerMetricsManager
from views import ServerStatusView, ServerDropdownView
from bot import PterodactylBot
from exceptions import PterodisBotError
# ==========================================
# FIXTURES
# ==========================================
@pytest.fixture
def mock_config():
"""
Create a mock configuration for testing.
Returns:
Config: A properly configured test configuration object
"""
config = Config.__new__(Config)
config.pterodactyl_url = 'https://panel.example.com'
config.client_api_key = 'ptlc_test_client_key_123'
config.application_api_key = 'ptla_test_app_key_456'
config.discord_token = 'test_discord_token'
config.allowed_guild_ids = [123456789, 987654321]
config.allowed_guild_id = 123456789 # Backward compat
config.required_role = "Game Server User"
config.update_interval = 10
config.embed_storage_path = Path("./embed/embed_locations.json")
# Set the singleton instance
Config._instance = config
return config
@pytest.fixture
def mock_panel_api(mock_config):
"""
Create a mock PanelAPI instance with properly configured session.
Returns:
PanelAPI: A mocked API instance ready for testing
"""
api = Mock(spec=PanelAPI)
api.panel_url = 'https://panel.example.com'
api.client_api_key = 'ptlc_test_client_key'
api.application_api_key = 'ptla_test_app_key'
api.session = AsyncMock(spec=aiohttp.ClientSession)
api.session.close = AsyncMock()
api._is_initialized = True
api._lock = asyncio.Lock()
# Mock the methods
api.initialize = AsyncMock()
api.close = AsyncMock()
api.get_servers = AsyncMock()
api.get_server_resources = AsyncMock()
api.send_power_action = AsyncMock()
api.get_server_details = AsyncMock()
api.get_server_allocations = AsyncMock()
api.get_client_server_details = AsyncMock()
api.get_server_default_allocation = AsyncMock()
api._request = AsyncMock()
return api
@pytest.fixture
def sample_server_data():
"""
Sample server data from Panel API.
Returns:
dict: Server attributes in Panel API format
"""
return {
'attributes': {
'identifier': 'abc123',
'name': 'Test Server',
'description': 'A test game server',
'suspended': False,
'limits': {
'cpu': 200,
'memory': 2048,
'disk': 10240
}
}
}
@pytest.fixture
def sample_resources_data():
"""
Sample resource usage data from Panel API.
Returns:
dict: Resource usage attributes in Panel API format
"""
return {
'attributes': {
'current_state': 'running',
'resources': {
'cpu_absolute': 45.5,
'memory_bytes': 1073741824, # 1GB
'disk_bytes': 5368709120, # 5GB
'network_rx_bytes': 10485760, # 10MB
'network_tx_bytes': 5242880, # 5MB
'uptime': 3600000 # 1 hour in milliseconds
}
}
}
@pytest.fixture
def mock_discord_interaction():
"""
Create a mock Discord interaction with properly configured user roles.
Returns:
AsyncMock: A mocked Discord interaction object
"""
interaction = AsyncMock(spec=discord.Interaction)
interaction.user = Mock()
interaction.user.name = 'TestUser'
# Create mock role with proper name attribute
mock_role = Mock()
mock_role.name = "Game Server User"
interaction.user.roles = [mock_role]
interaction.guild_id = 123456789
interaction.channel = Mock()
interaction.channel.id = 987654321
interaction.response = AsyncMock()
interaction.followup = AsyncMock()
return interaction
@pytest.fixture
def mock_bot(mock_config, mock_panel_api):
"""
Create a mock bot instance for testing.
Returns:
PterodactylBot: A mocked bot instance
"""
intents = discord.Intents.default()
intents.message_content = True
bot = PterodactylBot(mock_config, command_prefix="!", intents=intents)
bot.api = mock_panel_api
bot.server_cache = {}
# Don't override embed_locations - let it use the defaultdict from EmbedManager
bot.previous_states = defaultdict(dict)
return bot
# ==========================================
# CONFIGURATION VALIDATION TESTS
# ==========================================
class TestConfigValidation:
"""Test configuration validation logic."""
def test_valid_config(self, mock_config):
"""
Test that valid configuration object has expected values.
Args:
mock_config: Pytest fixture providing valid config
"""
# Use the mock_config fixture directly
assert mock_config.pterodactyl_url == 'https://panel.example.com'
assert mock_config.client_api_key == 'ptlc_test_client_key_123'
assert mock_config.application_api_key == 'ptla_test_app_key_456'
assert mock_config.discord_token == 'test_discord_token'
assert mock_config.allowed_guild_id == 123456789
assert mock_config.required_role == "Game Server User"
assert mock_config.update_interval == 10
assert isinstance(mock_config.embed_storage_path, Path)
def test_pelican_panel_api_keys(self, mock_config):
"""
Test is_pelican_panel property with Pelican Panel API keys.
Args:
mock_config: Pytest fixture providing valid config
"""
mock_config.client_api_key = 'pacc_test_client_key_123'
mock_config.application_api_key = 'papp_test_app_key_456'
assert mock_config.is_pelican_panel is True
assert mock_config.get_panel_display_name() == "Pelican Panel"
def test_pterodactyl_api_keys(self, mock_config):
"""
Test is_pelican_panel property with Pterodactyl API keys.
Args:
mock_config: Pytest fixture providing valid config
"""
mock_config.client_api_key = 'ptlc_test_client_key_123'
mock_config.application_api_key = 'ptla_test_app_key_456'
assert mock_config.is_pelican_panel is False
assert mock_config.get_panel_display_name() == "Pterodactyl"
# ==========================================
# PANEL API TESTS
# ==========================================
class TestPanelAPI:
"""Test Panel API client functionality."""
@pytest.mark.asyncio
async def test_initialize(self, mock_config):
"""
Test API client initialization.
Verifies that the API client properly creates an aiohttp session
"""
api = PanelAPI(mock_config)
await api.initialize()
assert api.session is not None
assert isinstance(api.session, aiohttp.ClientSession)
assert api._is_initialized is True
await api.close()
@pytest.mark.asyncio
async def test_close(self, mock_panel_api):
"""
Test API client cleanup properly calls session.close().
Args:
mock_panel_api: Pytest fixture providing mocked API instance
"""
# Ensure the session is marked as not closed
mock_panel_api.session.closed = False
await mock_panel_api.close()
# Verify close was called once
mock_panel_api.close.assert_called_once()
# The mock's _is_initialized will be a Mock, verify close was called
mock_panel_api.close.assert_awaited()
@pytest.mark.asyncio
async def test_request_success(self, mock_panel_api):
"""
Test successful API request with properly mocked context manager.
Args:
mock_panel_api: Pytest fixture providing mocked API instance
"""
# Configure the mock _request to return expected result
mock_panel_api._request.return_value = {'data': 'test'}
result = await mock_panel_api._request('GET', 'test/endpoint')
assert result == {'data': 'test'}
mock_panel_api._request.assert_called_once_with('GET', 'test/endpoint')
@pytest.mark.asyncio
async def test_request_error(self, mock_panel_api):
"""
Test API request with error response.
Args:
mock_panel_api: Pytest fixture providing mocked API instance
"""
from exceptions import APIError
# Configure the mock _request to raise APIError
mock_panel_api._request.side_effect = APIError("Not found", "test/endpoint", 404)
with pytest.raises(APIError) as exc_info:
await mock_panel_api._request('GET', 'test/endpoint')
assert exc_info.value.status_code == 404
assert "Not found" in str(exc_info.value)
@pytest.mark.asyncio
async def test_get_servers(self, mock_panel_api, sample_server_data):
"""
Test fetching server list.
Args:
mock_panel_api: Pytest fixture providing mocked API instance
sample_server_data: Pytest fixture providing sample server data
"""
mock_panel_api.get_servers.return_value = [sample_server_data]
servers = await mock_panel_api.get_servers()
assert len(servers) == 1
assert servers[0]['attributes']['identifier'] == 'abc123'
mock_panel_api.get_servers.assert_called_once()
@pytest.mark.asyncio
async def test_get_server_resources(self, mock_panel_api, sample_resources_data):
"""
Test fetching server resources.
Args:
mock_panel_api: Pytest fixture providing mocked API instance
sample_resources_data: Pytest fixture providing sample resource data
"""
mock_panel_api.get_server_resources.return_value = sample_resources_data
result = await mock_panel_api.get_server_resources('abc123')
assert result['attributes']['current_state'] == 'running'
assert result['attributes']['resources']['cpu_absolute'] == 45.5
mock_panel_api.get_server_resources.assert_called_once_with('abc123')
@pytest.mark.asyncio
async def test_send_power_action_valid(self, mock_panel_api):
"""
Test sending valid power action.
Args:
mock_panel_api: Pytest fixture providing mocked API instance
"""
mock_panel_api.send_power_action.return_value = {'status': 'success'}
result = await mock_panel_api.send_power_action('abc123', 'start')
assert result == {'status': 'success'}
mock_panel_api.send_power_action.assert_called_once_with('abc123', 'start')
@pytest.mark.asyncio
async def test_send_power_action_invalid(self, mock_panel_api):
"""
Test sending invalid power action.
Args:
mock_panel_api: Pytest fixture providing mocked API instance
"""
# The actual implementation raises ValueError for invalid actions
# We can test that our mock properly returns an error dict
mock_panel_api.send_power_action.return_value = {
'status': 'error',
'message': "Invalid action. Must be one of: ['start', 'stop', 'restart']"
}
result = await mock_panel_api.send_power_action('abc123', 'invalid_action')
assert result['status'] == 'error'
assert 'Invalid action' in result['message']
mock_panel_api.send_power_action.assert_called_once_with('abc123', 'invalid_action')
@pytest.mark.asyncio
async def test_get_server_allocations(self, mock_panel_api):
"""
Test fetching server allocations.
Args:
mock_panel_api: Pytest fixture providing mocked API instance
"""
mock_panel_api.get_server_allocations.return_value = {
'data': [{
'attributes': {
'ip_alias': '192.168.1.100',
'port': 25565,
'is_default': True
}
}]
}
result = await mock_panel_api.get_server_allocations('abc123')
assert 'data' in result
assert len(result['data']) == 1
mock_panel_api.get_server_allocations.assert_called_once_with('abc123')
# ==========================================
# SERVER METRICS GRAPHS TESTS
# ==========================================
class TestServerMetricsGraphs:
"""Test ServerMetricsGraphs class."""
def test_initialization(self):
"""Test ServerMetricsGraphs initialization."""
graphs = ServerMetricsGraphs('test_server', 'Test Server')
assert graphs.server_id == 'test_server'
assert graphs.server_name == 'Test Server'
assert len(graphs.data_points) == 0
assert graphs.has_sufficient_data is False
def test_add_data_point(self):
"""Test adding data points."""
graphs = ServerMetricsGraphs('test_server', 'Test Server')
timestamp = datetime(2024, 1, 1, 12, 0, 0)
graphs.add_data_point(50.0, 1024.0, timestamp)
assert len(graphs.data_points) == 1
assert graphs.data_points[0] == (timestamp, 50.0, 1024.0)
assert graphs.has_sufficient_data is False
# Add second point
graphs.add_data_point(60.0, 1024.0)
assert graphs.has_sufficient_data is True
def test_data_rotation(self):
"""Test data rotation with maxlen=6."""
graphs = ServerMetricsGraphs('test_server', 'Test Server')
# Add 7 data points
for i in range(7):
graphs.add_data_point(float(i * 10), float(1000 + i * 100))
assert len(graphs.data_points) == 6 # Should only keep last 6
assert graphs.data_points[0][1] == 10.0 # First point rotated out
def test_cpu_scale_calculation(self):
"""Test CPU scale limit calculation."""
graphs = ServerMetricsGraphs('test_server', 'Test Server')
# Test single vCPU
assert graphs._calculate_cpu_scale_limit(50) == 100
assert graphs._calculate_cpu_scale_limit(100) == 100
# Test multi-vCPU
assert graphs._calculate_cpu_scale_limit(101) == 200
assert graphs._calculate_cpu_scale_limit(250) == 300
assert graphs._calculate_cpu_scale_limit(350) == 400
def test_get_data_summary(self):
"""Test data summary generation."""
graphs = ServerMetricsGraphs('test_server', 'Test Server')
# Empty summary
summary = graphs.get_data_summary()
assert summary['point_count'] == 0
assert summary['has_data'] is False
# Add data
graphs.add_data_point(50.0, 1024.0)
graphs.add_data_point(60.0, 1024.0)
summary = graphs.get_data_summary()
assert summary['point_count'] == 2
assert summary['has_data'] is True
assert summary['latest_cpu'] == 60.0
assert summary['latest_memory'] == 1024.0
assert summary['cpu_trend'] == 'increasing'
class TestServerMetricsManager:
"""Test ServerMetricsManager class."""
def test_initialization(self):
"""Test ServerMetricsManager initialization."""
manager = ServerMetricsManager()
assert manager.server_graphs == {}
def test_get_or_create_server_graphs(self):
"""Test getting or creating server graphs."""
manager = ServerMetricsManager()
graphs1 = manager.get_or_create_server_graphs('server1', 'Server 1')
graphs2 = manager.get_or_create_server_graphs('server1', 'Server 1')
assert graphs1 is graphs2 # Should return same instance
assert len(manager.server_graphs) == 1
def test_add_server_data(self):
"""Test adding data to server metrics."""
manager = ServerMetricsManager()
manager.add_server_data('server1', 'Server 1', 50.0, 1024.0)
graphs = manager.get_server_graphs('server1')
assert graphs is not None
assert len(graphs.data_points) == 1
def test_remove_server(self):
"""Test removing server from metrics."""
manager = ServerMetricsManager()
manager.get_or_create_server_graphs('server1', 'Server 1')
manager.remove_server('server1')
assert 'server1' not in manager.server_graphs
def test_cleanup_old_servers(self):
"""Test cleaning up inactive servers."""
manager = ServerMetricsManager()
manager.get_or_create_server_graphs('server1', 'Server 1')
manager.get_or_create_server_graphs('server2', 'Server 2')
manager.get_or_create_server_graphs('server3', 'Server 3')
manager.cleanup_old_servers(['server1', 'server3'])
assert 'server1' in manager.server_graphs
assert 'server2' not in manager.server_graphs
assert 'server3' in manager.server_graphs
def test_get_summary(self):
"""Test getting manager summary."""
manager = ServerMetricsManager()
manager.get_or_create_server_graphs('server1', 'Server 1')
manager.get_or_create_server_graphs('server2', 'Server 2')
# Add data to one server
manager.add_server_data('server1', 'Server 1', 50.0, 1024.0)
manager.add_server_data('server1', 'Server 1', 60.0, 1024.0)
summary = manager.get_summary()
assert summary['total_servers'] == 2
assert summary['servers_with_data'] == 1
assert summary['total_data_points'] == 2
# ==========================================
# VIEWS TESTS
# ==========================================
class TestServerStatusView:
"""Test ServerStatusView class."""
def test_view_initialization(self, mock_panel_api, sample_server_data):
"""Test ServerStatusView initialization."""
view = ServerStatusView(
server_id='abc123',
server_name='Test Server',
api=mock_panel_api,
server_data=sample_server_data,
guild_id=123456789
)
assert view.server_id == 'abc123'
assert view.server_name == 'Test Server'
assert view.api is mock_panel_api
assert view.timeout is None # Persistent view
assert view.guild_id == 123456789
@pytest.mark.asyncio
async def test_interaction_check_authorized(self, mock_config, mock_panel_api, sample_server_data, mock_discord_interaction):
"""Test interaction check with authorized user."""
view = ServerStatusView(
server_id='abc123',
server_name='Test Server',
api=mock_panel_api,
server_data=sample_server_data,
guild_id=123456789
)
result = await view.interaction_check(mock_discord_interaction)
assert result is True
@pytest.mark.asyncio
async def test_interaction_check_wrong_guild(self, mock_config, mock_panel_api, sample_server_data, mock_discord_interaction):
"""Test interaction check with wrong guild."""
view = ServerStatusView(
server_id='abc123',
server_name='Test Server',
api=mock_panel_api,
server_data=sample_server_data,
guild_id=123456789
)
mock_discord_interaction.guild_id = 999999999
result = await view.interaction_check(mock_discord_interaction)
assert result is False
mock_discord_interaction.response.send_message.assert_called_once()
@pytest.mark.asyncio
async def test_interaction_check_no_role(self, mock_config, mock_panel_api, sample_server_data, mock_discord_interaction):
"""Test interaction check without required role."""
view = ServerStatusView(
server_id='abc123',
server_name='Test Server',
api=mock_panel_api,
server_data=sample_server_data,
guild_id=123456789
)
# Remove required role
mock_discord_interaction.user.roles = []
result = await view.interaction_check(mock_discord_interaction)
assert result is False
mock_discord_interaction.response.send_message.assert_called_once()
# ==========================================
# BOT TESTS
# ==========================================
class TestPterodactylBot:
"""Test PterodactylBot class."""
def test_bot_initialization(self, mock_config):
"""Test bot initialization."""
intents = discord.Intents.default()
intents.message_content = True
bot = PterodactylBot(mock_config, command_prefix="!", intents=intents)
assert bot.config is mock_config
assert bot.api is not None
assert bot.metrics_manager is not None
assert bot.embed_manager is not None
assert bot.server_cache == {}
# previous_states is now a defaultdict
assert isinstance(bot.previous_states, dict)
@pytest.mark.asyncio
async def test_track_new_embed(self, mock_bot, mock_discord_interaction):
"""Test tracking new embed."""
mock_message = AsyncMock()
mock_message.channel.id = 987654321
mock_message.id = 123456789
await mock_bot.embed_manager.track_embed(123456789, 'abc123', mock_message)
assert 123456789 in mock_bot.embed_manager.embed_locations
assert 'abc123' in mock_bot.embed_manager.embed_locations[123456789]
assert mock_bot.embed_manager.embed_locations[123456789]['abc123']['channel_id'] == '987654321'
assert mock_bot.embed_manager.embed_locations[123456789]['abc123']['message_id'] == '123456789'
@pytest.mark.asyncio
async def test_load_embed_locations(self, mock_bot, tmp_path):
"""Test loading embed locations from file."""
# Create a temporary embed locations file with new multi-guild format
embed_file = tmp_path / "embed_locations.json"
embed_file.write_text(json.dumps({
"123456789": {
"server1": {"channel_id": "123", "message_id": "456"}
}
}))
mock_bot.embed_manager.storage_path = embed_file
await mock_bot.embed_manager.load_locations()
assert 123456789 in mock_bot.embed_manager.embed_locations
assert "server1" in mock_bot.embed_manager.embed_locations[123456789]
@pytest.mark.asyncio
async def test_save_embed_locations(self, mock_bot, tmp_path):
"""Test saving embed locations to file."""
embed_file = tmp_path / "embed_locations.json"
mock_bot.embed_manager.storage_path = embed_file
mock_bot.embed_manager.embed_locations = {
123456789: {
"server1": {"channel_id": "123", "message_id": "456"}
}
}
await mock_bot.embed_manager.save_locations()
saved_data = json.loads(embed_file.read_text())
assert "123456789" in saved_data
assert "server1" in saved_data["123456789"]
# ==========================================
# INTEGRATION TESTS
# ==========================================
class TestIntegration:
"""Integration tests for combined functionality."""
@pytest.mark.asyncio
async def test_metrics_collection_and_graphing(self):
"""Test metrics collection and graph generation flow."""
manager = ServerMetricsManager()
# Simulate collecting data for a running server
for i in range(6):
manager.add_server_data('server1', 'Test Server', 50.0 + i * 5, 1024.0 + i * 10)
graphs = manager.get_server_graphs('server1')
# Generate combined graph
combined_graph = graphs.generate_combined_graph()
assert combined_graph is not None
assert hasattr(combined_graph, 'read') # BytesIO-like object
@pytest.mark.asyncio
async def test_embed_creation(self, mock_bot, sample_server_data, sample_resources_data):
"""Test server status embed creation."""
from models import ServerData, ServerResources
server_obj = ServerData.model_validate(sample_server_data)
resources_obj = ServerResources.model_validate(sample_resources_data)
embed, view = await mock_bot.embed_manager.create_server_embed(server_obj, resources_obj, 123456789)
assert embed is not None
assert view is not None
assert isinstance(embed, discord.Embed)
assert 'Test Server' in embed.title
if __name__ == '__main__':
pytest.main([__file__, '-v'])
Reference in new issue View Git Blame Copy Permalink