225 lines
7.6 KiB
Python
225 lines
7.6 KiB
Python
"""
|
|
Tests for the Proxmox MCP server.
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import pytest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
from mcp.server.fastmcp.exceptions import ToolError
|
|
from proxmox_mcp.server import ProxmoxMCPServer
|
|
|
|
@pytest.fixture
|
|
def mock_env_vars():
|
|
"""Fixture to set up test environment variables."""
|
|
env_vars = {
|
|
"PROXMOX_HOST": "test.proxmox.com",
|
|
"PROXMOX_USER": "test@pve",
|
|
"PROXMOX_TOKEN_NAME": "test_token",
|
|
"PROXMOX_TOKEN_VALUE": "test_value",
|
|
"LOG_LEVEL": "DEBUG"
|
|
}
|
|
with patch.dict(os.environ, env_vars):
|
|
yield env_vars
|
|
|
|
@pytest.fixture
|
|
def mock_proxmox():
|
|
"""Fixture to mock ProxmoxAPI."""
|
|
with patch("proxmox_mcp.server.ProxmoxAPI") as mock:
|
|
mock.return_value.nodes.get.return_value = [
|
|
{"node": "node1", "status": "online"},
|
|
{"node": "node2", "status": "online"}
|
|
]
|
|
yield mock
|
|
|
|
@pytest.fixture
|
|
def server(mock_env_vars, mock_proxmox):
|
|
"""Fixture to create a ProxmoxMCPServer instance."""
|
|
return ProxmoxMCPServer()
|
|
|
|
def test_server_initialization(server, mock_proxmox):
|
|
"""Test server initialization with environment variables."""
|
|
assert server.config.proxmox.host == "test.proxmox.com"
|
|
assert server.config.auth.user == "test@pve"
|
|
assert server.config.auth.token_name == "test_token"
|
|
assert server.config.auth.token_value == "test_value"
|
|
assert server.config.logging.level == "DEBUG"
|
|
|
|
mock_proxmox.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tools(server):
|
|
"""Test listing available tools."""
|
|
tools = await server.mcp.list_tools()
|
|
|
|
assert len(tools) > 0
|
|
tool_names = [tool.name for tool in tools]
|
|
assert "get_nodes" in tool_names
|
|
assert "get_vms" in tool_names
|
|
assert "get_containers" in tool_names
|
|
assert "execute_vm_command" in tool_names
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_nodes(server, mock_proxmox):
|
|
"""Test get_nodes tool."""
|
|
mock_proxmox.return_value.nodes.get.return_value = [
|
|
{"node": "node1", "status": "online"},
|
|
{"node": "node2", "status": "online"}
|
|
]
|
|
response = await server.mcp.call_tool("get_nodes", {})
|
|
result = json.loads(response[0].text)
|
|
|
|
assert len(result) == 2
|
|
assert result[0]["node"] == "node1"
|
|
assert result[1]["node"] == "node2"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_node_status_missing_parameter(server):
|
|
"""Test get_node_status tool with missing parameter."""
|
|
with pytest.raises(ToolError, match="Field required"):
|
|
await server.mcp.call_tool("get_node_status", {})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_node_status(server, mock_proxmox):
|
|
"""Test get_node_status tool with valid parameter."""
|
|
mock_proxmox.return_value.nodes.return_value.status.get.return_value = {
|
|
"status": "running",
|
|
"uptime": 123456
|
|
}
|
|
|
|
response = await server.mcp.call_tool("get_node_status", {"node": "node1"})
|
|
result = json.loads(response[0].text)
|
|
assert result["status"] == "running"
|
|
assert result["uptime"] == 123456
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_vms(server, mock_proxmox):
|
|
"""Test get_vms tool."""
|
|
mock_proxmox.return_value.nodes.get.return_value = [{"node": "node1", "status": "online"}]
|
|
mock_proxmox.return_value.nodes.return_value.qemu.get.return_value = [
|
|
{"vmid": "100", "name": "vm1", "status": "running"},
|
|
{"vmid": "101", "name": "vm2", "status": "stopped"}
|
|
]
|
|
|
|
response = await server.mcp.call_tool("get_vms", {})
|
|
result = json.loads(response[0].text)
|
|
assert len(result) > 0
|
|
assert result[0]["name"] == "vm1"
|
|
assert result[1]["name"] == "vm2"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_containers(server, mock_proxmox):
|
|
"""Test get_containers tool."""
|
|
mock_proxmox.return_value.nodes.get.return_value = [{"node": "node1", "status": "online"}]
|
|
mock_proxmox.return_value.nodes.return_value.lxc.get.return_value = [
|
|
{"vmid": "200", "name": "container1", "status": "running"},
|
|
{"vmid": "201", "name": "container2", "status": "stopped"}
|
|
]
|
|
|
|
response = await server.mcp.call_tool("get_containers", {})
|
|
result = json.loads(response[0].text)
|
|
assert len(result) > 0
|
|
assert result[0]["name"] == "container1"
|
|
assert result[1]["name"] == "container2"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_storage(server, mock_proxmox):
|
|
"""Test get_storage tool."""
|
|
mock_proxmox.return_value.storage.get.return_value = [
|
|
{"storage": "local", "type": "dir"},
|
|
{"storage": "ceph", "type": "rbd"}
|
|
]
|
|
|
|
response = await server.mcp.call_tool("get_storage", {})
|
|
result = json.loads(response[0].text)
|
|
assert len(result) == 2
|
|
assert result[0]["storage"] == "local"
|
|
assert result[1]["storage"] == "ceph"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_cluster_status(server, mock_proxmox):
|
|
"""Test get_cluster_status tool."""
|
|
mock_proxmox.return_value.cluster.status.get.return_value = {
|
|
"quorate": True,
|
|
"nodes": 2
|
|
}
|
|
|
|
response = await server.mcp.call_tool("get_cluster_status", {})
|
|
result = json.loads(response[0].text)
|
|
assert result["quorate"] is True
|
|
assert result["nodes"] == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_vm_command_success(server, mock_proxmox):
|
|
"""Test successful VM command execution."""
|
|
# Mock VM status check
|
|
mock_proxmox.return_value.nodes.return_value.qemu.return_value.status.current.get.return_value = {
|
|
"status": "running"
|
|
}
|
|
# Mock command execution
|
|
mock_proxmox.return_value.nodes.return_value.qemu.return_value.agent.exec.post.return_value = {
|
|
"out": "command output",
|
|
"err": "",
|
|
"exitcode": 0
|
|
}
|
|
|
|
response = await server.mcp.call_tool("execute_vm_command", {
|
|
"node": "node1",
|
|
"vmid": "100",
|
|
"command": "ls -l"
|
|
})
|
|
result = json.loads(response[0].text)
|
|
|
|
assert result["success"] is True
|
|
assert result["output"] == "command output"
|
|
assert result["error"] == ""
|
|
assert result["exit_code"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_vm_command_missing_parameters(server):
|
|
"""Test VM command execution with missing parameters."""
|
|
with pytest.raises(ToolError):
|
|
await server.mcp.call_tool("execute_vm_command", {})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_vm_command_vm_not_running(server, mock_proxmox):
|
|
"""Test VM command execution when VM is not running."""
|
|
mock_proxmox.return_value.nodes.return_value.qemu.return_value.status.current.get.return_value = {
|
|
"status": "stopped"
|
|
}
|
|
|
|
with pytest.raises(ToolError, match="not running"):
|
|
await server.mcp.call_tool("execute_vm_command", {
|
|
"node": "node1",
|
|
"vmid": "100",
|
|
"command": "ls -l"
|
|
})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_vm_command_with_error(server, mock_proxmox):
|
|
"""Test VM command execution with command error."""
|
|
# Mock VM status check
|
|
mock_proxmox.return_value.nodes.return_value.qemu.return_value.status.current.get.return_value = {
|
|
"status": "running"
|
|
}
|
|
# Mock command execution with error
|
|
mock_proxmox.return_value.nodes.return_value.qemu.return_value.agent.exec.post.return_value = {
|
|
"out": "",
|
|
"err": "command not found",
|
|
"exitcode": 1
|
|
}
|
|
|
|
response = await server.mcp.call_tool("execute_vm_command", {
|
|
"node": "node1",
|
|
"vmid": "100",
|
|
"command": "invalid-command"
|
|
})
|
|
result = json.loads(response[0].text)
|
|
|
|
assert result["success"] is True # API call succeeded
|
|
assert result["output"] == ""
|
|
assert result["error"] == "command not found"
|
|
assert result["exit_code"] == 1
|