CmdForge/tests/test_mcp.py

"""Tests for MCP client and McpStep execution."""

import sys
import subprocess
import time
from pathlib import Path

import pytest
import yaml
from unittest.mock import patch
from cmdforge.mcp_client import (
    DEFAULT_INHERITED_ENV,
    McpServerConfig,
    McpClientManager,
    _build_server_env,
    _build_http_headers,
    _normalize_result,
    _sanitize,
    _fingerprint,
    load_mcp_config,
    save_mcp_config,
)
from cmdforge.runner import _substitute_mcp_args, _deep_substitute, run_tool
from cmdforge.tool import McpStep, Tool


class TestAgentHostSetup:
    def test_builds_current_vendor_cli_commands(self):
        from cmdforge.agent_setup import build_host_add_command

        codex = build_host_add_command(
            "codex", "/usr/bin/codex", "cmdforge", "user"
        )
        assert codex[:5] == [
            "/usr/bin/codex", "mcp", "add", "cmdforge", "--"
        ]
        assert codex[-4:] == ["-m", "cmdforge.cli", "mcp", "serve"]

        claude = build_host_add_command(
            "claude-code", "/usr/bin/claude", "cmdforge", "project"
        )
        assert claude[:9] == [
            "/usr/bin/claude", "mcp", "add", "--transport", "stdio",
            "--scope", "project", "cmdforge", "--",
        ]

    def test_codex_rejects_unsupported_scope(self):
        from cmdforge.agent_setup import build_host_add_command

        with pytest.raises(ValueError, match="user scope only"):
            build_host_add_command(
                "codex", "/usr/bin/codex", "cmdforge", "project"
            )

    def test_rejects_unsafe_registration_name(self):
        from cmdforge.agent_setup import build_host_add_command

        with pytest.raises(ValueError, match="registration name"):
            build_host_add_command(
                "codex", "/usr/bin/codex", "--config", "user"
            )

    def test_policy_merge_is_idempotent_and_preserves_user_text(self):
        from cmdforge.agent_setup import POLICY_BODY, merge_agent_policy

        original = "# Existing guidance\n\nKeep this.\n"
        merged = merge_agent_policy(original)
        assert merged.startswith(original)
        assert POLICY_BODY.strip() in merged
        assert merge_agent_policy(merged) == merged

    def test_policy_merge_rejects_malformed_markers(self):
        from cmdforge.agent_setup import POLICY_START, merge_agent_policy

        with pytest.raises(ValueError, match="malformed"):
            merge_agent_policy(f"text\n{POLICY_START}\n")

    def test_policy_writer_refuses_symlink(self, tmp_path):
        from cmdforge.agent_setup import write_agent_policy

        target = tmp_path / "target.md"
        target.write_text("unchanged")
        link = tmp_path / "AGENTS.md"
        link.symlink_to(target)
        with pytest.raises(ValueError, match="symlinked"):
            write_agent_policy(link, "replacement")
        assert target.read_text() == "unchanged"

    def test_dry_run_changes_nothing(self, tmp_path, capsys):
        from cmdforge.agent_setup import configure_agent_host

        with (
            patch("cmdforge.agent_setup.shutil.which", return_value="/usr/bin/codex"),
            patch("cmdforge.agent_setup.subprocess.run") as run,
        ):
            assert configure_agent_host(
                "codex", project_dir=tmp_path, dry_run=True
            ) == 0
        run.assert_not_called()
        assert not (tmp_path / "AGENTS.md").exists()
        output = capsys.readouterr().out
        assert "Would run:" in output
        assert "cmdforge-agent-policy:start" in output

    def test_configures_host_and_managed_policy(self, tmp_path, capsys):
        from cmdforge.agent_setup import configure_agent_host

        results = [
            subprocess.CompletedProcess([], 1, "", "not found"),
            subprocess.CompletedProcess([], 0, "added", ""),
        ]
        with (
            patch("cmdforge.agent_setup.shutil.which", return_value="/usr/bin/codex"),
            patch("cmdforge.agent_setup.subprocess.run", side_effect=results) as run,
            patch("cmdforge.agent_setup._warn_if_no_tools_exposed"),
        ):
            assert configure_agent_host("codex", project_dir=tmp_path) == 0

        assert run.call_count == 2
        assert run.call_args_list[0].args[0] == [
            "/usr/bin/codex", "mcp", "get", "cmdforge"
        ]
        assert run.call_args_list[1].args[0][:5] == [
            "/usr/bin/codex", "mcp", "add", "cmdforge", "--"
        ]
        policy = (tmp_path / "AGENTS.md").read_text()
        assert "cmdforge-agent-policy:start" in policy
        assert "Do not install, publish, expose" in policy
        assert "Configured CmdForge MCP" in capsys.readouterr().out

    def test_existing_registration_is_not_overwritten(self, tmp_path):
        from cmdforge.agent_setup import configure_agent_host

        result = subprocess.CompletedProcess([], 0, "exists", "")
        with (
            patch("cmdforge.agent_setup.shutil.which", return_value="/usr/bin/claude"),
            patch("cmdforge.agent_setup.subprocess.run", return_value=result) as run,
            patch("cmdforge.agent_setup._warn_if_no_tools_exposed"),
        ):
            assert configure_agent_host(
                "claude-code", project_dir=tmp_path
            ) == 0
        assert run.call_count == 1
        assert (tmp_path / "CLAUDE.md").exists()

    def test_force_removes_then_readds_registration(self, tmp_path):
        from cmdforge.agent_setup import configure_agent_host

        results = [
            subprocess.CompletedProcess([], 0, "exists", ""),
            subprocess.CompletedProcess([], 0, "removed", ""),
            subprocess.CompletedProcess([], 0, "added", ""),
        ]
        with (
            patch("cmdforge.agent_setup.shutil.which", return_value="/usr/bin/claude"),
            patch("cmdforge.agent_setup.subprocess.run", side_effect=results) as run,
            patch("cmdforge.agent_setup._warn_if_no_tools_exposed"),
        ):
            assert configure_agent_host(
                "claude-code", project_dir=tmp_path,
                install_policy=False, force=True,
            ) == 0

        assert run.call_args_list[1].args[0] == [
            "/usr/bin/claude", "mcp", "remove", "cmdforge"
        ]
        assert run.call_args_list[2].args[0][:4] == [
            "/usr/bin/claude", "mcp", "add", "--transport"
        ]

    def test_failed_host_add_does_not_write_policy(self, tmp_path):
        from cmdforge.agent_setup import configure_agent_host

        results = [
            subprocess.CompletedProcess([], 1, "", "missing"),
            subprocess.CompletedProcess([], 1, "", "bad config"),
        ]
        with (
            patch("cmdforge.agent_setup.shutil.which", return_value="/usr/bin/codex"),
            patch("cmdforge.agent_setup.subprocess.run", side_effect=results),
        ):
            assert configure_agent_host(
                "codex", project_dir=tmp_path
            ) == 1

        assert not (tmp_path / "AGENTS.md").exists()

    def test_host_command_timeout_is_reported(self, tmp_path, capsys):
        from cmdforge.agent_setup import configure_agent_host

        with (
            patch("cmdforge.agent_setup.shutil.which", return_value="/usr/bin/codex"),
            patch(
                "cmdforge.agent_setup.subprocess.run",
                side_effect=subprocess.TimeoutExpired("codex", 30),
            ),
        ):
            assert configure_agent_host(
                "codex", project_dir=tmp_path, install_policy=False
            ) == 1

        assert "timed out after 30 seconds" in capsys.readouterr().err


