Blame for tests/test_tool_spaces.py

from __future__ import annotations
import os
import sys
import json
from types import SimpleNamespace
from pathlib import Path
from argparse import Namespace

import httpx
import pytest
from huggingface_hub.errors import RepositoryNotFoundError

import peachy.config as config_mod
import peachy.tool_spaces as tool_spaces_mod
from peachy.main import main
from peachy.mcp_client import RemoteToolSpec
from peachy.tool_spaces import (
    resolve_tool_space_sync,
    handle_tool_spaces_command,
    read_installed_tool_spaces,
)


SEARCH_SPACE_SLUG = "example/search-tool"
COLLIDING_SEARCH_SPACE_SLUG = "example/search_tool"
PRIVATE_SPACE_SLUG = "example/private-space"
SEARCH_ALIAS = "example_search_tool"
SEARCH_REMOTE_NAME = "search_tool_search_web"
SEARCH_TOOL_ID = f"{SEARCH_ALIAS}__search_web"
SEARCH_CLIENT_TOOL_ID = f"{SEARCH_ALIAS}__{SEARCH_REMOTE_NAME}"


@pytest.fixture(autouse=True)
def _isolate_peachy_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("PEACHY_STATE_DIR", str((tmp_path / "peachy-state").resolve()))


def _test_state_root() -> Path:
    return Path(os.environ["PEACHY_STATE_DIR"])


def _mock_public_space_info(slug: str) -> SimpleNamespace:
    return SimpleNamespace(
        id=slug,
        private=False,
        disabled=False,
        sdk="gradio",
        host=None,
        subdomain=slug.replace("/", "-"),
        tags=["reachy-mini-tool", "mcp"],
    )


def _mock_private_space_info(slug: str) -> SimpleNamespace:
    info = _mock_public_space_info(slug)
    info.private = True
    return info


async def _mock_list_tool_specs(self: object) -> list[RemoteToolSpec]:
    return [
        RemoteToolSpec(
            server_alias=SEARCH_ALIAS,
            remote_name=SEARCH_REMOTE_NAME,
            namespaced_name=SEARCH_CLIENT_TOOL_ID,
            description="Search the web",
            parameters_schema={
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"],
            },
        )
    ]


def _run_cli(monkeypatch: pytest.MonkeyPatch, argv: list[str]) -> int:
    monkeypatch.setattr(sys, "argv", argv)
    with pytest.raises(SystemExit) as exc:
        main()
    return int(exc.value.code)


