freecad-robust-mcp-fc111/tests/unit/test_server.py

"""Tests for the main server module."""

import os
import sys
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from freecad_mcp.config import FreecadMode

# Default argv for main() tests to avoid argparse errors
DEFAULT_ARGV: list[str] = ["freecad-mcp"]


class TestGetInstanceId:
    """Tests for get_instance_id function."""

    def test_returns_string(self):
        """Instance ID should be a string."""
        from freecad_mcp.server import get_instance_id

        instance_id = get_instance_id()
        assert isinstance(instance_id, str)

    def test_returns_uuid_format(self):
        """Instance ID should be a valid UUID format."""
        from freecad_mcp.server import get_instance_id

        instance_id = get_instance_id()
        # UUID format: 8-4-4-4-12 hex characters
        parts = instance_id.split("-")
        assert len(parts) == 5
        assert len(parts[0]) == 8
        assert len(parts[1]) == 4
        assert len(parts[2]) == 4
        assert len(parts[3]) == 4
        assert len(parts[4]) == 12

    def test_consistent_across_calls(self):
        """Instance ID should be consistent within a process."""
        from freecad_mcp.server import get_instance_id

        id1 = get_instance_id()
        id2 = get_instance_id()
        assert id1 == id2


class TestGetBridge:
    """Tests for get_bridge function."""

    @pytest.mark.asyncio
    async def test_raises_when_not_initialized(self):
        """Should raise RuntimeError when bridge is not initialized."""
        import freecad_mcp.server as server_module

        # Save original bridge
        original_bridge = server_module._bridge

        try:
            # Set bridge to None
            server_module._bridge = None

            with pytest.raises(RuntimeError, match="not initialized"):
                await server_module.get_bridge()
        finally:
            # Restore original bridge
            server_module._bridge = original_bridge

    @pytest.mark.asyncio
    async def test_returns_bridge_when_initialized(self):
        """Should return bridge when it's initialized."""
        import freecad_mcp.server as server_module

        # Save original bridge
        original_bridge = server_module._bridge

        try:
            # Set up mock bridge
            mock_bridge = MagicMock()
            server_module._bridge = mock_bridge

            bridge = await server_module.get_bridge()
            assert bridge is mock_bridge
        finally:
            # Restore original bridge
            server_module._bridge = original_bridge


