From 1acec0a69d960ed052896b6334e8f83dc9b9d4f9 Mon Sep 17 00:00:00 2001 From: Drew Cain Date: Tue, 26 May 2026 13:14:15 -0500 Subject: [PATCH] test(cli): cover workspace sync guard branches Signed-off-by: Drew Cain --- .../cli/commands/cloud/project_sync.py | 12 +- tests/cli/cloud/test_project_sync_command.py | 125 +++++++++++++++++- 2 files changed, 129 insertions(+), 8 deletions(-) diff --git a/src/basic_memory/cli/commands/cloud/project_sync.py b/src/basic_memory/cli/commands/cloud/project_sync.py index 5277ddc1..66a0628f 100644 --- a/src/basic_memory/cli/commands/cloud/project_sync.py +++ b/src/basic_memory/cli/commands/cloud/project_sync.py @@ -23,7 +23,7 @@ from basic_memory.cli.commands.cloud.rclone_commands import ( ) from basic_memory.cli.commands.command_utils import run_with_cleanup from basic_memory.cli.commands.routing import force_routing -from basic_memory.config import ConfigManager, ProjectEntry +from basic_memory.config import BasicMemoryConfig, ConfigManager, ProjectEntry from basic_memory.mcp.async_client import get_client from basic_memory.mcp.clients import ProjectClient from basic_memory.mcp.project_context import get_available_workspaces @@ -44,14 +44,14 @@ TEAM_WORKSPACE_SYNC_UNSUPPORTED = ( # --- Shared helpers --- -def _has_cloud_credentials(config) -> bool: +def _has_cloud_credentials(config: BasicMemoryConfig) -> bool: """Return whether cloud credentials are available (API key or OAuth token).""" from basic_memory.config import has_cloud_credentials return has_cloud_credentials(config) -def _require_cloud_credentials(config) -> None: +def _require_cloud_credentials(config: BasicMemoryConfig) -> None: """Exit with actionable guidance when cloud credentials are missing.""" if _has_cloud_credentials(config): return @@ -61,7 +61,7 @@ def _require_cloud_credentials(config) -> None: raise typer.Exit(1) -async def _get_workspace_for_project(name: str, config) -> WorkspaceInfo: +async def _get_workspace_for_project(name: str, config: BasicMemoryConfig) -> WorkspaceInfo: """Resolve the cloud workspace targeted by a project-scoped sync command.""" workspaces = await get_available_workspaces() if not workspaces: @@ -94,7 +94,7 @@ async def _get_workspace_for_project(name: str, config) -> WorkspaceInfo: ) -def _require_personal_workspace(name: str, config) -> WorkspaceInfo: +def _require_personal_workspace(name: str, config: BasicMemoryConfig) -> WorkspaceInfo: """Exit before rclone work when the target workspace is not personal.""" try: workspace = run_with_cleanup(_get_workspace_for_project(name, config)) @@ -120,7 +120,7 @@ async def _get_cloud_project(name: str) -> ProjectItem | None: def _get_sync_project( - name: str, config, project_data: ProjectItem + name: str, config: BasicMemoryConfig, project_data: ProjectItem ) -> tuple[SyncProject, str | None]: """Build a SyncProject and resolve local_sync_path from config. diff --git a/tests/cli/cloud/test_project_sync_command.py b/tests/cli/cloud/test_project_sync_command.py index 0a1241ce..2375cbca 100644 --- a/tests/cli/cloud/test_project_sync_command.py +++ b/tests/cli/cloud/test_project_sync_command.py @@ -4,6 +4,7 @@ import importlib from types import SimpleNamespace import pytest +import typer from typer.testing import CliRunner from basic_memory.cli.app import app @@ -161,6 +162,124 @@ def test_require_personal_workspace_allows_personal_workspace(monkeypatch, confi assert workspace.tenant_id == "personal-tenant" +def test_require_personal_workspace_uses_default_workspace(monkeypatch, config_manager): + """When no project workspace is set, the single cloud default is used.""" + project_sync_command = importlib.import_module("basic_memory.cli.commands.cloud.project_sync") + + config = config_manager.load_config() + config.default_workspace = None + config.projects["research"] = ProjectEntry(path="/tmp/research", mode=ProjectMode.CLOUD) + config_manager.save_config(config) + + monkeypatch.setattr( + project_sync_command, + "get_available_workspaces", + lambda: _async_value( + [ + _workspace("team-tenant", "organization", "team"), + _workspace("personal-tenant", "personal", "personal", is_default=True), + ] + ), + ) + + workspace = project_sync_command._require_personal_workspace("research", config) + + assert workspace.tenant_id == "personal-tenant" + + +def test_require_personal_workspace_uses_single_workspace(monkeypatch, config_manager): + """A single accessible workspace is unambiguous even when none is marked default.""" + project_sync_command = importlib.import_module("basic_memory.cli.commands.cloud.project_sync") + + config = config_manager.load_config() + config.default_workspace = None + config.projects["research"] = ProjectEntry(path="/tmp/research", mode=ProjectMode.CLOUD) + config_manager.save_config(config) + + monkeypatch.setattr( + project_sync_command, + "get_available_workspaces", + lambda: _async_value([_workspace("personal-tenant", "personal", "personal")]), + ) + + workspace = project_sync_command._require_personal_workspace("research", config) + + assert workspace.tenant_id == "personal-tenant" + + +def test_require_personal_workspace_reports_no_accessible_workspaces(monkeypatch, config_manager): + """Workspace resolution exits with a clear error when the account has no workspaces.""" + project_sync_command = importlib.import_module("basic_memory.cli.commands.cloud.project_sync") + + config = config_manager.load_config() + config.default_workspace = None + config.projects["research"] = ProjectEntry(path="/tmp/research", mode=ProjectMode.CLOUD) + config_manager.save_config(config) + + monkeypatch.setattr( + project_sync_command, + "get_available_workspaces", + lambda: _async_value([]), + ) + + with pytest.raises(typer.Exit) as exc_info: + project_sync_command._require_personal_workspace("research", config) + + assert exc_info.value.exit_code == 1 + + +def test_require_personal_workspace_reports_inaccessible_configured_workspace( + monkeypatch, config_manager +): + """A configured workspace id must be present in the accessible workspace list.""" + project_sync_command = importlib.import_module("basic_memory.cli.commands.cloud.project_sync") + + config = config_manager.load_config() + config.projects["research"] = ProjectEntry( + path="/tmp/research", + mode=ProjectMode.CLOUD, + workspace_id="missing-tenant", + ) + config_manager.save_config(config) + + monkeypatch.setattr( + project_sync_command, + "get_available_workspaces", + lambda: _async_value([_workspace("personal-tenant", "personal", "personal")]), + ) + + with pytest.raises(typer.Exit) as exc_info: + project_sync_command._require_personal_workspace("research", config) + + assert exc_info.value.exit_code == 1 + + +def test_require_personal_workspace_reports_ambiguous_workspace(monkeypatch, config_manager): + """Multiple accessible workspaces need an explicit project or account default.""" + project_sync_command = importlib.import_module("basic_memory.cli.commands.cloud.project_sync") + + config = config_manager.load_config() + config.default_workspace = None + config.projects["research"] = ProjectEntry(path="/tmp/research", mode=ProjectMode.CLOUD) + config_manager.save_config(config) + + monkeypatch.setattr( + project_sync_command, + "get_available_workspaces", + lambda: _async_value( + [ + _workspace("personal-tenant-a", "personal", "personal-a"), + _workspace("personal-tenant-b", "personal", "personal-b"), + ] + ), + ) + + with pytest.raises(typer.Exit) as exc_info: + project_sync_command._require_personal_workspace("research", config) + + assert exc_info.value.exit_code == 1 + + def test_bisync_reset_skips_workspace_check_without_credentials(monkeypatch, tmp_path): """Resetting local bisync state stays harmless when no cloud credentials exist.""" project_sync_command = importlib.import_module("basic_memory.cli.commands.cloud.project_sync") @@ -186,13 +305,15 @@ async def _async_value(value): return value -def _workspace(tenant_id: str, workspace_type: str, slug: str) -> WorkspaceInfo: +def _workspace( + tenant_id: str, workspace_type: str, slug: str, *, is_default: bool = False +) -> WorkspaceInfo: return WorkspaceInfo( tenant_id=tenant_id, workspace_type=workspace_type, slug=slug, name=slug.title(), role="owner", - is_default=False, + is_default=is_default, has_active_subscription=True, )