class TestMcpServerConfig:
    def test_defaults(self):
        cfg = McpServerConfig(name="test")
        assert cfg.transport == "stdio"
        assert cfg.timeout == 30
        assert cfg.args == []
        assert cfg.env == {}
        assert cfg.approved is False

    @pytest.mark.parametrize(
        "change, message",
        [
            ({"transport": "streamable-http"}, "requires a url"),
            ({"args": "-y"}, "list of strings"),
            ({"timeout": 0}, "greater than 0"),
            ({"timeout": True}, "must be a number"),
        ],
    )
    def test_validation_rejects_invalid_config(self, change, message):
        command = None if change.get("transport") == "streamable-http" else "server"
        cfg = McpServerConfig(name="test", command=command, **change)
        with pytest.raises(ValueError, match=message):
            cfg.validate()

    def test_fingerprint_changes_with_command(self):
        a = McpServerConfig(name="a", command="cmd-a")
        b = McpServerConfig(name="b", command="cmd-b")
        assert _fingerprint(a) != _fingerprint(b)

    def test_fingerprint_same_for_same_config(self):
        a = McpServerConfig(name="a", command="cmd", args=["-v"])
        b = McpServerConfig(name="b", command="cmd", args=["-v"])
        assert _fingerprint(a) == _fingerprint(b)

    def test_streamable_http_accepts_https(self):
        cfg = McpServerConfig(
            name="remote", transport="streamable-http",
            url="https://example.com/mcp", approved=True,
        )
        cfg.validate()

    def test_streamable_http_allows_loopback_http_only(self):
        McpServerConfig(
            name="local", transport="streamable-http",
            url="http://127.0.0.1:8000/mcp",
        ).validate()
        with pytest.raises(ValueError, match="must use HTTPS"):
            McpServerConfig(
                name="remote", transport="streamable-http",
                url="http://example.com/mcp",
            ).validate()

    def test_streamable_http_rejects_url_credentials(self):
        with pytest.raises(ValueError, match="must not contain credentials"):
            McpServerConfig(
                name="remote", transport="streamable-http",
                url="https://user:pass@example.com/mcp",
            ).validate()


