haproxy-mcp/tests/unit/tools/test_health.py

"""Unit tests for health check tools."""

import json
from unittest.mock import patch, MagicMock

import pytest

from haproxy_mcp.exceptions import HaproxyError


class TestHaproxyHealth:
    """Tests for haproxy_health tool function."""

    def test_health_all_ok(self, mock_socket_class, mock_select, patch_config_paths, response_builder, mock_subprocess):
        """Health check returns healthy when all components are OK."""
        mock_sock = mock_socket_class(responses={
            "show info": response_builder.info(version="3.3.2", uptime=3600),
        })

        mock_subprocess.return_value = MagicMock(
            returncode=0,
            stdout="running"
        )

        with patch("socket.socket", return_value=mock_sock):
            from haproxy_mcp.tools.health import register_health_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_health_tools(mcp)

            result_str = registered_tools["haproxy_health"]()
            result = json.loads(result_str)

        assert result["status"] == "healthy"
        assert result["components"]["mcp"]["status"] == "ok"
        assert result["components"]["haproxy"]["status"] == "ok"
        assert result["components"]["haproxy"]["version"] == "3.3.2"

    def test_health_haproxy_error(self, mock_socket_class, mock_select, patch_config_paths, mock_subprocess):
        """Health check returns degraded when HAProxy is unreachable."""

        def raise_error(*args, **kwargs):
            raise ConnectionRefusedError()

        with patch("socket.socket", side_effect=raise_error):
            from haproxy_mcp.tools.health import register_health_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_health_tools(mcp)

            result_str = registered_tools["haproxy_health"]()
            result = json.loads(result_str)

        assert result["status"] == "degraded"
        assert result["components"]["haproxy"]["status"] == "error"

    def test_health_missing_config_files(self, mock_socket_class, mock_select, tmp_path, response_builder, mock_subprocess):
        """Health check returns degraded when config files are missing."""
        mock_sock = mock_socket_class(responses={
            "show info": response_builder.info(),
        })

        mock_subprocess.return_value = MagicMock(returncode=0, stdout="running")

        # Use paths that don't exist
        with patch("haproxy_mcp.tools.health.MAP_FILE", str(tmp_path / "nonexistent.map")):
            with patch("haproxy_mcp.tools.health.SERVERS_FILE", str(tmp_path / "nonexistent.json")):
                with patch("socket.socket", return_value=mock_sock):
                    from haproxy_mcp.tools.health import register_health_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_health_tools(mcp)

                    result_str = registered_tools["haproxy_health"]()
                    result = json.loads(result_str)

        assert result["status"] == "degraded"
        assert result["components"]["config_files"]["status"] == "warning"

    def test_health_container_not_running(self, mock_socket_class, mock_select, patch_config_paths, response_builder, mock_subprocess):
        """Health check returns unhealthy when container is not running."""
        mock_sock = mock_socket_class(responses={
            "show info": response_builder.info(),
        })

        mock_subprocess.return_value = MagicMock(
            returncode=1,
            stdout="",
            stderr="No such container"
        )

        with patch("socket.socket", return_value=mock_sock):
            from haproxy_mcp.tools.health import register_health_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_health_tools(mcp)

            result_str = registered_tools["haproxy_health"]()
            result = json.loads(result_str)

        assert result["status"] == "unhealthy"
        assert result["components"]["container"]["status"] == "error"


