"""Tests for workspace MCP tools.""" import pytest from basic_memory.mcp.project_context import get_available_workspaces, set_workspace_provider from basic_memory.mcp.tools.workspaces import list_workspaces from basic_memory.schemas.cloud import WorkspaceInfo from tests.mcp.conftest import ContextState, ctx def _workspace( *, tenant_id: str, workspace_type: str, name: str, role: str, slug: str | None = None, is_default: bool = False, ) -> WorkspaceInfo: return WorkspaceInfo( tenant_id=tenant_id, workspace_type=workspace_type, slug=slug or name.casefold().replace(" ", "-"), name=name, role=role, is_default=is_default, ) @pytest.mark.asyncio async def test_list_workspaces_formats_workspace_rows(monkeypatch): async def fake_get_available_workspaces(context=None): return [ _workspace( tenant_id="11111111-1111-1111-1111-111111111111", workspace_type="personal", slug="personal", name="Personal", role="owner", is_default=True, ), _workspace( tenant_id="22222222-2222-2222-2222-222222222222", workspace_type="organization", slug="team", name="Team", role="editor", ), ] monkeypatch.setattr( "basic_memory.mcp.tools.workspaces.get_available_workspaces", fake_get_available_workspaces, ) result = await list_workspaces() assert "# Available Workspaces (2)" in result assert "Personal (slug=personal, type=personal, role=owner" in result assert "Team (slug=team, type=organization, role=editor" in result @pytest.mark.asyncio async def test_list_workspaces_json_uses_workspace_list_schema(monkeypatch): async def fake_get_available_workspaces(context=None): return [ _workspace( tenant_id="11111111-1111-1111-1111-111111111111", workspace_type="personal", slug="personal", name="Personal", role="owner", is_default=True, ), _workspace( tenant_id="22222222-2222-2222-2222-222222222222", workspace_type="organization", slug="team", name="Team", role="editor", ), ] monkeypatch.setattr( "basic_memory.mcp.tools.workspaces.get_available_workspaces", fake_get_available_workspaces, ) result = await list_workspaces(output_format="json") assert isinstance(result, dict) assert result["count"] == 2 assert result["default_workspace_id"] == "11111111-1111-1111-1111-111111111111" assert result["current_workspace_id"] is None assert result["workspaces"][0]["slug"] == "personal" assert result["workspaces"][0]["is_default"] is True assert result["workspaces"][1]["slug"] == "team" @pytest.mark.asyncio async def test_list_workspaces_handles_empty_list(monkeypatch): async def fake_get_available_workspaces(context=None): return [] monkeypatch.setattr( "basic_memory.mcp.tools.workspaces.get_available_workspaces", fake_get_available_workspaces, ) result = await list_workspaces() assert "# Available Workspaces (1)" in result assert "Personal (slug=personal, type=personal, role=owner, default" in result json_result = await list_workspaces(output_format="json") assert isinstance(json_result, dict) assert json_result["count"] == 1 assert json_result["default_workspace_id"] == "personal" assert json_result["workspaces"][0]["slug"] == "personal" @pytest.mark.asyncio async def test_list_workspaces_oauth_error_bubbles_up(monkeypatch): async def fake_get_available_workspaces(context=None): raise RuntimeError("Workspace discovery requires OAuth login. Run 'bm cloud login' first.") monkeypatch.setattr( "basic_memory.mcp.tools.workspaces.get_available_workspaces", fake_get_available_workspaces, ) with pytest.raises(RuntimeError, match="Workspace discovery requires OAuth login"): await list_workspaces() @pytest.mark.asyncio async def test_list_workspaces_uses_context_cache_path(monkeypatch): context = ContextState() call_count = {"fetches": 0} workspace = _workspace( tenant_id="33333333-3333-3333-3333-333333333333", workspace_type="personal", slug="cached", name="Cached", role="owner", ) async def fake_get_available_workspaces(context=None): assert context is not None cached = await context.get_state("available_workspaces") if cached: return cached call_count["fetches"] += 1 await context.set_state("available_workspaces", [workspace]) return [workspace] monkeypatch.setattr( "basic_memory.mcp.tools.workspaces.get_available_workspaces", fake_get_available_workspaces, ) first = await list_workspaces(context=ctx(context)) second = await list_workspaces(context=ctx(context)) assert "# Available Workspaces (1)" in first assert "# Available Workspaces (1)" in second assert call_count["fetches"] == 1 # --- Workspace provider injection tests --- @pytest.fixture def _reset_workspace_provider(monkeypatch): """Ensure _workspace_provider is reset after each test.""" import basic_memory.mcp.project_context as _mod monkeypatch.setattr(_mod, "_workspace_provider", None) @pytest.mark.asyncio @pytest.mark.usefixtures("_reset_workspace_provider") async def test_get_available_workspaces_uses_provider_when_set(): """When a workspace provider is injected, it is called instead of the control-plane client.""" expected = [ _workspace( tenant_id="aaaa-bbbb", workspace_type="personal", slug="injected", name="Injected", role="owner", is_default=True, ), ] async def fake_provider() -> list[WorkspaceInfo]: return expected set_workspace_provider(fake_provider) result = await get_available_workspaces() assert len(result) == 1 assert result[0].tenant_id == "aaaa-bbbb" assert result[0].name == "Injected" @pytest.mark.asyncio @pytest.mark.usefixtures("_reset_workspace_provider") async def test_get_available_workspaces_falls_back_without_provider(monkeypatch): """Without a provider, get_available_workspaces uses the control-plane client (existing path).""" called = {"control_plane": False} async def fake_control_plane_path(context=None): called["control_plane"] = True return [] # Patch the entire function to avoid needing real credentials monkeypatch.setattr( "basic_memory.mcp.tools.workspaces.get_available_workspaces", fake_control_plane_path, ) result = await list_workspaces() assert called["control_plane"] assert "# Available Workspaces (1)" in result assert "Personal (slug=personal, type=personal, role=owner, default" in result @pytest.mark.asyncio @pytest.mark.usefixtures("_reset_workspace_provider") async def test_get_available_workspaces_provider_caches_in_context(): """Provider results are cached in the MCP context for subsequent calls.""" call_count = {"provider": 0} workspace = _workspace( tenant_id="cccc-dddd", workspace_type="organization", slug="cached-provider", name="Cached Provider", role="editor", ) async def counting_provider() -> list[WorkspaceInfo]: call_count["provider"] += 1 return [workspace] set_workspace_provider(counting_provider) context = ContextState() # First call: provider is invoked, result cached first = await get_available_workspaces(context=ctx(context)) assert len(first) == 1 assert call_count["provider"] == 1 # Second call: served from context cache, provider not called again second = await get_available_workspaces(context=ctx(context)) assert len(second) == 1 assert call_count["provider"] == 1