class TestSanitize:
    def test_redacts_bearer(self):
        assert "redacted" in _sanitize("Bearer abcdefghijklmnopqrstuvwxyz")

    def test_redacts_sk_prefix(self):
        assert "redacted" in _sanitize("sk-abcdefghijklmnopqrstuvwxyz")

    def test_preserves_normal_text(self):
        assert _sanitize("hello world") == "hello world"

    def test_redacts_explicit_secret(self):
        assert _sanitize("failed with hunter2", ["hunter2"]) == "failed with [redacted]"


class TestEnvironmentIsolation:
    def test_only_allowlisted_environment_is_inherited(self, monkeypatch):
        monkeypatch.setenv("CMDFORGE_TEST_SECRET", "do-not-leak")
        cfg = McpServerConfig(name="test", command="server")
        environment = _build_server_env(cfg)
        assert "CMDFORGE_TEST_SECRET" not in environment
        assert set(environment).issubset(set(DEFAULT_INHERITED_ENV))

    def test_explicit_environment_reference_is_resolved(self, monkeypatch):
        monkeypatch.setenv("CMDFORGE_TEST_SECRET", "allowed")
        cfg = McpServerConfig(
            name="test", command="server", env={"SERVER_TOKEN": "${CMDFORGE_TEST_SECRET}"}
        )
        assert _build_server_env(cfg)["SERVER_TOKEN"] == "allowed"

    def test_http_header_reference_is_resolved(self, monkeypatch):
        monkeypatch.setenv("MCP_TOKEN", "secret")
        cfg = McpServerConfig(
            name="remote", transport="streamable-http",
            url="https://example.com/mcp",
            headers={"Authorization": "Bearer ${MCP_TOKEN}"},
        )
        assert _build_http_headers(cfg)["Authorization"] == "Bearer secret"

    def test_missing_environment_reference_fails(self, monkeypatch):
        monkeypatch.delenv("CMDFORGE_MISSING", raising=False)
        cfg = McpServerConfig(
            name="test", command="server", env={"SERVER_TOKEN": "${CMDFORGE_MISSING}"}
        )
        with pytest.raises(ValueError, match="is not set"):
            _build_server_env(cfg)


class TestMcpConfigPersistence:
    @pytest.fixture
    def temp_mcp_file(self, tmp_path, monkeypatch):
        config_file = tmp_path / "mcp.yaml"
        monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", config_file)
        yield config_file

    def test_save_and_load(self, temp_mcp_file):
        servers = [
            McpServerConfig(
                name="filesystem",
                command="npx",
                args=["-y", "@scope/server", "/tmp"],
                timeout=30,
                approved=True,
            )
        ]
        save_mcp_config(servers)
        loaded = load_mcp_config()
        assert len(loaded) == 1
        assert loaded[0].name == "filesystem"
        assert loaded[0].command == "npx"
        assert loaded[0].args == ["-y", "@scope/server", "/tmp"]
        assert loaded[0].approved is True

    def test_load_no_file(self, temp_mcp_file):
        assert load_mcp_config() == []

    def test_streamable_http_round_trip(self, temp_mcp_file):
        save_mcp_config([McpServerConfig(
            name="remote", transport="streamable-http",
            url="https://example.com/mcp",
            headers={"Authorization": "Bearer ${MCP_TOKEN}"},
            approved=True,
        )])
        loaded = load_mcp_config()[0]
        assert loaded.transport == "streamable-http"
        assert loaded.command is None
        assert loaded.url == "https://example.com/mcp"
        assert loaded.headers == {"Authorization": "Bearer ${MCP_TOKEN}"}

    def test_save_creates_0600_permissions(self, temp_mcp_file):
        save_mcp_config([McpServerConfig(name="test", command="echo")])
        perms = oct(temp_mcp_file.stat().st_mode & 0o777)
        assert perms == "0o600"

    def test_save_preserves_server_exposure_policy(self, temp_mcp_file):
        temp_mcp_file.write_text(
            "version: 1\nserver:\n  expose: [summarize]\n  deny: [dangerous]\n"
        )
        save_mcp_config([McpServerConfig(name="test", command="echo")])
        data = yaml.safe_load(temp_mcp_file.read_text())
        assert data["server"] == {
            "expose": ["summarize"],
            "deny": ["dangerous"],
        }