class TestLifespan:
    """Tests for the lifespan context manager."""

    @pytest.mark.asyncio
    async def test_embedded_mode_initialization(self):
        """Should initialize embedded bridge in embedded mode."""
        import freecad_mcp.server as server_module

        mock_config = MagicMock()
        mock_config.mode = FreecadMode.EMBEDDED
        mock_config.freecad_path = None

        mock_embedded_bridge = AsyncMock()
        mock_embedded_bridge.get_freecad_version = AsyncMock(
            return_value={"version": "1.0.0", "gui_available": False}
        )

        with (
            patch.object(server_module, "get_config", return_value=mock_config),
            patch(
                "freecad_mcp.bridge.embedded.EmbeddedBridge",
                return_value=mock_embedded_bridge,
            ) as mock_embedded_class,
        ):
            mock_server = MagicMock()

            async with server_module.lifespan(mock_server):
                # Bridge should be initialized
                mock_embedded_class.assert_called_once_with(freecad_path=None)
                mock_embedded_bridge.connect.assert_called_once()

            # After exiting, disconnect should be called
            mock_embedded_bridge.disconnect.assert_called_once()

    @pytest.mark.asyncio
    async def test_xmlrpc_mode_initialization(self):
        """Should initialize XML-RPC bridge in xmlrpc mode."""
        import freecad_mcp.server as server_module

        mock_config = MagicMock()
        mock_config.mode = FreecadMode.XMLRPC
        mock_config.socket_host = "localhost"
        mock_config.xmlrpc_port = 9875

        mock_xmlrpc_bridge = AsyncMock()
        mock_xmlrpc_bridge.get_freecad_version = AsyncMock(
            return_value={"version": "1.0.0", "gui_available": True}
        )

        with (
            patch.object(server_module, "get_config", return_value=mock_config),
            patch(
                "freecad_mcp.bridge.xmlrpc.XmlRpcBridge",
                return_value=mock_xmlrpc_bridge,
            ) as mock_xmlrpc_class,
        ):
            mock_server = MagicMock()

            async with server_module.lifespan(mock_server):
                mock_xmlrpc_class.assert_called_once_with(host="localhost", port=9875)
                mock_xmlrpc_bridge.connect.assert_called_once()

            mock_xmlrpc_bridge.disconnect.assert_called_once()

    @pytest.mark.asyncio
    async def test_socket_mode_initialization(self):
        """Should initialize socket bridge in socket mode."""
        import freecad_mcp.server as server_module

        mock_config = MagicMock()
        mock_config.mode = FreecadMode.SOCKET
        mock_config.socket_host = "localhost"
        mock_config.socket_port = 9876

        mock_socket_bridge = AsyncMock()
        mock_socket_bridge.get_freecad_version = AsyncMock(
            return_value={"version": "1.0.0", "gui_available": True}
        )

        with (
            patch.object(server_module, "get_config", return_value=mock_config),
            patch(
                "freecad_mcp.bridge.socket.SocketBridge",
                return_value=mock_socket_bridge,
            ) as mock_socket_class,
        ):
            mock_server = MagicMock()

            async with server_module.lifespan(mock_server):
                mock_socket_class.assert_called_once_with(host="localhost", port=9876)
                mock_socket_bridge.connect.assert_called_once()

            mock_socket_bridge.disconnect.assert_called_once()

    @pytest.mark.asyncio
    async def test_version_fetch_failure_logs_warning(self):
        """Should log warning if version fetch fails."""
        import freecad_mcp.server as server_module

        mock_config = MagicMock()
        mock_config.mode = FreecadMode.EMBEDDED
        mock_config.freecad_path = None

        mock_bridge = AsyncMock()
        mock_bridge.get_freecad_version = AsyncMock(
            side_effect=Exception("Connection failed")
        )

        with (
            patch.object(server_module, "get_config", return_value=mock_config),
            patch(
                "freecad_mcp.bridge.embedded.EmbeddedBridge",
                return_value=mock_bridge,
            ),
            patch.object(server_module.logger, "warning") as mock_warning,
        ):
            mock_server = MagicMock()

            async with server_module.lifespan(mock_server):
                # Warning should be logged
                mock_warning.assert_called_once()
                assert "Could not get FreeCAD version" in str(mock_warning.call_args)


class TestRegisterAllComponents:
    """Tests for register_all_components function."""

    def test_registers_tools(self):
        """Should register all tool categories."""
        from freecad_mcp.server import mcp

        # The function is called at module load, but we can verify
        # that the mcp instance exists and has tools registered
        assert mcp is not None
        assert mcp.name == "freecad-mcp"


