diff --git a/tests/test_issue3_startup.py b/tests/test_issue3_startup.py deleted file mode 100644 index 9a7b894..0000000 --- a/tests/test_issue3_startup.py +++ /dev/null @@ -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()" diff --git a/tests/test_max_iterations.py b/tests/test_max_iterations.py deleted file mode 100644 index 04e6863..0000000 --- a/tests/test_max_iterations.py +++ /dev/null @@ -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 diff --git a/tests/test_workflow_id.py b/tests/test_workflow_id.py deleted file mode 100644 index 84bf3e1..0000000 --- a/tests/test_workflow_id.py +++ /dev/null @@ -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() diff --git a/tests/test_zip_path_traversal.py b/tests/test_zip_path_traversal.py deleted file mode 100644 index 275176b..0000000 --- a/tests/test_zip_path_traversal.py +++ /dev/null @@ -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()