class TestResultNormalization:
    def test_auto_prefers_structured_content(self):
        class FakeResult:
            isError = False
            structuredContent = {"sum": 3}
            content = [type("Block", (), {"text": "plain text"})()]

        result = _normalize_result(FakeResult(), "auto")
        assert result == {"sum": 3}

    def test_auto_falls_back_to_text(self):
        class FakeResult:
            isError = False
            structuredContent = None
            content = [type("Block", (), {"type": "text", "text": "hello"})()]

        result = _normalize_result(FakeResult(), "auto")
        assert result == "hello"

    def test_auto_preserves_mixed_content(self):
        class FakeResult:
            isError = False
            structuredContent = None
            content = [
                {"type": "text", "text": "hello"},
                {"type": "image", "mimeType": "image/png", "data": "AA=="},
            ]

        result = _normalize_result(FakeResult(), "auto")
        assert result == FakeResult.content

    def test_content_mode_always_returns_blocks(self):
        class FakeResult:
            isError = False
            structuredContent = None
            content = [{"type": "text", "text": "hello"}]

        assert _normalize_result(FakeResult(), "content") == FakeResult.content

    def test_structured_fails_without_structured_content(self):
        class FakeResult:
            isError = False
            structuredContent = None
            content = [type("Block", (), {"type": "text", "text": "hello"})()]

        with pytest.raises(ValueError, match="structuredContent"):
            _normalize_result(FakeResult(), "structured")

    def test_text_mode(self):
        class FakeResult:
            isError = False
            structuredContent = None
            content = [
                type("Block", (), {"type": "text", "text": "line1"})(),
                type("Block", (), {"type": "text", "text": "line2"})(),
            ]

        result = _normalize_result(FakeResult(), "text")
        assert result == "line1\nline2"

    def test_iserror_causes_failure(self):
        class FakeResult:
            isError = True
            structuredContent = None
            content = [type("Block", (), {"type": "text", "text": "error message"})()]

        with pytest.raises(RuntimeError, match="error"):
            _normalize_result(FakeResult(), "auto")


class TestMcpClientManager:
    @pytest.fixture
    def manager(self, tmp_path, monkeypatch):
        config_file = tmp_path / "mcp.yaml"
        monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", config_file)
        return McpClientManager()

    def test_list_servers_empty(self, manager):
        assert manager.list_servers() == []

    def test_list_servers_with_config(self, manager, tmp_path, monkeypatch):
        config_file = tmp_path / "mcp.yaml"
        monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", config_file)
        save_mcp_config([McpServerConfig(name="test", command="echo")])

        mgr = McpClientManager()
        servers = mgr.list_servers()
        assert len(servers) == 1
        assert servers[0].name == "test"

    def test_call_tool_unknown_server(self, manager):
        with pytest.raises(KeyError, match="not configured"):
            manager.call_tool("unknown", "tool", {})

    def test_unapproved_server_is_not_executed(self, tmp_path, monkeypatch):
        config_file = tmp_path / "mcp.yaml"
        monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", config_file)
        save_mcp_config([McpServerConfig(name="test", command="echo")])
        with pytest.raises(PermissionError, match="not approved"):
            McpClientManager().discover("test")

    def test_missing_sdk_has_actionable_error(self, tmp_path, monkeypatch):
        config_file = tmp_path / "mcp.yaml"
        monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", config_file)
        save_mcp_config([
            McpServerConfig(name="test", command="echo", approved=True)
        ])
        monkeypatch.setitem(sys.modules, "mcp", None)
        with pytest.raises(ImportError, match=r"cmdforge\[mcp\]"):
            McpClientManager().discover("test")