class TestMain:
    """Tests for main function."""

    def test_main_prints_instance_id(self):
        """Main should print instance ID on startup when FREECAD_MCP_TESTING is set."""
        import freecad_mcp.server as server_module
        from freecad_mcp.config import TransportType

        mock_config = MagicMock()
        mock_config.log_level = "INFO"
        mock_config.mode = FreecadMode.EMBEDDED
        mock_config.transport = TransportType.STDIO

        with (
            patch.object(sys, "argv", DEFAULT_ARGV),
            patch.object(server_module, "get_config", return_value=mock_config),
            patch.object(server_module.mcp, "run") as mock_run,
            patch("builtins.print") as mock_print,
            patch.dict(os.environ, {"FREECAD_MCP_TESTING": "1"}),
        ):
            # Mock run to exit immediately
            mock_run.return_value = None

            server_module.main()

            # Check that instance ID was printed to stderr (not stdout, to avoid
            # corrupting JSON-RPC in stdio mode)
            print_calls = [str(call) for call in mock_print.call_args_list]
            assert any("FREECAD_MCP_INSTANCE_ID=" in call for call in print_calls)
            # Verify it was printed to stderr
            instance_id_call = next(
                call
                for call in mock_print.call_args_list
                if "FREECAD_MCP_INSTANCE_ID=" in str(call)
            )
            assert instance_id_call.kwargs.get("file") == sys.stderr

    def test_main_no_instance_id_without_testing_env(self):
        """Main should NOT print instance ID when FREECAD_MCP_TESTING is unset."""
        import freecad_mcp.server as server_module
        from freecad_mcp.config import TransportType

        mock_config = MagicMock()
        mock_config.log_level = "INFO"
        mock_config.mode = FreecadMode.EMBEDDED
        mock_config.transport = TransportType.STDIO

        # Ensure FREECAD_MCP_TESTING is not set
        env_without_testing = {
            k: v for k, v in os.environ.items() if k != "FREECAD_MCP_TESTING"
        }

        with (
            patch.object(sys, "argv", DEFAULT_ARGV),
            patch.object(server_module, "get_config", return_value=mock_config),
            patch.object(server_module.mcp, "run") as mock_run,
            patch("builtins.print") as mock_print,
            patch.dict(os.environ, env_without_testing, clear=True),
        ):
            # Mock run to exit immediately
            mock_run.return_value = None

            server_module.main()

            # Check that instance ID was NOT printed
            print_calls = [str(call) for call in mock_print.call_args_list]
            assert not any("FREECAD_MCP_INSTANCE_ID=" in call for call in print_calls)

    def test_main_http_transport(self):
        """Main should start HTTP transport when configured."""
        import freecad_mcp.server as server_module
        from freecad_mcp.config import TransportType

        mock_config = MagicMock()
        mock_config.log_level = "INFO"
        mock_config.mode = FreecadMode.EMBEDDED
        mock_config.transport = TransportType.HTTP
        mock_config.http_port = 8080

        with (
            patch.object(sys, "argv", DEFAULT_ARGV),
            patch.object(server_module, "get_config", return_value=mock_config),
            patch.object(server_module.mcp, "run") as mock_run,
            patch("builtins.print"),
        ):
            server_module.main()

            # Should call run with HTTP transport settings
            mock_run.assert_called_once()
            call_kwargs = mock_run.call_args.kwargs
            assert call_kwargs.get("transport") == "streamable-http"
            assert call_kwargs.get("port") == 8080

    def test_main_stdio_transport(self):
        """Main should start stdio transport by default."""
        import freecad_mcp.server as server_module
        from freecad_mcp.config import TransportType

        mock_config = MagicMock()
        mock_config.log_level = "INFO"
        mock_config.mode = FreecadMode.EMBEDDED
        mock_config.transport = TransportType.STDIO

        with (
            patch.object(sys, "argv", DEFAULT_ARGV),
            patch.object(server_module, "get_config", return_value=mock_config),
            patch.object(server_module.mcp, "run") as mock_run,
            patch("builtins.print"),
        ):
            server_module.main()

            # Should call run without transport arguments (stdio is default)
            mock_run.assert_called_once_with()