def test_tool_spaces_add_list_remove_round_trip(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The CLI should install, list, and remove a public Space tool source cleanly."""
    monkeypatch.chdir(tmp_path)
    monkeypatch.setattr(
        "peachy.tool_spaces.HfApi.space_info",
        lambda self, slug, **kwargs: _mock_public_space_info(slug),
    )
    monkeypatch.setattr(
        "peachy.tool_spaces.RemoteMcpToolClient.list_tool_specs",
        _mock_list_tool_specs,
    )

    assert (
        _run_cli(
            monkeypatch,
            [
                "reachy-mini-conversation-app",
                "tool-spaces",
                "add",
                SEARCH_SPACE_SLUG,
                "--install-only",
            ],
        )
        == 0
    )

    manifest_path = _test_state_root() / "installed_tool_spaces.json"
    assert manifest_path.is_file()
    written = json.loads(manifest_path.read_text(encoding="utf-8"))
    assert written["version"] == 2
    added_entry = next(space for space in written["spaces"] if space["slug"] == SEARCH_SPACE_SLUG)
    assert added_entry == {
        "slug": SEARCH_SPACE_SLUG,
        "alias": SEARCH_ALIAS,
        "mcp_url": "https://example-search-tool.hf.space/gradio_api/mcp/",
        "private": False,
        "tools": [
            {
                "local_name": SEARCH_TOOL_ID,
                "client_tool_name": SEARCH_CLIENT_TOOL_ID,
                "remote_name": SEARCH_REMOTE_NAME,
                "description": "Search the web",
                "parameters_schema": {
                    "type": "object",
                    "properties": {"query": {"type": "string"}},
                    "required": ["query"],
                },
            }
        ],
    }

    assert _run_cli(monkeypatch, ["reachy-mini-conversation-app", "tool-spaces", "list"]) == 0

    assert _run_cli(monkeypatch, ["reachy-mini-conversation-app", "tool-spaces", "remove", SEARCH_SPACE_SLUG]) == 0
    assert SEARCH_SPACE_SLUG not in [space.slug for space in read_installed_tool_spaces(_test_state_root()).spaces]


def test_tool_spaces_add_installs_private_space_with_token(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A private Space resolves and installs when an HF token is available."""
    monkeypatch.chdir(tmp_path)
    monkeypatch.setattr(
        "peachy.tool_spaces.HfApi.space_info",
        lambda self, slug, **kwargs: _mock_private_space_info(slug),
    )
    monkeypatch.setattr("peachy.tool_spaces.get_token", lambda: "hf_test_token")
    monkeypatch.setattr(
        "peachy.tool_spaces.RemoteMcpToolClient.list_tool_specs",
        _mock_list_tool_specs,
    )

    assert _run_cli(monkeypatch, ["app", "tool-spaces", "add", PRIVATE_SPACE_SLUG, "--install-only"]) == 0
    assert PRIVATE_SPACE_SLUG in [space.slug for space in read_installed_tool_spaces(_test_state_root()).spaces]


def test_resolve_tool_space_attaches_auth_header_for_private_space(monkeypatch: pytest.MonkeyPatch) -> None:
    """A private Space's MCP client must carry the HF bearer token from the hf-login fallback."""
    monkeypatch.setattr(
        "peachy.tool_spaces.HfApi.space_info",
        lambda self, slug, **kwargs: _mock_private_space_info(slug),
    )
    monkeypatch.setattr(config_mod.config, "HF_TOKEN", None)
    monkeypatch.setattr("peachy.tool_spaces.get_token", lambda: "hf_test_token")
    monkeypatch.setattr(
        "peachy.tool_spaces.RemoteMcpToolClient.list_tool_specs",
        _mock_list_tool_specs,
    )

    resolved = resolve_tool_space_sync(PRIVATE_SPACE_SLUG)
    assert resolved.client.server.headers["Authorization"] == "Bearer hf_test_token"


def test_resolve_tool_space_prefers_app_config_hf_token(monkeypatch: pytest.MonkeyPatch) -> None:
    """The token must come from the app config (loaded from .env), not only from get_token()."""
    monkeypatch.setattr(
        "peachy.tool_spaces.HfApi.space_info",
        lambda self, slug, **kwargs: _mock_private_space_info(slug),
    )
    monkeypatch.setattr(config_mod.config, "HF_TOKEN", "hf_env_token")
    monkeypatch.setattr("peachy.tool_spaces.get_token", lambda: None)
    monkeypatch.setattr(
        "peachy.tool_spaces.RemoteMcpToolClient.list_tool_specs",
        _mock_list_tool_specs,
    )

    resolved = resolve_tool_space_sync(PRIVATE_SPACE_SLUG)
    assert resolved.client.server.headers["Authorization"] == "Bearer hf_env_token"


def test_resolve_tool_space_omits_auth_header_for_public_space(monkeypatch: pytest.MonkeyPatch) -> None:
    """A public Space must never receive the HF token, even when one is set."""
    monkeypatch.setattr(
        "peachy.tool_spaces.HfApi.space_info",
        lambda self, slug, **kwargs: _mock_public_space_info(slug),
    )
    monkeypatch.setattr(config_mod.config, "HF_TOKEN", None)
    monkeypatch.setattr("peachy.tool_spaces.get_token", lambda: "hf_test_token")
    monkeypatch.setattr(
        "peachy.tool_spaces.RemoteMcpToolClient.list_tool_specs",
        _mock_list_tool_specs,
    )

    resolved = resolve_tool_space_sync(SEARCH_SPACE_SLUG)
    assert "Authorization" not in resolved.client.server.headers


def test_tool_spaces_add_private_space_without_token_hints_at_auth(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    """Without a token, adding a private Space fails cleanly and points at HF auth."""
    monkeypatch.chdir(tmp_path)

    def _raise_not_found(self: object, slug: str, **kwargs: object) -> SimpleNamespace:
        raise RepositoryNotFoundError(
            "404 Client Error", response=httpx.Response(404, request=httpx.Request("GET", "https://hf.co"))
        )

    monkeypatch.setattr("peachy.tool_spaces.HfApi.space_info", _raise_not_found)
    monkeypatch.setattr(config_mod.config, "HF_TOKEN", None)
    monkeypatch.setattr("peachy.tool_spaces.get_token", lambda: None)

    assert _run_cli(monkeypatch, ["app", "tool-spaces", "add", PRIVATE_SPACE_SLUG]) == 1
    assert "hf auth login" in capsys.readouterr().err
    assert not (_test_state_root() / "installed_tool_spaces.json").exists()


def test_tool_spaces_manifest_uses_instance_path_when_provided(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Managed instance paths should store the manifest beside other instance-local state."""
    monkeypatch.setattr(
        "peachy.tool_spaces.HfApi.space_info",
        lambda self, slug, **kwargs: _mock_public_space_info(slug),
    )
    monkeypatch.setattr(
        "peachy.tool_spaces.RemoteMcpToolClient.list_tool_specs",
        _mock_list_tool_specs,
    )

    args = Namespace(
        tool_spaces_command="add",
        space_slug=SEARCH_SPACE_SLUG,
        install_only=True,
        profile=None,
    )
    assert handle_tool_spaces_command(args, instance_path=tmp_path) == 0
    assert (tmp_path / "installed_tool_spaces.json").is_file()
    assert not (tmp_path / "external_content" / "installed_tool_spaces.json").exists()


def test_read_installed_tool_spaces_raises_on_alias_collision_in_manifest(tmp_path: Path) -> None:
    """A manifest with two slugs that normalize to the same alias must be rejected on read."""
    mcp_url = "https://example.hf.space/gradio_api/mcp/"
    payload = {
        "version": 2,
        "spaces": [
            {"slug": "owner/my-tool", "alias": "owner_my_tool", "mcp_url": mcp_url, "private": False, "tools": []},
            {"slug": "owner/my_tool", "alias": "owner_my_tool", "mcp_url": mcp_url, "private": False, "tools": []},
        ],
    }
    (tmp_path / "installed_tool_spaces.json").write_text(json.dumps(payload), encoding="utf-8")

    with pytest.raises(RuntimeError, match="alias collision"):
        read_installed_tool_spaces(tmp_path)


def test_tool_spaces_add_rejects_alias_collision(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A second Space whose slug normalizes to the same alias must be rejected."""
    monkeypatch.chdir(tmp_path)
    monkeypatch.setattr(
        "peachy.tool_spaces.HfApi.space_info",
        lambda self, slug, **kwargs: _mock_public_space_info(slug),
    )
    monkeypatch.setattr(
        "peachy.tool_spaces.RemoteMcpToolClient.list_tool_specs",
        _mock_list_tool_specs,
    )

    assert (
        _run_cli(
            monkeypatch,
            ["app", "tool-spaces", "add", SEARCH_SPACE_SLUG, "--install-only"],
        )
        == 0
    )

    # The owner separator style differs, but both slugs normalize to the same alias.
    assert (
        _run_cli(
            monkeypatch,
            ["app", "tool-spaces", "add", COLLIDING_SEARCH_SPACE_SLUG, "--install-only"],
        )
        == 1
    )


def _setup_profile(tmp_path: Path, profile: str, existing_tools: list[str] | None = None) -> Path:
    """Create a profile directory with an optional tools.txt."""
    profile_dir = tmp_path / profile
    profile_dir.mkdir(parents=True)
    tools_txt = profile_dir / "tools.txt"
    tools_txt.write_text("\n".join(existing_tools or []) + "\n" if existing_tools else "", encoding="utf-8")
    return tools_txt


def _mock_add(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    monkeypatch.chdir(tmp_path)
    monkeypatch.setattr(
        "peachy.tool_spaces.HfApi.space_info",
        lambda self, slug, **kwargs: _mock_public_space_info(slug),
    )
    monkeypatch.setattr(
        "peachy.tool_spaces.RemoteMcpToolClient.list_tool_specs",
        _mock_list_tool_specs,
    )


def test_tool_spaces_add_enables_in_active_profile_by_default(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Add without flags should enable tools in the active profile."""
    _mock_add(monkeypatch, tmp_path)
    tools_txt = _setup_profile(tmp_path, "default")
    monkeypatch.setattr(config_mod.config, "PROFILES_DIRECTORY", tmp_path)
    monkeypatch.setattr(config_mod.config, "REACHY_MINI_CUSTOM_PROFILE", None)

    assert _run_cli(monkeypatch, ["app", "tool-spaces", "add", SEARCH_SPACE_SLUG]) == 0

    assert SEARCH_TOOL_ID in tools_txt.read_text(encoding="utf-8")


def test_tool_spaces_remove_disables_tools_in_profile(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Removing a Space strips its tool IDs from the profile they were enabled in."""
    _mock_add(monkeypatch, tmp_path)
    tools_txt = _setup_profile(_test_state_root() / "user_personalities", "default")
    monkeypatch.setattr(config_mod.config, "PROFILES_DIRECTORY", tmp_path)
    monkeypatch.setattr(config_mod.config, "REACHY_MINI_CUSTOM_PROFILE", "user_personalities/default")

    assert _run_cli(monkeypatch, ["app", "tool-spaces", "add", SEARCH_SPACE_SLUG]) == 0
    assert SEARCH_TOOL_ID in tools_txt.read_text(encoding="utf-8")

    assert _run_cli(monkeypatch, ["app", "tool-spaces", "remove", SEARCH_SPACE_SLUG]) == 0
    assert SEARCH_TOOL_ID not in tools_txt.read_text(encoding="utf-8")
    assert SEARCH_SPACE_SLUG not in [space.slug for space in read_installed_tool_spaces(_test_state_root()).spaces]


def test_tool_spaces_add_install_only_skips_tools_txt(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """--install-only should not modify any profile's tools.txt."""
    _mock_add(monkeypatch, tmp_path)
    tools_txt = _setup_profile(tmp_path, "default")
    monkeypatch.setattr(config_mod.config, "PROFILES_DIRECTORY", tmp_path)

    assert _run_cli(monkeypatch, ["app", "tool-spaces", "add", SEARCH_SPACE_SLUG, "--install-only"]) == 0

    assert tools_txt.read_text(encoding="utf-8") == ""


def test_tool_spaces_add_profile_flag_enables_in_specified_profile(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """--profile should enable tools in the named profile, not the active one."""
    _mock_add(monkeypatch, tmp_path)
    default_tools_txt = _setup_profile(tmp_path, "default")
    canary_tools_txt = _setup_profile(tmp_path, "canary")
    monkeypatch.setattr(config_mod.config, "PROFILES_DIRECTORY", tmp_path)
    monkeypatch.setattr(config_mod.config, "REACHY_MINI_CUSTOM_PROFILE", "default")

    assert _run_cli(monkeypatch, ["app", "tool-spaces", "add", SEARCH_SPACE_SLUG, "--profile", "canary"]) == 0

    assert SEARCH_TOOL_ID in canary_tools_txt.read_text(encoding="utf-8")
    assert default_tools_txt.read_text(encoding="utf-8") == ""


def test_read_installed_tool_spaces_seeds_bundled_pollen_spaces(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """With no manifest, the three bundled Pollen Spaces are seeded with their tools cached offline."""
    monkeypatch.chdir(tmp_path)

    spaces = read_installed_tool_spaces(_test_state_root()).spaces
    assert [space.slug for space in spaces] == [
        "pollen-robotics/reachy-mini-search-tool",
        "pollen-robotics/reachy-mini-time-tool",
        "pollen-robotics/reachy-mini-weather-tool",
    ]
    search_tool = spaces[0].tools[0]
    assert search_tool.local_name == "pollen_robotics_reachy_mini_search_tool__search_web"
    assert search_tool.remote_name == "reachy_mini_search_tool_search_web"
    assert spaces[0].private is False


def test_builtin_profile_cannot_be_mutated(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    builtin = tmp_path / "package-profiles"
    tools_txt = _setup_profile(builtin, "default")
    monkeypatch.setattr(config_mod.config, "PROFILES_DIRECTORY", builtin)
    monkeypatch.setattr(tool_spaces_mod, "DEFAULT_PROFILES_DIRECTORY", builtin)

    with pytest.raises(RuntimeError, match="read-only"):
        tool_spaces_mod._append_tools_to_profile("default", [SEARCH_TOOL_ID])

    assert tools_txt.read_text(encoding="utf-8") == ""


def test_append_tools_rejects_profile_traversal(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    external_profiles = tmp_path / "external-profiles"
    escaped_tools = _setup_profile(tmp_path, "static")
    monkeypatch.setattr(config_mod.config, "PROFILES_DIRECTORY", external_profiles)

    with pytest.raises(RuntimeError, match="Invalid profile selector"):
        tool_spaces_mod._append_tools_to_profile("../static", [SEARCH_TOOL_ID])

    assert escaped_tools.read_text(encoding="utf-8") == ""


def test_append_tools_rejects_tools_file_symlink_to_builtin_profile(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    builtin = tmp_path / "package-profiles"
    builtin_tools = _setup_profile(builtin, "default", ["builtin_tool"])
    external_profiles = tmp_path / "external-profiles"
    external_profile = external_profiles / "custom"
    external_profile.mkdir(parents=True)
    (external_profile / "tools.txt").symlink_to(builtin_tools)
    monkeypatch.setattr(config_mod.config, "PROFILES_DIRECTORY", external_profiles)
    monkeypatch.setattr(tool_spaces_mod, "DEFAULT_PROFILES_DIRECTORY", builtin)
    original_contents = builtin_tools.read_bytes()

    with pytest.raises(RuntimeError, match="not inside a writable profile root"):
        tool_spaces_mod._append_tools_to_profile("custom", [SEARCH_TOOL_ID])

    assert builtin_tools.read_bytes() == original_contents


def test_remove_space_only_rewrites_user_profiles(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    builtin = tmp_path / "package-profiles"
    user = tmp_path / "state" / "user_personalities"
    builtin_tools = _setup_profile(builtin, "default", [SEARCH_TOOL_ID])
    user_tools = _setup_profile(user, "custom", [SEARCH_TOOL_ID])
    monkeypatch.setattr(config_mod.config, "PROFILES_DIRECTORY", builtin)
    monkeypatch.setattr(config_mod.config, "user_personalities_root", lambda: user)

    removed = tool_spaces_mod._disable_space_tools_in_profiles(SEARCH_ALIAS)

    assert removed == [("custom", [SEARCH_TOOL_ID])]
    assert SEARCH_TOOL_ID in builtin_tools.read_text(encoding="utf-8")
    assert SEARCH_TOOL_ID not in user_tools.read_text(encoding="utf-8")


def test_remove_space_skips_symlinked_user_profile_that_escapes_state_root(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    builtin = tmp_path / "package-profiles"
    builtin_tools = _setup_profile(builtin, "default", [SEARCH_TOOL_ID])
    user_root = tmp_path / "state" / "user_personalities"
    user_root.mkdir(parents=True)
    (user_root / "escaped").symlink_to(builtin / "default", target_is_directory=True)
    monkeypatch.setattr(config_mod.config, "user_personalities_root", lambda: user_root)

    removed = tool_spaces_mod._disable_space_tools_in_profiles(SEARCH_ALIAS)

    assert removed == []
    assert SEARCH_TOOL_ID in builtin_tools.read_text(encoding="utf-8")


def test_remove_space_skips_user_root_symlink_to_builtin_profiles(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    builtin = tmp_path / "package-profiles"
    builtin_tools = _setup_profile(builtin, "default", [SEARCH_TOOL_ID])
    user_root_link = tmp_path / "state" / "user_personalities"
    user_root_link.parent.mkdir(parents=True)
    user_root_link.symlink_to(builtin, target_is_directory=True)
    monkeypatch.setattr(config_mod.config, "user_personalities_root", lambda: user_root_link)
    monkeypatch.setattr(tool_spaces_mod, "DEFAULT_PROFILES_DIRECTORY", builtin)

    removed = tool_spaces_mod._disable_space_tools_in_profiles(SEARCH_ALIAS)

    assert removed == []
    assert SEARCH_TOOL_ID in builtin_tools.read_text(encoding="utf-8")