class TestArgumentSubstitution:
    def test_single_variable_preserves_integer(self):
        result = _deep_substitute("42", {"limit": 42})
        assert result == "42"  # no braces, so no substitution

    def test_braced_variable_preserves_type(self):
        variables = {"settings": {"limit": 50}}
        result = _substitute_mcp_args(
            {"options": {"limit": "{settings.limit}"}}, variables
        )
        assert result["options"]["limit"] == 50

    def test_nested_dict_substitution(self):
        variables = {"path": "/tmp/data.csv"}
        result = _substitute_mcp_args(
            {"files": [{"source": "{path}"}]}, variables
        )
        assert result["files"][0]["source"] == "/tmp/data.csv"

    def test_string_with_variable_inside_text(self):
        variables = {"name": "report"}
        result = _deep_substitute("/tmp/{name}.csv", variables)
        assert result == "/tmp/report.csv"

    @pytest.mark.parametrize(
        "value",
        [
            {"a": 1},
            [1, 2],
            -5,
            "001",
            "true",
            False,
            None,
        ],
    )
    def test_exact_reference_preserves_original_type(self, value):
        assert _deep_substitute("{value}", {"value": value}) == value


class TestMcpStep:
    def test_round_trip(self):
        original = McpStep(
            server="fixture", tool="add", arguments={"a": 1, "b": 2},
            output_var="sum", result_mode="structured",
        )
        assert McpStep.from_dict(original.to_dict()) == original

    @pytest.mark.parametrize(
        "field, value",
        [("server", ""), ("tool", None), ("arguments", []), ("result_mode", "raw")],
    )
    def test_rejects_invalid_fields(self, field, value):
        data = {"server": "fixture", "tool": "echo", field: value}
        with pytest.raises(ValueError):
            McpStep.from_dict(data)

    def test_runner_preserves_structured_result(self, monkeypatch):
        calls = []

        class FakeManager:
            def call_tool(self, server, tool, arguments, result_mode="auto"):
                calls.append((server, tool, arguments, result_mode))
                return {"sum": 3}

        monkeypatch.setattr("cmdforge.runner.McpClientManager", FakeManager)
        tool = Tool(
            name="mcp-add",
            steps=[
                McpStep(
                    server="fixture",
                    tool="add",
                    arguments={"a": "{input}", "b": 2},
                    output_var="result",
                )
            ],
            output="{result.sum}",
        )
        output, exit_code = run_tool(tool, 1, {})
        assert (output, exit_code) == ("3", 0)
        assert calls == [("fixture", "add", {"a": 1, "b": 2}, "auto")]


class TestRealMcpSdk:
    @pytest.fixture
    def manager(self, tmp_path, monkeypatch):
        pytest.importorskip("mcp")
        config_file = tmp_path / "mcp.yaml"
        monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", config_file)
        fixture = Path(__file__).parent / "fixtures" / "mcp_test_server.py"
        save_mcp_config([
            McpServerConfig(
                name="fixture",
                command=sys.executable,
                args=[str(fixture)],
                timeout=10,
                approved=True,
            )
        ])
        return McpClientManager()

    def test_discovery_uses_mcp_handshake_and_tools_list(self, manager):
        tools = manager.discover("fixture")
        schemas = {tool["name"]: tool for tool in tools}
        assert {"echo", "add", "read_env", "current_directory", "wait"}.issubset(schemas)
        assert schemas["add"]["inputSchema"]["required"] == ["a", "b"]

    def test_calls_text_and_structured_tools(self, manager):
        assert manager.call_tool("fixture", "echo", {"message": "hello"}, "text") == "hello"
        assert manager.call_tool("fixture", "add", {"a": 1, "b": 2}) == {"sum": 3}

    def test_parent_secret_is_not_leaked(self, manager, monkeypatch):
        monkeypatch.setenv("CMDFORGE_TEST_SECRET", "do-not-leak")
        result = manager.call_tool(
            "fixture", "read_env", {"name": "CMDFORGE_TEST_SECRET"}, "text"
        )
        assert result == ""

    def test_configured_working_directory_is_used(self, manager, tmp_path):
        manager.list_servers()
        manager._configs["fixture"].cwd = str(tmp_path)
        assert manager.call_tool("fixture", "current_directory", {}, "text") == str(tmp_path)

    def test_timeout_is_enforced(self, manager):
        manager.list_servers()
        manager._configs["fixture"].timeout = 0.5
        started = time.monotonic()
        with pytest.raises(TimeoutError, match="timed out"):
            manager.call_tool("fixture", "wait", {"seconds": 10}, "text")
        assert time.monotonic() - started < 4


