Spaces:
Running
Running
| 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}" | |
| 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") | |