Merge branch 'main' into rewbs/tool-use-charge-to-subscription
This commit is contained in:
Robin Fernandes
290
tests/tools/test_browser_camofox.py
Normal file
290
tests/tools/test_browser_camofox.py
Normal file
@@ -0,0 +1,290 @@
|
||||
"""Tests for the Camofox browser backend."""
|
||||
|
||||
import json
|
||||
import os
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from tools.browser_camofox import (
|
||||
camofox_back,
|
||||
camofox_click,
|
||||
camofox_close,
|
||||
camofox_console,
|
||||
camofox_get_images,
|
||||
camofox_navigate,
|
||||
camofox_press,
|
||||
camofox_scroll,
|
||||
camofox_snapshot,
|
||||
camofox_type,
|
||||
camofox_vision,
|
||||
check_camofox_available,
|
||||
cleanup_all_camofox_sessions,
|
||||
is_camofox_mode,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Configuration detection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCamofoxMode:
|
||||
def test_disabled_by_default(self, monkeypatch):
|
||||
monkeypatch.delenv("CAMOFOX_URL", raising=False)
|
||||
assert is_camofox_mode() is False
|
||||
|
||||
def test_enabled_when_url_set(self, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
assert is_camofox_mode() is True
|
||||
|
||||
def test_health_check_unreachable(self, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:19999")
|
||||
assert check_camofox_available() is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _mock_response(status=200, json_data=None):
|
||||
resp = MagicMock()
|
||||
resp.status_code = status
|
||||
resp.json.return_value = json_data or {}
|
||||
resp.content = b"\x89PNG\r\n\x1a\nfake"
|
||||
resp.raise_for_status = MagicMock()
|
||||
return resp
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Navigate
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCamofoxNavigate:
|
||||
@patch("tools.browser_camofox.requests.post")
|
||||
def test_creates_tab_on_first_navigate(self, mock_post, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
mock_post.return_value = _mock_response(json_data={"tabId": "tab1", "url": "https://example.com"})
|
||||
|
||||
result = json.loads(camofox_navigate("https://example.com", task_id="t1"))
|
||||
assert result["success"] is True
|
||||
assert result["url"] == "https://example.com"
|
||||
|
||||
@patch("tools.browser_camofox.requests.post")
|
||||
def test_navigates_existing_tab(self, mock_post, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
# First call creates tab
|
||||
mock_post.return_value = _mock_response(json_data={"tabId": "tab2", "url": "https://a.com"})
|
||||
camofox_navigate("https://a.com", task_id="t2")
|
||||
|
||||
# Second call navigates
|
||||
mock_post.return_value = _mock_response(json_data={"ok": True, "url": "https://b.com"})
|
||||
result = json.loads(camofox_navigate("https://b.com", task_id="t2"))
|
||||
assert result["success"] is True
|
||||
assert result["url"] == "https://b.com"
|
||||
|
||||
def test_connection_error_returns_helpful_message(self, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:19999")
|
||||
result = json.loads(camofox_navigate("https://example.com", task_id="t_err"))
|
||||
assert result["success"] is False
|
||||
assert "Cannot connect" in result["error"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Snapshot
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCamofoxSnapshot:
|
||||
def test_no_session_returns_error(self, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
result = json.loads(camofox_snapshot(task_id="no_such_task"))
|
||||
assert result["success"] is False
|
||||
assert "browser_navigate" in result["error"]
|
||||
|
||||
@patch("tools.browser_camofox.requests.post")
|
||||
@patch("tools.browser_camofox.requests.get")
|
||||
def test_returns_snapshot(self, mock_get, mock_post, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
# Create session
|
||||
mock_post.return_value = _mock_response(json_data={"tabId": "tab3", "url": "https://x.com"})
|
||||
camofox_navigate("https://x.com", task_id="t3")
|
||||
|
||||
# Return snapshot
|
||||
mock_get.return_value = _mock_response(json_data={
|
||||
"snapshot": "- heading \"Test\" [e1]\n- button \"Submit\" [e2]",
|
||||
"refsCount": 2,
|
||||
})
|
||||
result = json.loads(camofox_snapshot(task_id="t3"))
|
||||
assert result["success"] is True
|
||||
assert "[e1]" in result["snapshot"]
|
||||
assert result["element_count"] == 2
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Click / Type / Scroll / Back / Press
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCamofoxInteractions:
|
||||
@patch("tools.browser_camofox.requests.post")
|
||||
def test_click(self, mock_post, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
mock_post.return_value = _mock_response(json_data={"tabId": "tab4", "url": "https://x.com"})
|
||||
camofox_navigate("https://x.com", task_id="t4")
|
||||
|
||||
mock_post.return_value = _mock_response(json_data={"ok": True, "url": "https://x.com"})
|
||||
result = json.loads(camofox_click("@e5", task_id="t4"))
|
||||
assert result["success"] is True
|
||||
assert result["clicked"] == "e5"
|
||||
|
||||
@patch("tools.browser_camofox.requests.post")
|
||||
def test_type(self, mock_post, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
mock_post.return_value = _mock_response(json_data={"tabId": "tab5", "url": "https://x.com"})
|
||||
camofox_navigate("https://x.com", task_id="t5")
|
||||
|
||||
mock_post.return_value = _mock_response(json_data={"ok": True})
|
||||
result = json.loads(camofox_type("@e3", "hello world", task_id="t5"))
|
||||
assert result["success"] is True
|
||||
assert result["typed"] == "hello world"
|
||||
|
||||
@patch("tools.browser_camofox.requests.post")
|
||||
def test_scroll(self, mock_post, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
mock_post.return_value = _mock_response(json_data={"tabId": "tab6", "url": "https://x.com"})
|
||||
camofox_navigate("https://x.com", task_id="t6")
|
||||
|
||||
mock_post.return_value = _mock_response(json_data={"ok": True})
|
||||
result = json.loads(camofox_scroll("down", task_id="t6"))
|
||||
assert result["success"] is True
|
||||
assert result["scrolled"] == "down"
|
||||
|
||||
@patch("tools.browser_camofox.requests.post")
|
||||
def test_back(self, mock_post, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
mock_post.return_value = _mock_response(json_data={"tabId": "tab7", "url": "https://x.com"})
|
||||
camofox_navigate("https://x.com", task_id="t7")
|
||||
|
||||
mock_post.return_value = _mock_response(json_data={"ok": True, "url": "https://prev.com"})
|
||||
result = json.loads(camofox_back(task_id="t7"))
|
||||
assert result["success"] is True
|
||||
|
||||
@patch("tools.browser_camofox.requests.post")
|
||||
def test_press(self, mock_post, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
mock_post.return_value = _mock_response(json_data={"tabId": "tab8", "url": "https://x.com"})
|
||||
camofox_navigate("https://x.com", task_id="t8")
|
||||
|
||||
mock_post.return_value = _mock_response(json_data={"ok": True})
|
||||
result = json.loads(camofox_press("Enter", task_id="t8"))
|
||||
assert result["success"] is True
|
||||
assert result["pressed"] == "Enter"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Close
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCamofoxClose:
|
||||
@patch("tools.browser_camofox.requests.delete")
|
||||
@patch("tools.browser_camofox.requests.post")
|
||||
def test_close_session(self, mock_post, mock_delete, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
mock_post.return_value = _mock_response(json_data={"tabId": "tab9", "url": "https://x.com"})
|
||||
camofox_navigate("https://x.com", task_id="t9")
|
||||
|
||||
mock_delete.return_value = _mock_response(json_data={"ok": True})
|
||||
result = json.loads(camofox_close(task_id="t9"))
|
||||
assert result["success"] is True
|
||||
assert result["closed"] is True
|
||||
|
||||
def test_close_nonexistent_session(self, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
result = json.loads(camofox_close(task_id="nonexistent"))
|
||||
assert result["success"] is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Console (limited support)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCamofoxConsole:
|
||||
def test_console_returns_empty_with_note(self, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
result = json.loads(camofox_console(task_id="t_console"))
|
||||
assert result["success"] is True
|
||||
assert result["total_messages"] == 0
|
||||
assert "not available" in result["note"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Images
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCamofoxGetImages:
|
||||
@patch("tools.browser_camofox.requests.post")
|
||||
@patch("tools.browser_camofox.requests.get")
|
||||
def test_get_images(self, mock_get, mock_post, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
mock_post.return_value = _mock_response(json_data={"tabId": "tab10", "url": "https://x.com"})
|
||||
camofox_navigate("https://x.com", task_id="t10")
|
||||
|
||||
mock_get.return_value = _mock_response(json_data={
|
||||
"images": [{"src": "https://x.com/img.png", "alt": "Logo"}],
|
||||
})
|
||||
result = json.loads(camofox_get_images(task_id="t10"))
|
||||
assert result["success"] is True
|
||||
assert result["count"] == 1
|
||||
assert result["images"][0]["src"] == "https://x.com/img.png"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Routing integration — verify browser_tool routes to camofox
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBrowserToolRouting:
|
||||
"""Verify that browser_tool.py delegates to camofox when CAMOFOX_URL is set."""
|
||||
|
||||
@patch("tools.browser_camofox.requests.post")
|
||||
def test_browser_navigate_routes_to_camofox(self, mock_post, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
mock_post.return_value = _mock_response(json_data={"tabId": "tab_rt", "url": "https://example.com"})
|
||||
|
||||
from tools.browser_tool import browser_navigate
|
||||
# Bypass SSRF check for test URL
|
||||
with patch("tools.browser_tool._is_safe_url", return_value=True):
|
||||
result = json.loads(browser_navigate("https://example.com", task_id="t_route"))
|
||||
assert result["success"] is True
|
||||
|
||||
def test_check_requirements_passes_with_camofox(self, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
from tools.browser_tool import check_browser_requirements
|
||||
assert check_browser_requirements() is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Cleanup helper
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCamofoxCleanup:
|
||||
@patch("tools.browser_camofox.requests.post")
|
||||
@patch("tools.browser_camofox.requests.delete")
|
||||
def test_cleanup_all(self, mock_delete, mock_post, monkeypatch):
|
||||
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
|
||||
mock_post.return_value = _mock_response(json_data={"tabId": "tab_c", "url": "https://x.com"})
|
||||
camofox_navigate("https://x.com", task_id="t_cleanup")
|
||||
|
||||
mock_delete.return_value = _mock_response(json_data={"ok": True})
|
||||
cleanup_all_camofox_sessions()
|
||||
|
||||
# Session should be gone
|
||||
result = json.loads(camofox_snapshot(task_id="t_cleanup"))
|
||||
assert result["success"] is False
|
||||
@@ -1,13 +1,17 @@
|
||||
"""Tests for credential file passthrough registry (tools/credential_files.py)."""
|
||||
"""Tests for credential file passthrough and skills directory mounting."""
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from tools.credential_files import (
|
||||
clear_credential_files,
|
||||
get_credential_file_mounts,
|
||||
get_skills_directory_mount,
|
||||
iter_skills_files,
|
||||
register_credential_file,
|
||||
register_credential_files,
|
||||
reset_config_cache,
|
||||
@@ -15,8 +19,8 @@ from tools.credential_files import (
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clean_registry():
|
||||
"""Reset registry between tests."""
|
||||
def _clean_state():
|
||||
"""Reset module state between tests."""
|
||||
clear_credential_files()
|
||||
reset_config_cache()
|
||||
yield
|
||||
@@ -24,135 +28,172 @@ def _clean_registry():
|
||||
reset_config_cache()
|
||||
|
||||
|
||||
class TestRegisterCredentialFile:
|
||||
def test_registers_existing_file(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
(tmp_path / "token.json").write_text('{"token": "abc"}')
|
||||
class TestRegisterCredentialFiles:
|
||||
def test_dict_with_path_key(self, tmp_path):
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
(hermes_home / "token.json").write_text("{}")
|
||||
|
||||
result = register_credential_file("token.json")
|
||||
with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}):
|
||||
missing = register_credential_files([{"path": "token.json"}])
|
||||
|
||||
assert result is True
|
||||
assert missing == []
|
||||
mounts = get_credential_file_mounts()
|
||||
assert len(mounts) == 1
|
||||
assert mounts[0]["host_path"] == str(tmp_path / "token.json")
|
||||
assert mounts[0]["host_path"] == str(hermes_home / "token.json")
|
||||
assert mounts[0]["container_path"] == "/root/.hermes/token.json"
|
||||
|
||||
def test_skips_missing_file(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
def test_dict_with_name_key_fallback(self, tmp_path):
|
||||
"""Skills use 'name' instead of 'path' — both should work."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
(hermes_home / "google_token.json").write_text("{}")
|
||||
|
||||
result = register_credential_file("nonexistent.json")
|
||||
|
||||
assert result is False
|
||||
assert get_credential_file_mounts() == []
|
||||
|
||||
def test_custom_container_base(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
(tmp_path / "cred.json").write_text("{}")
|
||||
|
||||
register_credential_file("cred.json", container_base="/home/user/.hermes")
|
||||
with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}):
|
||||
missing = register_credential_files([
|
||||
{"name": "google_token.json", "description": "OAuth token"},
|
||||
])
|
||||
|
||||
assert missing == []
|
||||
mounts = get_credential_file_mounts()
|
||||
assert mounts[0]["container_path"] == "/home/user/.hermes/cred.json"
|
||||
assert len(mounts) == 1
|
||||
assert "google_token.json" in mounts[0]["container_path"]
|
||||
|
||||
def test_deduplicates_by_container_path(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
(tmp_path / "token.json").write_text("{}")
|
||||
def test_string_entry(self, tmp_path):
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
(hermes_home / "secret.key").write_text("key")
|
||||
|
||||
register_credential_file("token.json")
|
||||
register_credential_file("token.json")
|
||||
with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}):
|
||||
missing = register_credential_files(["secret.key"])
|
||||
|
||||
assert missing == []
|
||||
mounts = get_credential_file_mounts()
|
||||
assert len(mounts) == 1
|
||||
|
||||
def test_missing_file_reported(self, tmp_path):
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
|
||||
class TestRegisterCredentialFiles:
|
||||
def test_string_entries(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
(tmp_path / "a.json").write_text("{}")
|
||||
(tmp_path / "b.json").write_text("{}")
|
||||
with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}):
|
||||
missing = register_credential_files([
|
||||
{"name": "does_not_exist.json"},
|
||||
])
|
||||
|
||||
missing = register_credential_files(["a.json", "b.json"])
|
||||
|
||||
assert missing == []
|
||||
assert len(get_credential_file_mounts()) == 2
|
||||
|
||||
def test_dict_entries(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
(tmp_path / "token.json").write_text("{}")
|
||||
|
||||
missing = register_credential_files([
|
||||
{"path": "token.json", "description": "OAuth token"},
|
||||
])
|
||||
|
||||
assert missing == []
|
||||
assert len(get_credential_file_mounts()) == 1
|
||||
|
||||
def test_returns_missing_files(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
(tmp_path / "exists.json").write_text("{}")
|
||||
|
||||
missing = register_credential_files([
|
||||
"exists.json",
|
||||
"missing.json",
|
||||
{"path": "also_missing.json"},
|
||||
])
|
||||
|
||||
assert missing == ["missing.json", "also_missing.json"]
|
||||
assert len(get_credential_file_mounts()) == 1
|
||||
|
||||
def test_empty_list(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
assert register_credential_files([]) == []
|
||||
|
||||
|
||||
class TestConfigCredentialFiles:
|
||||
def test_loads_from_config(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
(tmp_path / "oauth.json").write_text("{}")
|
||||
(tmp_path / "config.yaml").write_text(
|
||||
"terminal:\n credential_files:\n - oauth.json\n"
|
||||
)
|
||||
|
||||
mounts = get_credential_file_mounts()
|
||||
|
||||
assert len(mounts) == 1
|
||||
assert mounts[0]["host_path"] == str(tmp_path / "oauth.json")
|
||||
|
||||
def test_config_skips_missing_files(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
(tmp_path / "config.yaml").write_text(
|
||||
"terminal:\n credential_files:\n - nonexistent.json\n"
|
||||
)
|
||||
|
||||
mounts = get_credential_file_mounts()
|
||||
assert mounts == []
|
||||
|
||||
def test_combines_skill_and_config(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
(tmp_path / "skill_token.json").write_text("{}")
|
||||
(tmp_path / "config_token.json").write_text("{}")
|
||||
(tmp_path / "config.yaml").write_text(
|
||||
"terminal:\n credential_files:\n - config_token.json\n"
|
||||
)
|
||||
|
||||
register_credential_file("skill_token.json")
|
||||
mounts = get_credential_file_mounts()
|
||||
|
||||
assert len(mounts) == 2
|
||||
paths = {m["container_path"] for m in mounts}
|
||||
assert "/root/.hermes/skill_token.json" in paths
|
||||
assert "/root/.hermes/config_token.json" in paths
|
||||
|
||||
|
||||
class TestGetMountsRechecksExistence:
|
||||
def test_removed_file_excluded_from_mounts(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
token = tmp_path / "token.json"
|
||||
token.write_text("{}")
|
||||
|
||||
register_credential_file("token.json")
|
||||
assert len(get_credential_file_mounts()) == 1
|
||||
|
||||
# Delete the file after registration
|
||||
token.unlink()
|
||||
assert "does_not_exist.json" in missing
|
||||
assert get_credential_file_mounts() == []
|
||||
|
||||
def test_path_takes_precedence_over_name(self, tmp_path):
|
||||
"""When both path and name are present, path wins."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
(hermes_home / "real.json").write_text("{}")
|
||||
|
||||
with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}):
|
||||
missing = register_credential_files([
|
||||
{"path": "real.json", "name": "wrong.json"},
|
||||
])
|
||||
|
||||
assert missing == []
|
||||
mounts = get_credential_file_mounts()
|
||||
assert "real.json" in mounts[0]["container_path"]
|
||||
|
||||
|
||||
class TestSkillsDirectoryMount:
|
||||
def test_returns_mount_when_skills_dir_exists(self, tmp_path):
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
skills_dir = hermes_home / "skills"
|
||||
skills_dir.mkdir(parents=True)
|
||||
(skills_dir / "test-skill").mkdir()
|
||||
(skills_dir / "test-skill" / "SKILL.md").write_text("# test")
|
||||
|
||||
with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}):
|
||||
mount = get_skills_directory_mount()
|
||||
|
||||
assert mount is not None
|
||||
assert mount["host_path"] == str(skills_dir)
|
||||
assert mount["container_path"] == "/root/.hermes/skills"
|
||||
|
||||
def test_returns_none_when_no_skills_dir(self, tmp_path):
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
|
||||
with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}):
|
||||
mount = get_skills_directory_mount()
|
||||
|
||||
assert mount is None
|
||||
|
||||
def test_custom_container_base(self, tmp_path):
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
(hermes_home / "skills").mkdir(parents=True)
|
||||
|
||||
with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}):
|
||||
mount = get_skills_directory_mount(container_base="/home/user/.hermes")
|
||||
|
||||
assert mount["container_path"] == "/home/user/.hermes/skills"
|
||||
|
||||
def test_symlinks_are_sanitized(self, tmp_path):
|
||||
"""Symlinks in skills dir should be excluded from the mount."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
skills_dir = hermes_home / "skills"
|
||||
skills_dir.mkdir(parents=True)
|
||||
(skills_dir / "legit.md").write_text("# real skill")
|
||||
# Create a symlink pointing outside the skills tree
|
||||
secret = tmp_path / "secret.txt"
|
||||
secret.write_text("TOP SECRET")
|
||||
(skills_dir / "evil_link").symlink_to(secret)
|
||||
|
||||
with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}):
|
||||
mount = get_skills_directory_mount()
|
||||
|
||||
assert mount is not None
|
||||
# The mount path should be a sanitized copy, not the original
|
||||
safe_path = Path(mount["host_path"])
|
||||
assert safe_path != skills_dir
|
||||
# Legitimate file should be present
|
||||
assert (safe_path / "legit.md").exists()
|
||||
assert (safe_path / "legit.md").read_text() == "# real skill"
|
||||
# Symlink should NOT be present
|
||||
assert not (safe_path / "evil_link").exists()
|
||||
|
||||
def test_no_symlinks_returns_original_dir(self, tmp_path):
|
||||
"""When no symlinks exist, the original dir is returned (no copy)."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
skills_dir = hermes_home / "skills"
|
||||
skills_dir.mkdir(parents=True)
|
||||
(skills_dir / "skill.md").write_text("ok")
|
||||
|
||||
with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}):
|
||||
mount = get_skills_directory_mount()
|
||||
|
||||
assert mount["host_path"] == str(skills_dir)
|
||||
|
||||
|
||||
class TestIterSkillsFiles:
|
||||
def test_returns_files_skipping_symlinks(self, tmp_path):
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
skills_dir = hermes_home / "skills"
|
||||
(skills_dir / "cat" / "myskill").mkdir(parents=True)
|
||||
(skills_dir / "cat" / "myskill" / "SKILL.md").write_text("# skill")
|
||||
(skills_dir / "cat" / "myskill" / "scripts").mkdir()
|
||||
(skills_dir / "cat" / "myskill" / "scripts" / "run.sh").write_text("#!/bin/bash")
|
||||
# Add a symlink that should be filtered
|
||||
secret = tmp_path / "secret"
|
||||
secret.write_text("nope")
|
||||
(skills_dir / "cat" / "myskill" / "evil").symlink_to(secret)
|
||||
|
||||
with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}):
|
||||
files = iter_skills_files()
|
||||
|
||||
paths = {f["container_path"] for f in files}
|
||||
assert "/root/.hermes/skills/cat/myskill/SKILL.md" in paths
|
||||
assert "/root/.hermes/skills/cat/myskill/scripts/run.sh" in paths
|
||||
# Symlink should be excluded
|
||||
assert not any("evil" in f["container_path"] for f in files)
|
||||
|
||||
def test_empty_when_no_skills_dir(self, tmp_path):
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
|
||||
with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}):
|
||||
assert iter_skills_files() == []
|
||||
|
||||
@@ -61,6 +61,10 @@ def make_env(daytona_sdk, monkeypatch):
|
||||
"""Factory that creates a DaytonaEnvironment with a mocked SDK."""
|
||||
# Prevent is_interrupted from interfering
|
||||
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
|
||||
# Prevent skills/credential sync from consuming mock exec calls
|
||||
monkeypatch.setattr("tools.credential_files.get_credential_file_mounts", lambda: [])
|
||||
monkeypatch.setattr("tools.credential_files.get_skills_directory_mount", lambda **kw: None)
|
||||
monkeypatch.setattr("tools.credential_files.iter_skills_files", lambda **kw: [])
|
||||
|
||||
def _factory(
|
||||
sandbox=None,
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import types
|
||||
from importlib.util import module_from_spec, spec_from_file_location
|
||||
@@ -28,6 +29,7 @@ def _reset_modules(prefixes: tuple[str, ...]):
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _restore_tool_modules():
|
||||
original_hermes_home = os.environ.get("HERMES_HOME")
|
||||
original_modules = {
|
||||
name: module
|
||||
for name, module in sys.modules.items()
|
||||
@@ -41,6 +43,10 @@ def _restore_tool_modules():
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
if original_hermes_home is None:
|
||||
os.environ.pop("HERMES_HOME", None)
|
||||
else:
|
||||
os.environ["HERMES_HOME"] = original_hermes_home
|
||||
_reset_modules(("tools", "hermes_cli", "modal"))
|
||||
sys.modules.update(original_modules)
|
||||
|
||||
@@ -57,6 +63,7 @@ def _install_modal_test_modules(
|
||||
hermes_cli.__path__ = [] # type: ignore[attr-defined]
|
||||
sys.modules["hermes_cli"] = hermes_cli
|
||||
hermes_home = tmp_path / "hermes-home"
|
||||
os.environ["HERMES_HOME"] = str(hermes_home)
|
||||
sys.modules["hermes_cli.config"] = types.SimpleNamespace(
|
||||
get_hermes_home=lambda: hermes_home,
|
||||
)
|
||||
|
||||
@@ -5,6 +5,7 @@ from pathlib import Path
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from tools.skills_hub import (
|
||||
GitHubAuth,
|
||||
@@ -648,6 +649,29 @@ class TestWellKnownSkillSource:
|
||||
assert bundle.files["SKILL.md"] == "# Code Review\n"
|
||||
assert bundle.files["references/checklist.md"] == "- [ ] security\n"
|
||||
|
||||
@patch("tools.skills_hub._write_index_cache")
|
||||
@patch("tools.skills_hub._read_index_cache", return_value=None)
|
||||
@patch("tools.skills_hub.httpx.get")
|
||||
def test_fetch_rejects_unsafe_file_paths_from_well_known_endpoint(self, mock_get, _mock_read_cache, _mock_write_cache):
|
||||
def fake_get(url, *args, **kwargs):
|
||||
if url.endswith("/index.json"):
|
||||
return MagicMock(status_code=200, json=lambda: {
|
||||
"skills": [{
|
||||
"name": "code-review",
|
||||
"description": "Review code",
|
||||
"files": ["SKILL.md", "../../../escape.txt"],
|
||||
}]
|
||||
})
|
||||
if url.endswith("/code-review/SKILL.md"):
|
||||
return MagicMock(status_code=200, text="# Code Review\n")
|
||||
raise AssertionError(url)
|
||||
|
||||
mock_get.side_effect = fake_get
|
||||
|
||||
bundle = self._source().fetch("well-known:https://example.com/.well-known/skills/code-review")
|
||||
|
||||
assert bundle is None
|
||||
|
||||
|
||||
class TestCheckForSkillUpdates:
|
||||
def test_bundle_content_hash_matches_installed_content_hash(self, tmp_path):
|
||||
@@ -1143,6 +1167,61 @@ class TestQuarantineBundleBinaryAssets:
|
||||
assert (q_path / "SKILL.md").read_text(encoding="utf-8").startswith("---")
|
||||
assert (q_path / "assets" / "neutts-cli" / "samples" / "jo.wav").read_bytes() == b"RIFF\x00\x01fakewav"
|
||||
|
||||
def test_quarantine_bundle_rejects_traversal_file_paths(self, tmp_path):
|
||||
import tools.skills_hub as hub
|
||||
|
||||
hub_dir = tmp_path / "skills" / ".hub"
|
||||
with patch.object(hub, "SKILLS_DIR", tmp_path / "skills"), \
|
||||
patch.object(hub, "HUB_DIR", hub_dir), \
|
||||
patch.object(hub, "LOCK_FILE", hub_dir / "lock.json"), \
|
||||
patch.object(hub, "QUARANTINE_DIR", hub_dir / "quarantine"), \
|
||||
patch.object(hub, "AUDIT_LOG", hub_dir / "audit.log"), \
|
||||
patch.object(hub, "TAPS_FILE", hub_dir / "taps.json"), \
|
||||
patch.object(hub, "INDEX_CACHE_DIR", hub_dir / "index-cache"):
|
||||
bundle = SkillBundle(
|
||||
name="demo",
|
||||
files={
|
||||
"SKILL.md": "---\nname: demo\n---\n",
|
||||
"../../../escape.txt": "owned",
|
||||
},
|
||||
source="well-known",
|
||||
identifier="well-known:https://example.com/.well-known/skills/demo",
|
||||
trust_level="community",
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="Unsafe bundle file path"):
|
||||
quarantine_bundle(bundle)
|
||||
|
||||
assert not (tmp_path / "skills" / "escape.txt").exists()
|
||||
|
||||
def test_quarantine_bundle_rejects_absolute_file_paths(self, tmp_path):
|
||||
import tools.skills_hub as hub
|
||||
|
||||
hub_dir = tmp_path / "skills" / ".hub"
|
||||
absolute_target = tmp_path / "outside.txt"
|
||||
with patch.object(hub, "SKILLS_DIR", tmp_path / "skills"), \
|
||||
patch.object(hub, "HUB_DIR", hub_dir), \
|
||||
patch.object(hub, "LOCK_FILE", hub_dir / "lock.json"), \
|
||||
patch.object(hub, "QUARANTINE_DIR", hub_dir / "quarantine"), \
|
||||
patch.object(hub, "AUDIT_LOG", hub_dir / "audit.log"), \
|
||||
patch.object(hub, "TAPS_FILE", hub_dir / "taps.json"), \
|
||||
patch.object(hub, "INDEX_CACHE_DIR", hub_dir / "index-cache"):
|
||||
bundle = SkillBundle(
|
||||
name="demo",
|
||||
files={
|
||||
"SKILL.md": "---\nname: demo\n---\n",
|
||||
str(absolute_target): "owned",
|
||||
},
|
||||
source="well-known",
|
||||
identifier="well-known:https://example.com/.well-known/skills/demo",
|
||||
trust_level="community",
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="Unsafe bundle file path"):
|
||||
quarantine_bundle(bundle)
|
||||
|
||||
assert not absolute_target.exists()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GitHubSource._download_directory — tree API + fallback (#2940)
|
||||
|
||||
27
tests/tools/test_terminal_timeout_output.py
Normal file
27
tests/tools/test_terminal_timeout_output.py
Normal file
@@ -0,0 +1,27 @@
|
||||
"""Verify that terminal command timeouts preserve partial output."""
|
||||
from tools.environments.local import LocalEnvironment
|
||||
|
||||
|
||||
class TestTimeoutPreservesPartialOutput:
|
||||
"""When a command times out, any output captured before the deadline
|
||||
should be included in the result — not discarded."""
|
||||
|
||||
def test_timeout_includes_partial_output(self):
|
||||
"""A command that prints then sleeps past the deadline should
|
||||
return both the printed text and the timeout notice."""
|
||||
env = LocalEnvironment()
|
||||
result = env.execute("echo 'hello from test' && sleep 30", timeout=2)
|
||||
|
||||
assert result["returncode"] == 124
|
||||
assert "hello from test" in result["output"]
|
||||
assert "timed out" in result["output"].lower()
|
||||
|
||||
def test_timeout_with_no_output(self):
|
||||
"""A command that produces nothing before timeout should still
|
||||
return a clean timeout message."""
|
||||
env = LocalEnvironment()
|
||||
result = env.execute("sleep 30", timeout=1)
|
||||
|
||||
assert result["returncode"] == 124
|
||||
assert "timed out" in result["output"].lower()
|
||||
assert not result["output"].startswith("\n")
|
||||
@@ -259,6 +259,12 @@ def test_check_website_access_uses_dynamic_hermes_home(monkeypatch, tmp_path):
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
|
||||
# Invalidate the module-level cache so the new HERMES_HOME is picked up.
|
||||
# A prior test may have cached a default policy (enabled=False) under the
|
||||
# old HERMES_HOME set by the autouse _isolate_hermes_home fixture.
|
||||
from tools.website_policy import invalidate_cache
|
||||
invalidate_cache()
|
||||
|
||||
blocked = check_website_access("https://dynamic.example/path")
|
||||
|
||||
assert blocked is not None
|
||||
|
||||
Reference in New Issue
Block a user