class TestMcpCli:
    def test_add_accepts_leading_dash_argument(self, tmp_path, monkeypatch, capsys):
        from cmdforge.cli import main

        config_file = tmp_path / "mcp.yaml"
        monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", config_file)
        monkeypatch.setattr(
            sys,
            "argv",
            [
                "cmdforge", "mcp", "add", "fixture", "--command", "npx",
                "--arg=-y", "--arg", "@scope/server",
            ],
        )
        assert main() == 0
        loaded = load_mcp_config()
        assert loaded[0].args == ["-y", "@scope/server"]
        assert loaded[0].approved is True
        assert "saved and approved" in capsys.readouterr().out

    def test_bare_mcp_lists_servers(self, tmp_path, monkeypatch, capsys):
        from cmdforge.cli import main

        monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", tmp_path / "mcp.yaml")
        monkeypatch.setattr(sys, "argv", ["cmdforge", "mcp"])
        assert main() == 0
        assert "No MCP servers configured" in capsys.readouterr().out


class TestMcpServeConfig:
    def test_empty_expose_exposes_nothing(self):
        from cmdforge.mcp_client import McpServeConfig
        config = McpServeConfig(expose=[], deny=[])
        assert not config.is_exposed("summarize")

    def test_wildcard_exposes_all_public(self):
        from cmdforge.mcp_client import McpServeConfig
        config = McpServeConfig(expose=["*"], deny=[])
        assert config.is_exposed("summarize")
        assert config.is_exposed("any-tool")

    def test_deny_overrides_expose(self):
        from cmdforge.mcp_client import McpServeConfig
        config = McpServeConfig(expose=["*"], deny=["dangerous"])
        assert config.is_exposed("summarize")
        assert not config.is_exposed("dangerous")

    def test_prefix_deny(self):
        from cmdforge.mcp_client import McpServeConfig
        config = McpServeConfig(expose=["*"], deny=["internal-*"])
        assert not config.is_exposed("internal-cleanup")
        assert not config.is_exposed("internal-")
        assert config.is_exposed("public-tool")

    def test_explicit_expose(self):
        from cmdforge.mcp_client import McpServeConfig
        config = McpServeConfig(expose=["summarize", "extract"])
        assert config.is_exposed("summarize")
        assert config.is_exposed("extract")
        assert not config.is_exposed("other")

    @pytest.mark.parametrize(
        "field, value",
        [("expose", "*"), ("deny", "dangerous"), ("expose", [""])],
    )
    def test_rejects_malformed_policy_lists(self, field, value):
        from cmdforge.mcp_client import McpServeConfig

        values = {"expose": [], "deny": [], field: value}
        with pytest.raises(ValueError, match="list of non-empty strings"):
            McpServeConfig(**values)

    def test_general_wildcard_matching(self):
        from cmdforge.mcp_client import McpServeConfig

        config = McpServeConfig(expose=["official/*"], deny=["*/dangerous"])
        assert config.is_exposed("official/summarize")
        assert not config.is_exposed("official/dangerous")

    def test_private_tool_still_honors_deny(self):
        from cmdforge.mcp_client import McpServeConfig
        from cmdforge.mcp_server import _is_exposable
        from cmdforge.tool import Tool

        tool = Tool(name="dangerous", visibility="private")
        config = McpServeConfig(expose=["official/dangerous"], deny=["official/*"])
        assert not _is_exposable(tool, config, qualified_name="official/dangerous")

    def test_private_qualified_tool_requires_qualified_expose(self):
        from cmdforge.mcp_client import McpServeConfig
        from cmdforge.mcp_server import _is_exposable
        from cmdforge.tool import Tool

        tool = Tool(name="extract", visibility="private")
        assert not _is_exposable(
            tool,
            McpServeConfig(expose=["extract"]),
            qualified_name="official/extract",
        )
        assert _is_exposable(
            tool,
            McpServeConfig(expose=["official/extract"]),
            qualified_name="official/extract",
        )


class TestToolNameMapping:
    def test_simple_name_passes_through(self):
        from cmdforge.mcp_server import _cmdforge_tool_name
        assert _cmdforge_tool_name("summarize") == "summarize"

    def test_qualified_name_uses_double_underscore(self):
        from cmdforge.mcp_server import _cmdforge_tool_name
        assert _cmdforge_tool_name("owner/tool") == "owner__tool"