class TestHaproxyDomainHealth:
    """Tests for haproxy_domain_health tool function."""

    def test_domain_health_invalid_domain(self, patch_config_paths):
        """Reject invalid domain format."""
        from haproxy_mcp.tools.health import register_health_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_health_tools(mcp)

        result_str = registered_tools["haproxy_domain_health"](domain="-invalid")
        result = json.loads(result_str)

        assert "error" in result
        assert "Invalid domain" in result["error"]

    def test_domain_health_healthy(self, mock_socket_class, mock_select, patch_config_paths, response_builder):
        """Domain health returns healthy when all servers are UP."""
        with open(patch_config_paths["map_file"], "w") as f:
            f.write("example.com pool_1\n")

        mock_sock = mock_socket_class(responses={
            "show servers state": response_builder.servers_state([
                {"be_name": "pool_1", "srv_name": "pool_1_1", "srv_addr": "10.0.0.1", "srv_port": 80},
                {"be_name": "pool_1", "srv_name": "pool_1_2", "srv_addr": "10.0.0.2", "srv_port": 80},
            ]),
            "show stat": response_builder.stat_csv([
                {"pxname": "pool_1", "svname": "pool_1_1", "status": "UP", "check_status": "L4OK"},
                {"pxname": "pool_1", "svname": "pool_1_2", "status": "UP", "check_status": "L4OK"},
            ]),
        })

        with patch("socket.socket", return_value=mock_sock):
            from haproxy_mcp.tools.health import register_health_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_health_tools(mcp)

            result_str = registered_tools["haproxy_domain_health"](domain="example.com")
            result = json.loads(result_str)

        assert result["status"] == "healthy"
        assert result["healthy_count"] == 2
        assert result["total_count"] == 2

    def test_domain_health_degraded(self, mock_socket_class, mock_select, patch_config_paths, response_builder):
        """Domain health returns degraded when some servers are DOWN."""
        with open(patch_config_paths["map_file"], "w") as f:
            f.write("example.com pool_1\n")

        mock_sock = mock_socket_class(responses={
            "show servers state": response_builder.servers_state([
                {"be_name": "pool_1", "srv_name": "pool_1_1", "srv_addr": "10.0.0.1", "srv_port": 80},
                {"be_name": "pool_1", "srv_name": "pool_1_2", "srv_addr": "10.0.0.2", "srv_port": 80},
            ]),
            "show stat": response_builder.stat_csv([
                {"pxname": "pool_1", "svname": "pool_1_1", "status": "UP"},
                {"pxname": "pool_1", "svname": "pool_1_2", "status": "DOWN"},
            ]),
        })

        with patch("socket.socket", return_value=mock_sock):
            from haproxy_mcp.tools.health import register_health_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_health_tools(mcp)

            result_str = registered_tools["haproxy_domain_health"](domain="example.com")
            result = json.loads(result_str)

        assert result["status"] == "degraded"
        assert result["healthy_count"] == 1
        assert result["total_count"] == 2

    def test_domain_health_down(self, mock_socket_class, mock_select, patch_config_paths, response_builder):
        """Domain health returns down when all servers are DOWN."""
        with open(patch_config_paths["map_file"], "w") as f:
            f.write("example.com pool_1\n")

        mock_sock = mock_socket_class(responses={
            "show servers state": response_builder.servers_state([
                {"be_name": "pool_1", "srv_name": "pool_1_1", "srv_addr": "10.0.0.1", "srv_port": 80},
            ]),
            "show stat": response_builder.stat_csv([
                {"pxname": "pool_1", "svname": "pool_1_1", "status": "DOWN"},
            ]),
        })

        with patch("socket.socket", return_value=mock_sock):
            from haproxy_mcp.tools.health import register_health_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_health_tools(mcp)

            result_str = registered_tools["haproxy_domain_health"](domain="example.com")
            result = json.loads(result_str)

        assert result["status"] == "down"
        assert result["healthy_count"] == 0
        assert result["total_count"] == 1

    def test_domain_health_no_servers(self, mock_socket_class, mock_select, patch_config_paths, response_builder):
        """Domain health returns no_servers when no servers configured."""
        with open(patch_config_paths["map_file"], "w") as f:
            f.write("example.com pool_1\n")

        mock_sock = mock_socket_class(responses={
            "show servers state": response_builder.servers_state([
                {"be_name": "pool_1", "srv_name": "pool_1_1", "srv_addr": "0.0.0.0", "srv_port": 0},
            ]),
            "show stat": response_builder.stat_csv([]),
        })

        with patch("socket.socket", return_value=mock_sock):
            from haproxy_mcp.tools.health import register_health_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_health_tools(mcp)

            result_str = registered_tools["haproxy_domain_health"](domain="example.com")
            result = json.loads(result_str)

        assert result["status"] == "no_servers"
        assert result["total_count"] == 0


class TestHaproxyGetServerHealth:
    """Tests for haproxy_get_server_health tool function."""

    def test_get_server_health_invalid_backend(self, patch_config_paths):
        """Reject invalid backend name."""
        from haproxy_mcp.tools.health import register_health_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_health_tools(mcp)

        result = registered_tools["haproxy_get_server_health"](backend="invalid@name")

        assert "Error" in result
        assert "Invalid backend" in result

    def test_get_server_health_all_backends(self, mock_socket_class, mock_select, patch_config_paths, response_builder):
        """Get health for all backends."""
        mock_sock = mock_socket_class(responses={
            "show stat": response_builder.stat_csv([
                {"pxname": "pool_1", "svname": "pool_1_1", "status": "UP", "weight": 1, "check_status": "L4OK"},
                {"pxname": "pool_2", "svname": "pool_2_1", "status": "DOWN", "weight": 1, "check_status": "L4TOUT"},
            ]),
        })

        with patch("socket.socket", return_value=mock_sock):
            from haproxy_mcp.tools.health import register_health_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_health_tools(mcp)

            result = registered_tools["haproxy_get_server_health"](backend="")

        assert "pool_1" in result
        assert "pool_2" in result
        assert "UP" in result
        assert "DOWN" in result

    def test_get_server_health_filter_backend(self, mock_socket_class, mock_select, patch_config_paths, response_builder):
        """Get health for specific backend."""
        mock_sock = mock_socket_class(responses={
            "show stat": response_builder.stat_csv([
                {"pxname": "pool_1", "svname": "pool_1_1", "status": "UP"},
                {"pxname": "pool_1", "svname": "pool_1_2", "status": "UP"},
                {"pxname": "pool_2", "svname": "pool_2_1", "status": "DOWN"},
            ]),
        })

        with patch("socket.socket", return_value=mock_sock):
            from haproxy_mcp.tools.health import register_health_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_health_tools(mcp)

            result = registered_tools["haproxy_get_server_health"](backend="pool_1")

        assert "pool_1" in result
        assert "pool_2" not in result

    def test_get_server_health_no_servers(self, mock_socket_class, mock_select, patch_config_paths, response_builder):
        """No servers returns appropriate message."""
        mock_sock = mock_socket_class(responses={
            "show stat": response_builder.stat_csv([
                {"pxname": "pool_1", "svname": "FRONTEND", "status": "OPEN"},
                {"pxname": "pool_1", "svname": "BACKEND", "status": "UP"},
            ]),
        })

        with patch("socket.socket", return_value=mock_sock):
            from haproxy_mcp.tools.health import register_health_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_health_tools(mcp)

            result = registered_tools["haproxy_get_server_health"](backend="")

        assert "No servers found" in result

    def test_get_server_health_haproxy_error(self, mock_socket_class, mock_select, patch_config_paths):
        """HAProxy error returns error message."""
        def raise_error(*args, **kwargs):
            raise ConnectionRefusedError()

        with patch("socket.socket", side_effect=raise_error):
            from haproxy_mcp.tools.health import register_health_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_health_tools(mcp)

            result = registered_tools["haproxy_get_server_health"](backend="")

        assert "Error" in result