class TestStdioProtocolCleanliness:
    """Tests to ensure stdio mode produces clean JSON-RPC output.

    These tests verify that stdout contains ONLY valid JSON-RPC messages,
    with no debug output, print statements, or other text that would corrupt
    the MCP protocol. This is critical for compatibility with MCP clients
    like Claude Desktop.

    The bug this catches: Any print() to stdout (instead of stderr) will
    cause MCP clients to fail with JSON parse errors like:
        "Unexpected token 'F', "FREECAD_MC"... is not valid JSON"
    """

    def test_no_stdout_before_jsonrpc(self):
        """Verify no stray output appears on stdout before JSON-RPC messages.

        This test spawns the MCP server as a subprocess and validates that
        ALL stdout output is valid JSON-RPC. Any non-JSON output on stdout
        will corrupt the MCP protocol.
        """
        import json
        import os
        import subprocess
        import time

        # Start the MCP server process
        # Use a non-existent FreeCAD host so it won't actually connect
        proc = subprocess.Popen(  # noqa: S603
            [
                sys.executable,
                "-m",
                "freecad_mcp.server",
            ],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            env={
                **dict(os.environ),
                "FREECAD_MODE": "xmlrpc",
                "FREECAD_XMLRPC_PORT": "59999",  # Non-existent port
                "FREECAD_SOCKET_HOST": "localhost",
            },
        )

        try:
            # Ensure pipes are available
            assert proc.stdin is not None
            assert proc.stdout is not None

            # Send a minimal MCP initialize request
            init_request = {
                "jsonrpc": "2.0",
                "id": 0,
                "method": "initialize",
                "params": {
                    "protocolVersion": "2024-11-05",
                    "capabilities": {},
                    "clientInfo": {"name": "test", "version": "1.0.0"},
                },
            }
            request_bytes = json.dumps(init_request).encode() + b"\n"
            proc.stdin.write(request_bytes)
            proc.stdin.flush()

            # Give the server a moment to respond
            time.sleep(0.5)

            # Set stdout to non-blocking mode
            os.set_blocking(proc.stdout.fileno(), False)

            # Read any available stdout
            stdout_data = b""
            try:
                while True:
                    chunk = proc.stdout.read(4096)
                    if not chunk:
                        break
                    stdout_data += chunk
            except (BlockingIOError, TypeError):
                pass  # No more data available

            # Validate that ALL stdout is valid JSON-RPC
            # Each line should be a valid JSON object
            stdout_text = stdout_data.decode("utf-8", errors="replace")
            lines = [line.strip() for line in stdout_text.split("\n") if line.strip()]

            for line in lines:
                try:
                    parsed = json.loads(line)
                    # Should be a JSON-RPC message (has jsonrpc field)
                    assert "jsonrpc" in parsed, (
                        f"stdout contains JSON but not JSON-RPC: {line[:100]}"
                    )
                except json.JSONDecodeError as e:
                    pytest.fail(
                        f"stdout contains non-JSON output which corrupts MCP protocol!\n"
                        f"Invalid line: {line[:200]!r}\n"
                        f"JSON error: {e}\n\n"
                        f"All stdout lines:\n{stdout_text[:1000]}"
                    )

        finally:
            # Clean up the process
            proc.terminate()
            try:
                proc.wait(timeout=2)
            except subprocess.TimeoutExpired:
                proc.kill()

    def test_instance_id_on_stderr_not_stdout(self) -> None:
        """Verify FREECAD_MCP_INSTANCE_ID is printed to stderr, not stdout.

        The instance ID must go to stderr because stdout is reserved for
        JSON-RPC messages in stdio mode.
        """
        import os
        import subprocess
        import time

        proc = subprocess.Popen(  # noqa: S603
            [
                sys.executable,
                "-m",
                "freecad_mcp.server",
            ],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            env={
                **dict(os.environ),
                "FREECAD_MODE": "xmlrpc",
                "FREECAD_XMLRPC_PORT": "59999",
                "FREECAD_SOCKET_HOST": "localhost",
                "FREECAD_MCP_TESTING": "1",  # Enable stderr instance ID output
            },
        )

        try:
            # Ensure pipes are available
            assert proc.stdout is not None
            assert proc.stderr is not None

            # Set pipes to non-blocking mode
            os.set_blocking(proc.stdout.fileno(), False)
            os.set_blocking(proc.stderr.fileno(), False)

            # Poll for stderr content with timeout (CI systems can be slower)
            stderr_data = b""
            max_wait = 5.0  # 5 second timeout
            poll_interval = 0.1
            elapsed = 0.0

            while elapsed < max_wait:
                try:
                    chunk = proc.stderr.read(4096)
                    if chunk:
                        stderr_data += chunk
                        # Check if we got the instance ID
                        if b"FREECAD_MCP_INSTANCE_ID=" in stderr_data:
                            break
                except (BlockingIOError, TypeError):
                    pass  # No data available yet

                time.sleep(poll_interval)
                elapsed += poll_interval

            stderr_text = stderr_data.decode("utf-8", errors="replace")

            # Instance ID should be in stderr
            assert "FREECAD_MCP_INSTANCE_ID=" in stderr_text, (
                f"Instance ID not found in stderr after {max_wait}s.\n"
                f"stderr: {stderr_text[:500]}"
            )

            # Read stdout (should NOT contain instance ID)
            stdout_data = b""
            try:
                while True:
                    chunk = proc.stdout.read(4096)
                    if not chunk:
                        break
                    stdout_data += chunk
            except (BlockingIOError, TypeError):
                pass  # No more data available

            stdout_text = stdout_data.decode("utf-8", errors="replace")

            # Instance ID should NOT be in stdout
            assert "FREECAD_MCP_INSTANCE_ID=" not in stdout_text, (
                f"Instance ID incorrectly appears in stdout, corrupting MCP protocol!\n"
                f"stdout: {stdout_text[:500]}"
            )

        finally:
            proc.terminate()
            try:
                proc.wait(timeout=2)
            except subprocess.TimeoutExpired:
                proc.kill()