class TestArgumentToJsonSchema:
    def test_string_arg(self):
        from cmdforge.mcp_server import _argument_to_json_schema
        from cmdforge.tool import ToolArgument
        arg = ToolArgument(flag="--name", variable="name", description="The name")
        schema = _argument_to_json_schema(arg)
        assert schema["type"] == "string"
        assert schema["description"] == "The name"

    def test_integer_arg_with_default(self):
        from cmdforge.tool import ToolArgument
        from cmdforge.mcp_server import _argument_to_json_schema
        arg = ToolArgument(
            flag="--limit", variable="limit", default="10", type="integer"
        )
        schema = _argument_to_json_schema(arg)
        assert schema["type"] == "integer"
        assert schema["default"] == 10

    def test_arg_with_enum(self):
        from cmdforge.tool import ToolArgument
        from cmdforge.mcp_server import _argument_to_json_schema
        arg = ToolArgument(
            flag="--mode", variable="mode", enum=["fast", "accurate"]
        )
        schema = _argument_to_json_schema(arg)
        assert schema["enum"] == ["fast", "accurate"]


class TestBuildToolSchema:
    def test_builds_input_schema(self, tmp_path):
        from cmdforge.tool import Tool, ToolArgument
        from cmdforge.mcp_server import _build_tool_schema

        with patch('cmdforge.tool.TOOLS_DIR', tmp_path / ".cmdforge"):
            tool = Tool(
                name="greet",
                description="Greet someone",
                arguments=[
                    ToolArgument(flag="--name", variable="name",
                                 description="Who to greet", default="World"),
                    ToolArgument(flag="--count", variable="count", default="1"),
                ],
                output="Hello {name} x{count}\n",
            )

            schema = _build_tool_schema(tool)

        assert schema["name"] == "greet"
        assert "Greet" in schema["description"]
        assert schema["inputSchema"]["type"] == "object"
        assert "name" in schema["inputSchema"]["properties"]
        assert "count" in schema["inputSchema"]["properties"]
        assert "input" in schema["inputSchema"]["properties"]


class TestRegisteredMcpTool:
    def test_fastmcp_uses_flat_typed_schema_and_invokes_tool(self):
        import asyncio
        from mcp.server.fastmcp import FastMCP
        from cmdforge.mcp_server import _build_tool_schema, _register_tool
        from cmdforge.tool import Tool, ToolArgument

        server = FastMCP("test")
        tool = Tool(
            name="greet",
            arguments=[
                ToolArgument(
                    flag="--name",
                    variable="name",
                    type="string",
                    required=True,
                    description="Who to greet",
                ),
                ToolArgument(
                    flag="--count",
                    variable="count",
                    type="integer",
                    default=1,
                    enum=[1, 2, 3],
                ),
            ],
            output="Hello {name} x{count}: {input}",
        )
        _register_tool(server, "greet", "greet", _build_tool_schema(tool), tool)

        async def exercise():
            listed = (await server.list_tools())[0]
            result = await server.call_tool(
                "greet", {"name": "Alice", "count": 2, "input": "welcome"}
            )
            return listed, result

        listed, result = asyncio.run(exercise())
        assert listed.inputSchema["required"] == ["name"]
        assert listed.inputSchema["properties"]["count"]["type"] == "integer"
        assert listed.inputSchema["properties"]["count"]["enum"] == [1, 2, 3]
        assert "kwargs" not in listed.inputSchema["properties"]
        assert result[0].text == "Hello Alice x2: welcome"

    def test_handler_starts_at_inherited_depth(self):
        import asyncio
        from mcp.server.fastmcp import FastMCP
        from cmdforge.mcp_server import _build_tool_schema, _register_tool
        from cmdforge.tool import Tool

        server = FastMCP("test")
        tool = Tool(name="depth", output="ok")
        with patch("cmdforge.runner.run_tool", return_value=("ok", 0)) as run:
            _register_tool(server, "depth", "depth", _build_tool_schema(tool), tool)
            asyncio.run(server.call_tool("depth", {}))
        assert run.call_args.kwargs["_depth"] == 0

    def test_handler_rejects_exhausted_mcp_depth(self, monkeypatch):
        import asyncio
        from mcp.server.fastmcp import FastMCP
        from cmdforge.mcp_server import (
            MAX_TOOL_DEPTH,
            _build_tool_schema,
            _register_tool,
        )
        from cmdforge.tool import Tool

        monkeypatch.setenv("CMDFORGE_MCP_DEPTH", str(MAX_TOOL_DEPTH))
        server = FastMCP("test")
        tool = Tool(name="depth", output="ok")
        _register_tool(server, "depth", "depth", _build_tool_schema(tool), tool)
        with pytest.raises(Exception, match="Maximum MCP nesting depth"):
            asyncio.run(server.call_tool("depth", {}))

    def test_serve_fails_closed_on_mapped_name_collision(self):
        from cmdforge.mcp_client import McpServeConfig
        from cmdforge.mcp_server import serve
        from cmdforge.tool import Tool

        tools = {
            "owner/tool": Tool(name="tool"),
            "owner__tool": Tool(name="owner__tool"),
        }
        with (
            patch(
                "cmdforge.mcp_server.load_mcp_serve_config",
                return_value=McpServeConfig(expose=["*"]),
            ),
            patch("cmdforge.tool.list_tools", return_value=list(tools)),
            patch("cmdforge.tool.load_tool", side_effect=tools.get),
            patch("mcp.server.fastmcp.FastMCP.run"),
        ):
            with pytest.raises(RuntimeError, match="name collision"):
                serve()


