mirror of
https://github.com/HKUDS/OpenSpace.git
synced 2026-09-10 22:41:02 +00:00
chore: remove outdated test files
This commit is contained in:
spidercatfly
parent
2fb8024ff6
commit
f05514845d
4 changed files with 0 additions and 424 deletions
|
|
@ -1,177 +0,0 @@
|
|||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from openspace import mcp_server
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_local_search_skills_does_not_initialize_openspace(monkeypatch, tmp_path):
|
||||
skill_root = tmp_path / "skills"
|
||||
skill_dir = skill_root / "demo-local-skill"
|
||||
skill_dir.mkdir(parents=True)
|
||||
(skill_dir / "SKILL.md").write_text(
|
||||
"""---
|
||||
name: Demo Local Skill
|
||||
description: Local regression test skill
|
||||
---
|
||||
|
||||
Find me when querying demo local skill.
|
||||
""",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
async def forbidden_get_openspace():
|
||||
pytest.fail("_get_openspace should not run for source='local'")
|
||||
|
||||
monkeypatch.setenv("OPENSPACE_HOST_SKILL_DIRS", str(skill_root))
|
||||
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
|
||||
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
|
||||
monkeypatch.delenv("OPENAI_BASE_URL", raising=False)
|
||||
monkeypatch.setattr(mcp_server, "_get_openspace", forbidden_get_openspace)
|
||||
monkeypatch.setattr(
|
||||
"openspace.cloud.embedding.generate_embedding",
|
||||
lambda text, api_key=None: None,
|
||||
)
|
||||
|
||||
response = await mcp_server.search_skills(
|
||||
query="demo local skill",
|
||||
source="local",
|
||||
limit=5,
|
||||
auto_import=False,
|
||||
)
|
||||
|
||||
payload = json.loads(response)
|
||||
assert payload["count"] >= 1
|
||||
assert any(
|
||||
item["name"] == "Demo Local Skill"
|
||||
for item in payload["results"]
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_local_search_skills_refreshes_registry_between_calls(monkeypatch, tmp_path):
|
||||
skill_root = tmp_path / "skills"
|
||||
first_skill_dir = skill_root / "first-local-skill"
|
||||
first_skill_dir.mkdir(parents=True)
|
||||
(first_skill_dir / "SKILL.md").write_text(
|
||||
"""---
|
||||
name: First Local Skill
|
||||
description: First local skill for cache regression coverage
|
||||
---
|
||||
|
||||
First local skill content.
|
||||
""",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
async def forbidden_get_openspace():
|
||||
pytest.fail("_get_openspace should not run for source='local'")
|
||||
|
||||
monkeypatch.setenv("OPENSPACE_HOST_SKILL_DIRS", str(skill_root))
|
||||
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
|
||||
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
|
||||
monkeypatch.delenv("OPENAI_BASE_URL", raising=False)
|
||||
monkeypatch.setattr(mcp_server, "_get_openspace", forbidden_get_openspace)
|
||||
monkeypatch.setattr(
|
||||
"openspace.cloud.embedding.generate_embedding",
|
||||
lambda text, api_key=None: None,
|
||||
)
|
||||
|
||||
first_response = await mcp_server.search_skills(
|
||||
query="first local skill",
|
||||
source="local",
|
||||
limit=5,
|
||||
auto_import=False,
|
||||
)
|
||||
|
||||
first_payload = json.loads(first_response)
|
||||
assert any(
|
||||
item["name"] == "First Local Skill"
|
||||
for item in first_payload["results"]
|
||||
)
|
||||
|
||||
second_skill_dir = skill_root / "second-local-skill"
|
||||
second_skill_dir.mkdir(parents=True)
|
||||
(second_skill_dir / "SKILL.md").write_text(
|
||||
"""---
|
||||
name: Second Local Skill
|
||||
description: Second local skill created after the first search
|
||||
---
|
||||
|
||||
Second local skill content.
|
||||
""",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
second_response = await mcp_server.search_skills(
|
||||
query="second local skill",
|
||||
source="local",
|
||||
limit=5,
|
||||
auto_import=False,
|
||||
)
|
||||
|
||||
second_payload = json.loads(second_response)
|
||||
assert any(
|
||||
item["name"] == "Second Local Skill"
|
||||
for item in second_payload["results"]
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_local_search_returns_empty_when_no_registry(monkeypatch):
|
||||
"""source='local' with no discoverable skills should return empty, not crash."""
|
||||
|
||||
async def forbidden_get_openspace():
|
||||
pytest.fail("_get_openspace should not run for source='local'")
|
||||
|
||||
monkeypatch.setattr(mcp_server, "_get_openspace", forbidden_get_openspace)
|
||||
monkeypatch.setattr(mcp_server, "_get_local_skill_registry", lambda: None)
|
||||
monkeypatch.setattr(
|
||||
"openspace.cloud.embedding.generate_embedding",
|
||||
lambda text, api_key=None: None,
|
||||
)
|
||||
|
||||
response = await mcp_server.search_skills(
|
||||
query="anything",
|
||||
source="local",
|
||||
limit=5,
|
||||
auto_import=False,
|
||||
)
|
||||
|
||||
payload = json.loads(response)
|
||||
assert payload["count"] == 0
|
||||
assert payload["results"] == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_source_all_still_calls_get_openspace(monkeypatch, tmp_path):
|
||||
"""source='all' must go through _get_openspace(), not the lightweight path."""
|
||||
called = {"openspace": False}
|
||||
|
||||
class FakeRegistry:
|
||||
def list_skills(self):
|
||||
return []
|
||||
|
||||
class FakeOpenSpace:
|
||||
_skill_registry = FakeRegistry()
|
||||
|
||||
async def tracking_get_openspace():
|
||||
called["openspace"] = True
|
||||
return FakeOpenSpace()
|
||||
|
||||
monkeypatch.delenv("OPENSPACE_HOST_SKILL_DIRS", raising=False)
|
||||
monkeypatch.setattr(mcp_server, "_get_openspace", tracking_get_openspace)
|
||||
monkeypatch.setattr(
|
||||
"openspace.cloud.embedding.generate_embedding",
|
||||
lambda text, api_key=None: None,
|
||||
)
|
||||
|
||||
await mcp_server.search_skills(
|
||||
query="anything",
|
||||
source="all",
|
||||
limit=5,
|
||||
auto_import=False,
|
||||
)
|
||||
|
||||
assert called["openspace"], "source='all' should call _get_openspace()"
|
||||
|
|
@ -1,64 +0,0 @@
|
|||
"""Test that max_iterations is passed to grounding agent in no-skill path."""
|
||||
|
||||
import asyncio
|
||||
import types
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from openspace.tool_layer import OpenSpace, OpenSpaceConfig
|
||||
|
||||
|
||||
def _make_openspace(configured_max: int = 50) -> OpenSpace:
|
||||
config = OpenSpaceConfig()
|
||||
config.grounding_max_iterations = configured_max
|
||||
os_inst = OpenSpace(config)
|
||||
os_inst._initialized = True
|
||||
os_inst._running = False
|
||||
os_inst._task_done = asyncio.Event()
|
||||
os_inst._task_done.set()
|
||||
os_inst._grounding_client = types.SimpleNamespace(_registry={})
|
||||
os_inst._recording_manager = None
|
||||
os_inst._skill_registry = None
|
||||
os_inst._execution_analyzer = None
|
||||
os_inst._skill_evolver = None
|
||||
return os_inst
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_skill_path_passes_configured_max_iterations():
|
||||
"""When no skills match, the configured grounding_max_iterations
|
||||
must be forwarded to the agent, not silently dropped."""
|
||||
os_inst = _make_openspace(configured_max=50)
|
||||
recorded = {}
|
||||
|
||||
async def fake_process(context):
|
||||
recorded.update(context)
|
||||
return {"status": "success", "iterations": 1, "tool_executions": []}
|
||||
|
||||
os_inst._grounding_agent = types.SimpleNamespace(process=fake_process)
|
||||
os_inst._maybe_analyze_execution = AsyncMock()
|
||||
os_inst._maybe_evolve_quality = AsyncMock()
|
||||
|
||||
await os_inst.execute("do something")
|
||||
|
||||
assert recorded.get("max_iterations") == 50
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_skill_path_uses_caller_override_when_larger():
|
||||
"""Caller-provided max_iterations should win when larger than config."""
|
||||
os_inst = _make_openspace(configured_max=20)
|
||||
recorded = {}
|
||||
|
||||
async def fake_process(context):
|
||||
recorded.update(context)
|
||||
return {"status": "success", "iterations": 1, "tool_executions": []}
|
||||
|
||||
os_inst._grounding_agent = types.SimpleNamespace(process=fake_process)
|
||||
os_inst._maybe_analyze_execution = AsyncMock()
|
||||
os_inst._maybe_evolve_quality = AsyncMock()
|
||||
|
||||
await os_inst.execute("do something", max_iterations=100)
|
||||
|
||||
assert recorded.get("max_iterations") == 100
|
||||
|
|
@ -1,94 +0,0 @@
|
|||
"""Tests for workflow ID uniqueness and discovery."""
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from openspace import dashboard_server
|
||||
|
||||
|
||||
def _make_workflow(path: Path):
|
||||
"""Create a minimal workflow directory with metadata.json."""
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
(path / "metadata.json").write_text("{}", encoding="utf-8")
|
||||
|
||||
|
||||
class TestWorkflowId:
|
||||
def test_same_leaf_name_different_roots_get_unique_ids(self, tmp_path):
|
||||
root_a = tmp_path / "root_a"
|
||||
root_b = tmp_path / "root_b"
|
||||
wf_a = root_a / "task1"
|
||||
wf_b = root_b / "task1"
|
||||
_make_workflow(wf_a)
|
||||
_make_workflow(wf_b)
|
||||
|
||||
with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root_a, root_b]):
|
||||
id_a = dashboard_server._workflow_id(wf_a)
|
||||
id_b = dashboard_server._workflow_id(wf_b)
|
||||
|
||||
assert id_a != id_b
|
||||
assert id_a.startswith("task1_")
|
||||
assert id_b.startswith("task1_")
|
||||
|
||||
def test_id_contains_dir_name_and_hash(self, tmp_path):
|
||||
root = tmp_path / "root"
|
||||
wf = root / "my-task"
|
||||
_make_workflow(wf)
|
||||
|
||||
with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root]):
|
||||
wf_id = dashboard_server._workflow_id(wf)
|
||||
|
||||
assert wf_id.startswith("my-task_")
|
||||
assert len(wf_id) == len("my-task_") + 8 # 8-char hex hash
|
||||
|
||||
def test_separator_collision_produces_different_ids(self, tmp_path):
|
||||
"""Regression: a dir named 'a__b' must not collide with path a/b."""
|
||||
root = tmp_path / "root"
|
||||
flat = root / "a__b"
|
||||
nested = root / "a" / "b"
|
||||
_make_workflow(flat)
|
||||
_make_workflow(nested)
|
||||
|
||||
with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root]):
|
||||
id_flat = dashboard_server._workflow_id(flat)
|
||||
id_nested = dashboard_server._workflow_id(nested)
|
||||
|
||||
assert id_flat != id_nested
|
||||
|
||||
def test_workflow_outside_roots_still_works(self, tmp_path):
|
||||
wf = tmp_path / "orphan_workflow"
|
||||
_make_workflow(wf)
|
||||
|
||||
with patch.object(dashboard_server, "WORKFLOW_ROOTS", []):
|
||||
wf_id = dashboard_server._workflow_id(wf)
|
||||
|
||||
assert wf_id.startswith("orphan_workflow_")
|
||||
|
||||
|
||||
class TestDiscoverWorkflowDirs:
|
||||
def test_same_name_workflows_both_discovered(self, tmp_path):
|
||||
root_a = tmp_path / "root_a"
|
||||
root_b = tmp_path / "root_b"
|
||||
_make_workflow(root_a / "task1")
|
||||
_make_workflow(root_b / "task1")
|
||||
|
||||
with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root_a, root_b]):
|
||||
dirs = dashboard_server._discover_workflow_dirs()
|
||||
|
||||
assert len(dirs) == 2
|
||||
|
||||
def test_get_workflow_dir_resolves_correct_path(self, tmp_path):
|
||||
root_a = tmp_path / "root_a"
|
||||
root_b = tmp_path / "root_b"
|
||||
_make_workflow(root_a / "task1")
|
||||
_make_workflow(root_b / "task1")
|
||||
|
||||
with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root_a, root_b]):
|
||||
id_a = dashboard_server._workflow_id(root_a / "task1")
|
||||
id_b = dashboard_server._workflow_id(root_b / "task1")
|
||||
found_a = dashboard_server._get_workflow_dir(id_a)
|
||||
found_b = dashboard_server._get_workflow_dir(id_b)
|
||||
|
||||
assert found_a.resolve() == (root_a / "task1").resolve()
|
||||
assert found_b.resolve() == (root_b / "task1").resolve()
|
||||
|
|
@ -1,89 +0,0 @@
|
|||
"""Regression tests for zip extraction and import_skill path traversal."""
|
||||
|
||||
import io
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def _make_zip(entries: dict[str, bytes]) -> bytes:
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, "w") as zf:
|
||||
for name, data in entries.items():
|
||||
zf.writestr(name, data)
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
def _get_extract_zip():
|
||||
import importlib
|
||||
mod = importlib.import_module("openspace.cloud.client")
|
||||
return mod.OpenSpaceClient._extract_zip
|
||||
|
||||
|
||||
class TestExtractZip:
|
||||
def test_normal_file_extracted(self, tmp_path):
|
||||
extract = _get_extract_zip()
|
||||
zip_data = _make_zip({"hello.txt": b"world"})
|
||||
result = extract(zip_data, tmp_path)
|
||||
assert "hello.txt" in result
|
||||
assert (tmp_path / "hello.txt").read_bytes() == b"world"
|
||||
|
||||
def test_dotdot_prefix_blocked(self, tmp_path):
|
||||
extract = _get_extract_zip()
|
||||
zip_data = _make_zip({"../escape.txt": b"bad"})
|
||||
result = extract(zip_data, tmp_path)
|
||||
assert result == []
|
||||
|
||||
def test_nested_traversal_blocked(self, tmp_path):
|
||||
"""The real bug: nested/../../escape.txt bypassed the old startswith check."""
|
||||
extract = _get_extract_zip()
|
||||
zip_data = _make_zip({"nested/../../escape.txt": b"bad"})
|
||||
result = extract(zip_data, tmp_path)
|
||||
assert result == []
|
||||
assert not (tmp_path.parent / "escape.txt").exists()
|
||||
|
||||
def test_absolute_path_blocked(self, tmp_path):
|
||||
extract = _get_extract_zip()
|
||||
zip_data = _make_zip({"/etc/passwd": b"bad"})
|
||||
result = extract(zip_data, tmp_path)
|
||||
assert result == []
|
||||
|
||||
|
||||
class TestImportSkillNameTraversal:
|
||||
def test_malicious_record_name_sanitized(self, tmp_path):
|
||||
from openspace.cloud.client import OpenSpaceClient
|
||||
|
||||
client = OpenSpaceClient.__new__(OpenSpaceClient)
|
||||
target_dir = tmp_path / "skills"
|
||||
target_dir.mkdir()
|
||||
|
||||
malicious_name = "../../escapedir"
|
||||
skill_id = "safe_skill_id"
|
||||
zip_data = _make_zip({"SKILL.md": b"---\nname: test\n---\ncontent"})
|
||||
|
||||
with patch.object(client, "fetch_record", return_value={"name": malicious_name}), \
|
||||
patch.object(client, "download_artifact", return_value=zip_data):
|
||||
result = client.import_skill(skill_id, target_dir)
|
||||
|
||||
assert result["status"] == "success"
|
||||
resolved = Path(result["local_path"]).resolve()
|
||||
assert resolved.is_relative_to(target_dir.resolve())
|
||||
assert not (tmp_path.parent / "escapedir").exists()
|
||||
|
||||
def test_normal_record_name_works(self, tmp_path):
|
||||
from openspace.cloud.client import OpenSpaceClient
|
||||
|
||||
client = OpenSpaceClient.__new__(OpenSpaceClient)
|
||||
target_dir = tmp_path / "skills"
|
||||
target_dir.mkdir()
|
||||
|
||||
zip_data = _make_zip({"SKILL.md": b"---\nname: test\n---\ncontent"})
|
||||
|
||||
with patch.object(client, "fetch_record", return_value={"name": "my-skill"}), \
|
||||
patch.object(client, "download_artifact", return_value=zip_data):
|
||||
result = client.import_skill("some_id", target_dir)
|
||||
|
||||
assert result["status"] == "success"
|
||||
assert (target_dir / "my-skill" / "SKILL.md").exists()
|
||||
Loading…
Add table
Reference in a new issue