6bcfee519c
- Add file_lock context manager to eliminate duplicate locking patterns - Add ValidationError, ConfigurationError, CertificateError exceptions - Improve rollback logic in haproxy_add_servers (track successful ops only) - Decompose haproxy_add_domain into smaller helper functions - Consolidate certificate constants (CERTS_DIR, ACME_HOME) to config.py - Enhance docstrings for internal functions and magic numbers - Add pytest framework with 48 new tests (269 -> 317 total) - Increase test coverage from 76% to 86% - servers.py: 58% -> 82% - certificates.py: 67% -> 86% - configuration.py: 69% -> 94% Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
326 lines
11 KiB
Python
326 lines
11 KiB
Python
"""Unit tests for monitoring tools."""
|
|
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
|
|
|
|
class TestHaproxyStats:
|
|
"""Tests for haproxy_stats tool function."""
|
|
|
|
def test_stats_success(self, mock_socket_class, mock_select, response_builder):
|
|
"""Get HAProxy stats successfully."""
|
|
mock_sock = mock_socket_class(responses={
|
|
"show info": response_builder.info(version="3.3.2", uptime=3600),
|
|
})
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
from haproxy_mcp.tools.monitoring import register_monitoring_tools
|
|
mcp = MagicMock()
|
|
registered_tools = {}
|
|
|
|
def capture_tool():
|
|
def decorator(func):
|
|
registered_tools[func.__name__] = func
|
|
return func
|
|
return decorator
|
|
|
|
mcp.tool = capture_tool
|
|
register_monitoring_tools(mcp)
|
|
|
|
result = registered_tools["haproxy_stats"]()
|
|
|
|
assert "Version" in result
|
|
assert "3.3.2" in result
|
|
|
|
def test_stats_haproxy_error(self, mock_select):
|
|
"""Handle HAProxy connection error."""
|
|
def raise_error(*args, **kwargs):
|
|
raise ConnectionRefusedError()
|
|
|
|
with patch("socket.socket", side_effect=raise_error):
|
|
from haproxy_mcp.tools.monitoring import register_monitoring_tools
|
|
mcp = MagicMock()
|
|
registered_tools = {}
|
|
|
|
def capture_tool():
|
|
def decorator(func):
|
|
registered_tools[func.__name__] = func
|
|
return func
|
|
return decorator
|
|
|
|
mcp.tool = capture_tool
|
|
register_monitoring_tools(mcp)
|
|
|
|
result = registered_tools["haproxy_stats"]()
|
|
|
|
assert "Error" in result
|
|
|
|
|
|
class TestHaproxyBackends:
|
|
"""Tests for haproxy_backends tool function."""
|
|
|
|
def test_backends_success(self, mock_socket_class, mock_select):
|
|
"""List backends successfully."""
|
|
mock_sock = mock_socket_class(responses={
|
|
"show backend": "pool_1\npool_2\npool_3",
|
|
})
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
from haproxy_mcp.tools.monitoring import register_monitoring_tools
|
|
mcp = MagicMock()
|
|
registered_tools = {}
|
|
|
|
def capture_tool():
|
|
def decorator(func):
|
|
registered_tools[func.__name__] = func
|
|
return func
|
|
return decorator
|
|
|
|
mcp.tool = capture_tool
|
|
register_monitoring_tools(mcp)
|
|
|
|
result = registered_tools["haproxy_backends"]()
|
|
|
|
assert "Backends" in result
|
|
assert "pool_1" in result
|
|
assert "pool_2" in result
|
|
assert "pool_3" in result
|
|
|
|
def test_backends_haproxy_error(self, mock_select):
|
|
"""Handle HAProxy connection error."""
|
|
def raise_error(*args, **kwargs):
|
|
raise ConnectionRefusedError()
|
|
|
|
with patch("socket.socket", side_effect=raise_error):
|
|
from haproxy_mcp.tools.monitoring import register_monitoring_tools
|
|
mcp = MagicMock()
|
|
registered_tools = {}
|
|
|
|
def capture_tool():
|
|
def decorator(func):
|
|
registered_tools[func.__name__] = func
|
|
return func
|
|
return decorator
|
|
|
|
mcp.tool = capture_tool
|
|
register_monitoring_tools(mcp)
|
|
|
|
result = registered_tools["haproxy_backends"]()
|
|
|
|
assert "Error" in result
|
|
|
|
|
|
class TestHaproxyListFrontends:
|
|
"""Tests for haproxy_list_frontends tool function."""
|
|
|
|
def test_list_frontends_success(self, mock_socket_class, mock_select, response_builder):
|
|
"""List frontends successfully."""
|
|
mock_sock = mock_socket_class(responses={
|
|
"show stat": response_builder.stat_csv([
|
|
{"pxname": "http_front", "svname": "FRONTEND", "status": "OPEN", "scur": 10},
|
|
{"pxname": "https_front", "svname": "FRONTEND", "status": "OPEN", "scur": 50},
|
|
{"pxname": "pool_1", "svname": "pool_1_1", "status": "UP", "scur": 5},
|
|
]),
|
|
})
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
from haproxy_mcp.tools.monitoring import register_monitoring_tools
|
|
mcp = MagicMock()
|
|
registered_tools = {}
|
|
|
|
def capture_tool():
|
|
def decorator(func):
|
|
registered_tools[func.__name__] = func
|
|
return func
|
|
return decorator
|
|
|
|
mcp.tool = capture_tool
|
|
register_monitoring_tools(mcp)
|
|
|
|
result = registered_tools["haproxy_list_frontends"]()
|
|
|
|
assert "Frontends" in result
|
|
assert "http_front" in result
|
|
assert "https_front" in result
|
|
# pool_1 is not a FRONTEND
|
|
assert "pool_1_1" not in result
|
|
|
|
def test_list_frontends_no_frontends(self, mock_socket_class, mock_select, response_builder):
|
|
"""No frontends found."""
|
|
mock_sock = mock_socket_class(responses={
|
|
"show stat": response_builder.stat_csv([
|
|
{"pxname": "pool_1", "svname": "pool_1_1", "status": "UP"},
|
|
]),
|
|
})
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
from haproxy_mcp.tools.monitoring import register_monitoring_tools
|
|
mcp = MagicMock()
|
|
registered_tools = {}
|
|
|
|
def capture_tool():
|
|
def decorator(func):
|
|
registered_tools[func.__name__] = func
|
|
return func
|
|
return decorator
|
|
|
|
mcp.tool = capture_tool
|
|
register_monitoring_tools(mcp)
|
|
|
|
result = registered_tools["haproxy_list_frontends"]()
|
|
|
|
assert "No frontends found" in result
|
|
|
|
def test_list_frontends_haproxy_error(self, mock_select):
|
|
"""Handle HAProxy connection error."""
|
|
def raise_error(*args, **kwargs):
|
|
raise ConnectionRefusedError()
|
|
|
|
with patch("socket.socket", side_effect=raise_error):
|
|
from haproxy_mcp.tools.monitoring import register_monitoring_tools
|
|
mcp = MagicMock()
|
|
registered_tools = {}
|
|
|
|
def capture_tool():
|
|
def decorator(func):
|
|
registered_tools[func.__name__] = func
|
|
return func
|
|
return decorator
|
|
|
|
mcp.tool = capture_tool
|
|
register_monitoring_tools(mcp)
|
|
|
|
result = registered_tools["haproxy_list_frontends"]()
|
|
|
|
assert "Error" in result
|
|
|
|
|
|
class TestHaproxyGetConnections:
|
|
"""Tests for haproxy_get_connections tool function."""
|
|
|
|
def test_get_connections_all_backends(self, mock_socket_class, mock_select, response_builder):
|
|
"""Get connections for all backends."""
|
|
mock_sock = mock_socket_class(responses={
|
|
"show stat": response_builder.stat_csv([
|
|
{"pxname": "pool_1", "svname": "FRONTEND", "status": "OPEN", "scur": 10, "smax": 100},
|
|
{"pxname": "pool_1", "svname": "pool_1_1", "status": "UP", "scur": 5},
|
|
{"pxname": "pool_1", "svname": "BACKEND", "status": "UP", "scur": 10, "smax": 100},
|
|
]),
|
|
})
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
from haproxy_mcp.tools.monitoring import register_monitoring_tools
|
|
mcp = MagicMock()
|
|
registered_tools = {}
|
|
|
|
def capture_tool():
|
|
def decorator(func):
|
|
registered_tools[func.__name__] = func
|
|
return func
|
|
return decorator
|
|
|
|
mcp.tool = capture_tool
|
|
register_monitoring_tools(mcp)
|
|
|
|
result = registered_tools["haproxy_get_connections"](backend="")
|
|
|
|
assert "pool_1" in result
|
|
assert "FRONTEND" in result or "BACKEND" in result
|
|
assert "connections" in result
|
|
|
|
def test_get_connections_filter_backend(self, mock_socket_class, mock_select, response_builder):
|
|
"""Filter connections by backend."""
|
|
mock_sock = mock_socket_class(responses={
|
|
"show stat": response_builder.stat_csv([
|
|
{"pxname": "pool_1", "svname": "FRONTEND", "status": "OPEN", "scur": 10, "smax": 100},
|
|
{"pxname": "pool_2", "svname": "FRONTEND", "status": "OPEN", "scur": 20, "smax": 200},
|
|
]),
|
|
})
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
from haproxy_mcp.tools.monitoring import register_monitoring_tools
|
|
mcp = MagicMock()
|
|
registered_tools = {}
|
|
|
|
def capture_tool():
|
|
def decorator(func):
|
|
registered_tools[func.__name__] = func
|
|
return func
|
|
return decorator
|
|
|
|
mcp.tool = capture_tool
|
|
register_monitoring_tools(mcp)
|
|
|
|
result = registered_tools["haproxy_get_connections"](backend="pool_1")
|
|
|
|
assert "pool_1" in result
|
|
assert "pool_2" not in result
|
|
|
|
def test_get_connections_invalid_backend(self):
|
|
"""Reject invalid backend name."""
|
|
from haproxy_mcp.tools.monitoring import register_monitoring_tools
|
|
mcp = MagicMock()
|
|
registered_tools = {}
|
|
|
|
def capture_tool():
|
|
def decorator(func):
|
|
registered_tools[func.__name__] = func
|
|
return func
|
|
return decorator
|
|
|
|
mcp.tool = capture_tool
|
|
register_monitoring_tools(mcp)
|
|
|
|
result = registered_tools["haproxy_get_connections"](backend="invalid@name")
|
|
|
|
assert "Error" in result
|
|
assert "Invalid backend" in result
|
|
|
|
def test_get_connections_no_data(self, mock_socket_class, mock_select, response_builder):
|
|
"""No connection data found."""
|
|
mock_sock = mock_socket_class(responses={
|
|
"show stat": response_builder.stat_csv([]),
|
|
})
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
from haproxy_mcp.tools.monitoring import register_monitoring_tools
|
|
mcp = MagicMock()
|
|
registered_tools = {}
|
|
|
|
def capture_tool():
|
|
def decorator(func):
|
|
registered_tools[func.__name__] = func
|
|
return func
|
|
return decorator
|
|
|
|
mcp.tool = capture_tool
|
|
register_monitoring_tools(mcp)
|
|
|
|
result = registered_tools["haproxy_get_connections"](backend="")
|
|
|
|
assert "No connection data" in result
|
|
|
|
def test_get_connections_haproxy_error(self, mock_select):
|
|
"""Handle HAProxy connection error."""
|
|
def raise_error(*args, **kwargs):
|
|
raise ConnectionRefusedError()
|
|
|
|
with patch("socket.socket", side_effect=raise_error):
|
|
from haproxy_mcp.tools.monitoring import register_monitoring_tools
|
|
mcp = MagicMock()
|
|
registered_tools = {}
|
|
|
|
def capture_tool():
|
|
def decorator(func):
|
|
registered_tools[func.__name__] = func
|
|
return func
|
|
return decorator
|
|
|
|
mcp.tool = capture_tool
|
|
register_monitoring_tools(mcp)
|
|
|
|
result = registered_tools["haproxy_get_connections"](backend="")
|
|
|
|
assert "Error" in result
|