class TestStreamableHttpServerSafety:
    def test_defaults_are_loopback_with_origin_allowlist(self):
        from cmdforge.mcp_server import _validate_http_server_options
        origins, token = _validate_http_server_options(
            "127.0.0.1", 8000, [], None, None
        )
        assert token is None
        assert origins == [
            "http://localhost:8000", "http://127.0.0.1:8000"
        ]

    def test_nonlocal_requires_https_and_auth(self):
        from cmdforge.mcp_server import _validate_http_server_options
        with pytest.raises(ValueError, match="external-url"):
            _validate_http_server_options("0.0.0.0", 8000, [], "secret", None)
        with pytest.raises(ValueError, match="auth-token"):
            _validate_http_server_options(
                "0.0.0.0", 8000, [], None, "https://mcp.example.com"
            )

    def test_nonlocal_infers_https_origin_and_expands_token(self, monkeypatch):
        from cmdforge.mcp_server import _validate_http_server_options
        monkeypatch.setenv("MCP_AUTH", "secret-value")
        origins, token = _validate_http_server_options(
            "0.0.0.0", 8000, [], "${MCP_AUTH}",
            "https://mcp.example.com",
        )
        assert origins == ["https://mcp.example.com"]
        assert token == "secret-value"

    def test_nonlocal_rejects_insecure_origin(self):
        from cmdforge.mcp_server import _validate_http_server_options
        with pytest.raises(ValueError, match="origins must use HTTPS"):
            _validate_http_server_options(
                "0.0.0.0", 8000, ["http://example.com"], "secret",
                "https://mcp.example.com",
            )


class TestCmdForgeMcpServerEndToEnd:
    def test_stdio_discovery_and_invocation(self, tmp_path, monkeypatch):
        pytest.importorskip("mcp")
        home = tmp_path / "home"
        cmdforge_dir = home / ".cmdforge"
        cmdforge_dir.mkdir(parents=True)

        (cmdforge_dir / "mcp.yaml").write_text(
            yaml.safe_dump({
                "version": 1,
                "server": {
                    "expose": ["greet", "echo-input", "hidden"],
                    "deny": ["hidden"],
                },
            })
        )
        tools = {
            "greet": {
                "name": "greet",
                "arguments": [{
                    "flag": "--name",
                    "variable": "name",
                    "type": "string",
                    "required": True,
                }],
                "steps": [],
                "output": "Hello {name}",
            },
            "echo-input": {
                "name": "echo-input",
                "steps": [],
                "output": "{input}",
            },
            "hidden": {
                "name": "hidden",
                "visibility": "private",
                "steps": [],
                "output": "secret",
            },
        }
        for name, data in tools.items():
            tool_dir = cmdforge_dir / name
            tool_dir.mkdir()
            (tool_dir / "config.yaml").write_text(yaml.safe_dump(data))

        client_config = tmp_path / "client-mcp.yaml"
        monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", client_config)
        save_mcp_config([
            McpServerConfig(
                name="cmdforge",
                command=sys.executable,
                args=["-m", "cmdforge.cli", "mcp", "serve"],
                env={"HOME": str(home)},
                timeout=10,
                approved=True,
            )
        ])
        manager = McpClientManager()

        discovered = {tool["name"]: tool for tool in manager.discover("cmdforge")}
        assert set(discovered) == {"greet", "echo-input"}
        assert discovered["greet"]["inputSchema"]["required"] == ["name"]
        assert manager.call_tool(
            "cmdforge", "greet", {"name": "Alice"}, "text"
        ) == "Hello Alice"
        assert manager.call_tool(
            "cmdforge", "echo-input", {"input": "through stdin"}, "text"
        ) == "through stdin"