324 lines
11 KiB
Python
324 lines
11 KiB
Python
"""
|
|
Tests for Podman client wrapper.
|
|
|
|
Following TDD approach - these tests are written before implementation.
|
|
Tests cover all requirements from todo.md section 1.3.1.
|
|
All tests use mocked Podman client (no actual Podman needed).
|
|
"""
|
|
|
|
import pytest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, MagicMock, patch
|
|
|
|
|
|
def test_connection_to_podman_socket_succeeds(tmp_path):
|
|
"""Test that connection to Podman socket succeeds."""
|
|
from mcp_forge.podman.client import PodmanClient
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
socket_path = tmp_path / "podman.sock"
|
|
socket_path.touch() # Create fake socket file
|
|
|
|
audit_logger = AuditLogger(tmp_path / "audit.log")
|
|
validator = OperationValidator(SecurityConfig(audit_log=tmp_path / "audit.log"))
|
|
|
|
client = PodmanClient(
|
|
socket_path=socket_path,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
with patch('mcp_forge.podman.client.BasePodmanClient') as mock_podman:
|
|
mock_client_instance = MagicMock()
|
|
mock_client_instance.ping.return_value = "OK"
|
|
mock_podman.return_value = mock_client_instance
|
|
|
|
client.connect()
|
|
assert client._client is not None
|
|
mock_podman.assert_called_once_with(base_url=f"unix://{socket_path}")
|
|
|
|
|
|
def test_connection_failure_raises_clear_error(tmp_path):
|
|
"""Test that connection failure raises clear error."""
|
|
from mcp_forge.podman.client import PodmanClient, PodmanConnectionError
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
socket_path = tmp_path / "nonexistent.sock"
|
|
|
|
audit_logger = AuditLogger(tmp_path / "audit.log")
|
|
validator = OperationValidator(SecurityConfig(audit_log=tmp_path / "audit.log"))
|
|
|
|
client = PodmanClient(
|
|
socket_path=socket_path,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
with patch('mcp_forge.podman.client.BasePodmanClient', side_effect=Exception("Connection failed")):
|
|
with pytest.raises(PodmanConnectionError) as exc_info:
|
|
client.connect()
|
|
assert "connection" in str(exc_info.value).lower() or "failed" in str(exc_info.value).lower()
|
|
|
|
|
|
def test_socket_path_validation(tmp_path):
|
|
"""Test that socket path is validated before connecting."""
|
|
from mcp_forge.podman.client import PodmanClient, PodmanConnectionError
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
nonexistent_socket = tmp_path / "nonexistent.sock"
|
|
|
|
audit_logger = AuditLogger(tmp_path / "audit.log")
|
|
validator = OperationValidator(SecurityConfig(audit_log=tmp_path / "audit.log"))
|
|
|
|
client = PodmanClient(
|
|
socket_path=nonexistent_socket,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
with pytest.raises(PodmanConnectionError) as exc_info:
|
|
client.verify_socket_access()
|
|
assert "not found" in str(exc_info.value).lower() or "does not exist" in str(exc_info.value).lower()
|
|
|
|
|
|
def test_socket_permissions_check(tmp_path):
|
|
"""Test that socket permissions are checked."""
|
|
from mcp_forge.podman.client import PodmanClient
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
socket_path = tmp_path / "podman.sock"
|
|
socket_path.touch()
|
|
socket_path.chmod(0o000) # Remove all permissions
|
|
|
|
audit_logger = AuditLogger(tmp_path / "audit.log")
|
|
validator = OperationValidator(SecurityConfig(audit_log=tmp_path / "audit.log"))
|
|
|
|
client = PodmanClient(
|
|
socket_path=socket_path,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
# verify_socket_access should check readability
|
|
# Depending on implementation, might raise error or just warn
|
|
try:
|
|
client.verify_socket_access()
|
|
except Exception as e:
|
|
# Should mention permissions or access
|
|
assert "permission" in str(e).lower() or "access" in str(e).lower() or "readable" in str(e).lower()
|
|
finally:
|
|
socket_path.chmod(0o644) # Restore for cleanup
|
|
|
|
|
|
def test_api_version_compatibility_check(tmp_path):
|
|
"""Test that API version is checked."""
|
|
from mcp_forge.podman.client import PodmanClient
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
socket_path = tmp_path / "podman.sock"
|
|
socket_path.touch()
|
|
|
|
audit_logger = AuditLogger(tmp_path / "audit.log")
|
|
validator = OperationValidator(SecurityConfig(audit_log=tmp_path / "audit.log"))
|
|
|
|
client = PodmanClient(
|
|
socket_path=socket_path,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
with patch('mcp_forge.podman.client.BasePodmanClient') as mock_podman:
|
|
mock_client = MagicMock()
|
|
mock_client.version.return_value = {
|
|
"Version": "4.5.0",
|
|
"ApiVersion": "4.5.0"
|
|
}
|
|
mock_client.ping.return_value = "OK"
|
|
mock_podman.return_value = mock_client
|
|
|
|
client.connect()
|
|
version_info = client.check_api_version()
|
|
|
|
assert "Version" in version_info or "ApiVersion" in version_info
|
|
|
|
|
|
def test_ping_health_check(tmp_path):
|
|
"""Test that ping/health check works."""
|
|
from mcp_forge.podman.client import PodmanClient
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
socket_path = tmp_path / "podman.sock"
|
|
socket_path.touch()
|
|
|
|
audit_logger = AuditLogger(tmp_path / "audit.log")
|
|
validator = OperationValidator(SecurityConfig(audit_log=tmp_path / "audit.log"))
|
|
|
|
client = PodmanClient(
|
|
socket_path=socket_path,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
with patch('mcp_forge.podman.client.BasePodmanClient') as mock_podman:
|
|
mock_client = MagicMock()
|
|
mock_client.ping.return_value = "OK"
|
|
mock_podman.return_value = mock_client
|
|
|
|
client.connect()
|
|
result = client.ping()
|
|
|
|
assert result is True or result == "OK"
|
|
|
|
|
|
def test_lazy_connection(tmp_path):
|
|
"""Test that connection is lazy (only connects when needed)."""
|
|
from mcp_forge.podman.client import PodmanClient
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
socket_path = tmp_path / "podman.sock"
|
|
socket_path.touch()
|
|
|
|
audit_logger = AuditLogger(tmp_path / "audit.log")
|
|
validator = OperationValidator(SecurityConfig(audit_log=tmp_path / "audit.log"))
|
|
|
|
# Creating client should not connect
|
|
client = PodmanClient(
|
|
socket_path=socket_path,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
assert client._client is None # Not connected yet
|
|
|
|
# Accessing client property should trigger connection
|
|
with patch('mcp_forge.podman.client.BasePodmanClient') as mock_podman:
|
|
mock_client_instance = MagicMock()
|
|
mock_client_instance.ping.return_value = "OK"
|
|
mock_podman.return_value = mock_client_instance
|
|
_ = client.client
|
|
assert client._client is not None
|
|
|
|
|
|
def test_disconnect_cleanup(tmp_path):
|
|
"""Test that disconnect cleans up properly."""
|
|
from mcp_forge.podman.client import PodmanClient
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
socket_path = tmp_path / "podman.sock"
|
|
socket_path.touch()
|
|
|
|
audit_logger = AuditLogger(tmp_path / "audit.log")
|
|
validator = OperationValidator(SecurityConfig(audit_log=tmp_path / "audit.log"))
|
|
|
|
client = PodmanClient(
|
|
socket_path=socket_path,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
with patch('mcp_forge.podman.client.BasePodmanClient') as mock_podman:
|
|
mock_client = MagicMock()
|
|
mock_client.ping.return_value = "OK"
|
|
mock_podman.return_value = mock_client
|
|
|
|
client.connect()
|
|
assert client._client is not None
|
|
|
|
client.disconnect()
|
|
assert client._client is None
|
|
|
|
|
|
def test_connection_error_includes_socket_path(tmp_path):
|
|
"""Test that connection errors include the socket path for debugging."""
|
|
from mcp_forge.podman.client import PodmanClient, PodmanConnectionError
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
socket_path = tmp_path / "test.sock"
|
|
|
|
audit_logger = AuditLogger(tmp_path / "audit.log")
|
|
validator = OperationValidator(SecurityConfig(audit_log=tmp_path / "audit.log"))
|
|
|
|
client = PodmanClient(
|
|
socket_path=socket_path,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
with pytest.raises(PodmanConnectionError) as exc_info:
|
|
client.verify_socket_access()
|
|
|
|
assert str(socket_path) in str(exc_info.value) or socket_path.name in str(exc_info.value)
|
|
|
|
|
|
def test_client_property_auto_connects(tmp_path):
|
|
"""Test that accessing client property auto-connects if not connected."""
|
|
from mcp_forge.podman.client import PodmanClient
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
socket_path = tmp_path / "podman.sock"
|
|
socket_path.touch()
|
|
|
|
audit_logger = AuditLogger(tmp_path / "audit.log")
|
|
validator = OperationValidator(SecurityConfig(audit_log=tmp_path / "audit.log"))
|
|
|
|
client = PodmanClient(
|
|
socket_path=socket_path,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
with patch('mcp_forge.podman.client.BasePodmanClient') as mock_podman:
|
|
mock_client = MagicMock()
|
|
mock_client.ping.return_value = "OK"
|
|
mock_podman.return_value = mock_client
|
|
|
|
# First access should trigger connect
|
|
_ = client.client
|
|
assert mock_podman.called
|
|
|
|
# Second access should reuse connection
|
|
mock_podman.reset_mock()
|
|
_ = client.client
|
|
assert not mock_podman.called # Should not connect again
|
|
|
|
|
|
def test_validator_and_audit_logger_stored(tmp_path):
|
|
"""Test that validator and audit logger are stored for later use."""
|
|
from mcp_forge.podman.client import PodmanClient
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
socket_path = tmp_path / "podman.sock"
|
|
|
|
audit_logger = AuditLogger(tmp_path / "audit.log")
|
|
validator = OperationValidator(SecurityConfig(audit_log=tmp_path / "audit.log"))
|
|
|
|
client = PodmanClient(
|
|
socket_path=socket_path,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
assert client.validator is validator
|
|
assert client.audit_logger is audit_logger
|