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>
280 lines
10 KiB
Python
280 lines
10 KiB
Python
"""Unit tests for haproxy_client module."""
|
|
|
|
import socket
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
|
|
from haproxy_mcp.haproxy_client import (
|
|
haproxy_cmd,
|
|
haproxy_cmd_checked,
|
|
haproxy_cmd_batch,
|
|
reload_haproxy,
|
|
)
|
|
from haproxy_mcp.exceptions import HaproxyError
|
|
|
|
|
|
class TestHaproxyCmd:
|
|
"""Tests for haproxy_cmd function."""
|
|
|
|
def test_successful_command(self, mock_socket_class, mock_select):
|
|
"""Successful command execution returns response."""
|
|
mock_sock = mock_socket_class(
|
|
responses={"show info": "Version: 3.3.2\nUptime_sec: 3600"}
|
|
)
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
result = haproxy_cmd("show info")
|
|
|
|
assert "Version: 3.3.2" in result
|
|
assert "show info" in mock_sock.sent_commands
|
|
|
|
def test_empty_response(self, mock_socket_class, mock_select):
|
|
"""Command with empty response returns empty string."""
|
|
mock_sock = mock_socket_class(default_response="")
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
result = haproxy_cmd("set server pool_1/pool_1_1 state ready")
|
|
|
|
assert result == ""
|
|
|
|
def test_connection_refused_error(self, mock_select):
|
|
"""Connection refused raises HaproxyError."""
|
|
with patch("socket.socket") as mock_socket:
|
|
mock_socket.return_value.__enter__ = MagicMock(side_effect=ConnectionRefusedError())
|
|
mock_socket.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
with pytest.raises(HaproxyError) as exc_info:
|
|
haproxy_cmd("show info")
|
|
|
|
assert "Connection refused" in str(exc_info.value)
|
|
|
|
def test_socket_timeout_error(self, mock_select):
|
|
"""Socket timeout raises HaproxyError."""
|
|
with patch("socket.socket") as mock_socket:
|
|
mock_socket.return_value.__enter__ = MagicMock(side_effect=socket.timeout())
|
|
mock_socket.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
with pytest.raises(HaproxyError) as exc_info:
|
|
haproxy_cmd("show info")
|
|
|
|
assert "timeout" in str(exc_info.value).lower()
|
|
|
|
def test_unicode_decode_error(self, mock_socket_class, mock_select):
|
|
"""Invalid UTF-8 response raises HaproxyError."""
|
|
# Create a mock that returns invalid UTF-8 bytes
|
|
class BadUtf8Socket(mock_socket_class):
|
|
def sendall(self, data):
|
|
self.sent_commands.append(data.decode().strip())
|
|
self._response_buffer = b"\xff\xfe" # Invalid UTF-8
|
|
|
|
mock_sock = BadUtf8Socket()
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
with pytest.raises(HaproxyError) as exc_info:
|
|
haproxy_cmd("show info")
|
|
|
|
assert "UTF-8" in str(exc_info.value)
|
|
|
|
def test_multiline_response(self, mock_socket_class, mock_select):
|
|
"""Multi-line response is properly returned."""
|
|
multi_line = "pool_1\npool_2\npool_3"
|
|
mock_sock = mock_socket_class(responses={"show backend": multi_line})
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
result = haproxy_cmd("show backend")
|
|
|
|
assert "pool_1" in result
|
|
assert "pool_2" in result
|
|
assert "pool_3" in result
|
|
|
|
|
|
class TestHaproxyCmdChecked:
|
|
"""Tests for haproxy_cmd_checked function."""
|
|
|
|
def test_successful_command(self, mock_socket_class, mock_select):
|
|
"""Successful command returns response."""
|
|
mock_sock = mock_socket_class(responses={"set server": ""})
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
result = haproxy_cmd_checked("set server pool_1/pool_1_1 state ready")
|
|
|
|
assert result == ""
|
|
|
|
def test_error_response_no_such(self, mock_socket_class, mock_select):
|
|
"""Response containing 'No such' raises HaproxyError."""
|
|
mock_sock = mock_socket_class(
|
|
responses={"set server": "No such server."}
|
|
)
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
with pytest.raises(HaproxyError) as exc_info:
|
|
haproxy_cmd_checked("set server pool_99/pool_99_1 state ready")
|
|
|
|
assert "No such" in str(exc_info.value)
|
|
|
|
def test_error_response_not_found(self, mock_socket_class, mock_select):
|
|
"""Response containing 'not found' raises HaproxyError."""
|
|
mock_sock = mock_socket_class(
|
|
responses={"del map": "Backend not found."}
|
|
)
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
with pytest.raises(HaproxyError) as exc_info:
|
|
haproxy_cmd_checked("del map /path example.com")
|
|
|
|
assert "not found" in str(exc_info.value)
|
|
|
|
def test_error_response_error(self, mock_socket_class, mock_select):
|
|
"""Response containing 'error' raises HaproxyError."""
|
|
mock_sock = mock_socket_class(
|
|
responses={"set server": "error: invalid state"}
|
|
)
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
with pytest.raises(HaproxyError) as exc_info:
|
|
haproxy_cmd_checked("set server pool_1/pool_1_1 state invalid")
|
|
|
|
assert "error" in str(exc_info.value).lower()
|
|
|
|
def test_error_response_failed(self, mock_socket_class, mock_select):
|
|
"""Response containing 'failed' raises HaproxyError."""
|
|
mock_sock = mock_socket_class(
|
|
responses={"set server": "Command failed"}
|
|
)
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
with pytest.raises(HaproxyError) as exc_info:
|
|
haproxy_cmd_checked("set server pool_1/pool_1_1 addr bad")
|
|
|
|
assert "failed" in str(exc_info.value).lower()
|
|
|
|
|
|
class TestHaproxyCmdBatch:
|
|
"""Tests for haproxy_cmd_batch function."""
|
|
|
|
def test_empty_commands(self):
|
|
"""Empty command list returns empty list."""
|
|
result = haproxy_cmd_batch([])
|
|
assert result == []
|
|
|
|
def test_single_command(self, mock_socket_class, mock_select):
|
|
"""Single command uses haproxy_cmd_checked."""
|
|
mock_sock = mock_socket_class(responses={"set server": ""})
|
|
|
|
with patch("socket.socket", return_value=mock_sock):
|
|
result = haproxy_cmd_batch(["set server pool_1/pool_1_1 state ready"])
|
|
|
|
assert len(result) == 1
|
|
|
|
def test_multiple_commands(self, mock_socket_class, mock_select):
|
|
"""Multiple commands are executed separately."""
|
|
# Each command gets its own socket connection
|
|
call_count = 0
|
|
def create_mock_socket(*args, **kwargs):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
return mock_socket_class(responses={"set server": ""})
|
|
|
|
with patch("socket.socket", side_effect=create_mock_socket):
|
|
result = haproxy_cmd_batch([
|
|
"set server pool_1/pool_1_1 addr 10.0.0.1 port 80",
|
|
"set server pool_1/pool_1_1 state ready",
|
|
])
|
|
|
|
assert len(result) == 2
|
|
assert call_count == 2 # One connection per command
|
|
|
|
def test_error_in_batch_raises(self, mock_socket_class, mock_select):
|
|
"""Error in batch command raises immediately."""
|
|
mock_sock = mock_socket_class(
|
|
responses={
|
|
"set server pool_1/pool_1_1 addr": "",
|
|
"set server pool_1/pool_1_1 state": "No such server",
|
|
}
|
|
)
|
|
|
|
call_count = 0
|
|
def create_socket(*args, **kwargs):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
if call_count == 1:
|
|
return mock_socket_class(responses={"set server": ""})
|
|
else:
|
|
return mock_socket_class(responses={"set server": "No such server"})
|
|
|
|
with patch("socket.socket", side_effect=create_socket):
|
|
with pytest.raises(HaproxyError):
|
|
haproxy_cmd_batch([
|
|
"set server pool_1/pool_1_1 addr 10.0.0.1 port 80",
|
|
"set server pool_1/pool_1_1 state ready",
|
|
])
|
|
|
|
|
|
class TestReloadHaproxy:
|
|
"""Tests for reload_haproxy function."""
|
|
|
|
def test_successful_reload(self, mock_subprocess):
|
|
"""Successful reload returns (True, 'OK')."""
|
|
mock_subprocess.return_value = MagicMock(returncode=0, stdout="", stderr="")
|
|
|
|
success, message = reload_haproxy()
|
|
|
|
assert success is True
|
|
assert message == "OK"
|
|
|
|
def test_validation_failure(self, mock_subprocess):
|
|
"""Config validation failure returns (False, error)."""
|
|
mock_subprocess.return_value = MagicMock(
|
|
returncode=1,
|
|
stdout="",
|
|
stderr="[ALERT] Invalid configuration"
|
|
)
|
|
|
|
success, message = reload_haproxy()
|
|
|
|
assert success is False
|
|
assert "validation failed" in message.lower()
|
|
assert "Invalid configuration" in message
|
|
|
|
def test_reload_failure(self, mock_subprocess):
|
|
"""Reload command failure returns (False, error)."""
|
|
# First call (validation) succeeds, second call (reload) fails
|
|
mock_subprocess.side_effect = [
|
|
MagicMock(returncode=0, stdout="", stderr=""),
|
|
MagicMock(returncode=1, stdout="", stderr="Container not found"),
|
|
]
|
|
|
|
success, message = reload_haproxy()
|
|
|
|
assert success is False
|
|
assert "Reload failed" in message
|
|
|
|
def test_podman_not_found(self, mock_subprocess):
|
|
"""Podman not found returns (False, error)."""
|
|
mock_subprocess.side_effect = FileNotFoundError()
|
|
|
|
success, message = reload_haproxy()
|
|
|
|
assert success is False
|
|
assert "podman" in message.lower()
|
|
|
|
def test_subprocess_timeout(self, mock_subprocess):
|
|
"""Subprocess timeout returns (False, error)."""
|
|
import subprocess
|
|
mock_subprocess.side_effect = subprocess.TimeoutExpired("podman", 30)
|
|
|
|
success, message = reload_haproxy()
|
|
|
|
assert success is False
|
|
assert "timed out" in message.lower()
|
|
|
|
def test_os_error(self, mock_subprocess):
|
|
"""OS error returns (False, error)."""
|
|
mock_subprocess.side_effect = OSError("Permission denied")
|
|
|
|
success, message = reload_haproxy()
|
|
|
|
assert success is False
|
|
assert "OS error" in message
|