fix(cli): block team workspace rclone sync

Signed-off-by: Drew Cain <groksrc@gmail.com>
This commit is contained in:
Drew Cain
2026-05-26 12:58:36 -05:00
committed by Paul Hernandez
parent ff5d872a8c
commit a2276a8a04
2 changed files with 172 additions and 1 deletions
@@ -26,11 +26,20 @@ from basic_memory.cli.commands.routing import force_routing
from basic_memory.config import 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
from basic_memory.schemas.cloud import WorkspaceInfo
from basic_memory.schemas.project_info import ProjectItem
from basic_memory.utils import generate_permalink, normalize_project_path
console = Console()
TEAM_WORKSPACE_SYNC_UNSUPPORTED = (
"Local rclone sync/bisync is supported only for Personal workspaces.\n"
"Team workspaces are accessed through the cloud API/MCP and do not support "
"local multi-user bisync.\n"
"Use `bm project list --workspace <workspace>` to inspect Team projects."
)
# --- Shared helpers ---
@@ -52,6 +61,54 @@ def _require_cloud_credentials(config) -> None:
raise typer.Exit(1)
async def _get_workspace_for_project(name: str, config) -> WorkspaceInfo:
"""Resolve the cloud workspace targeted by a project-scoped sync command."""
workspaces = await get_available_workspaces()
if not workspaces:
raise ValueError("No accessible cloud workspaces found for this account")
entry = config.projects.get(name)
workspace_id = entry.workspace_id if entry and entry.workspace_id else config.default_workspace
if workspace_id:
workspace = next(
(item for item in workspaces if item.tenant_id == workspace_id),
None,
)
if workspace is None:
raise ValueError(
f"Configured workspace '{workspace_id}' for project '{name}' is not accessible"
)
return workspace
default_workspaces = [item for item in workspaces if item.is_default]
if len(default_workspaces) == 1:
return default_workspaces[0]
if len(workspaces) == 1:
return workspaces[0]
raise ValueError(
f"Project '{name}' does not have an unambiguous cloud workspace. "
"Set a default workspace with `bm cloud workspace set-default <workspace>` "
"or attach the project with `bm project set-cloud <name> --workspace <workspace>`."
)
def _require_personal_workspace(name: str, config) -> WorkspaceInfo:
"""Exit before rclone work when the target workspace is not personal."""
try:
workspace = run_with_cleanup(_get_workspace_for_project(name, config))
except Exception as exc:
console.print(f"[red]Error resolving workspace for project '{name}': {exc}[/red]")
raise typer.Exit(1)
if workspace.workspace_type != "personal":
console.print(f"[red]{TEAM_WORKSPACE_SYNC_UNSUPPORTED}[/red]")
raise typer.Exit(1)
return workspace
async def _get_cloud_project(name: str) -> ProjectItem | None:
"""Fetch a project by name from the cloud API."""
async with get_client(project_name=name) as client:
@@ -103,6 +160,7 @@ def sync_project_command(
"""
config = ConfigManager().config
_require_cloud_credentials(config)
_require_personal_workspace(name, config)
try:
# Get tenant info for bucket name
@@ -152,6 +210,7 @@ def bisync_project_command(
"""
config = ConfigManager().config
_require_cloud_credentials(config)
_require_personal_workspace(name, config)
try:
# Get tenant info for bucket name
@@ -210,6 +269,7 @@ def check_project_command(
"""
config = ConfigManager().config
_require_cloud_credentials(config)
_require_personal_workspace(name, config)
try:
# Get tenant info for bucket name
@@ -253,6 +313,10 @@ def bisync_reset(
"""
import shutil
config = ConfigManager().config
if _has_cloud_credentials(config):
_require_personal_workspace(name, config)
try:
state_path = get_project_bisync_state(name)
@@ -288,6 +352,7 @@ def setup_project_sync(
config_manager = ConfigManager()
config = config_manager.config
_require_cloud_credentials(config)
_require_personal_workspace(name, config)
async def _verify_project_exists():
"""Verify the project exists on cloud by listing all projects."""
+107 -1
View File
@@ -7,7 +7,8 @@ import pytest
from typer.testing import CliRunner
from basic_memory.cli.app import app
from basic_memory.config import ProjectMode
from basic_memory.config import ProjectEntry, ProjectMode
from basic_memory.schemas.cloud import WorkspaceInfo
runner = CliRunner()
@@ -28,6 +29,9 @@ def test_cloud_sync_commands_skip_explicit_cloud_project_sync(monkeypatch, argv,
config_manager.save_config(config)
monkeypatch.setattr(project_sync_command, "_require_cloud_credentials", lambda _config: None)
monkeypatch.setattr(
project_sync_command, "_require_personal_workspace", lambda _name, _config: None
)
monkeypatch.setattr(
project_sync_command,
"get_mount_info",
@@ -63,6 +67,9 @@ def test_cloud_bisync_fails_fast_when_sync_entry_disappears(monkeypatch, config_
config_manager.save_config(config)
monkeypatch.setattr(project_sync_command, "_require_cloud_credentials", lambda _config: None)
monkeypatch.setattr(
project_sync_command, "_require_personal_workspace", lambda _name, _config: None
)
monkeypatch.setattr(
project_sync_command,
"get_mount_info",
@@ -88,5 +95,104 @@ def test_cloud_bisync_fails_fast_when_sync_entry_disappears(monkeypatch, config_
assert "unexpectedly missing after validation" in result.output
@pytest.mark.parametrize(
"argv",
[
["cloud", "sync", "--name", "research"],
["cloud", "bisync", "--name", "research"],
["cloud", "check", "--name", "research"],
["cloud", "bisync-reset", "research"],
["cloud", "sync-setup", "research", "/tmp/research"],
],
)
def test_cloud_sync_commands_block_organization_workspace(monkeypatch, argv, config_manager):
"""Rclone sync commands should fail before setup/execution for Team workspaces."""
project_sync_command = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
config = config_manager.load_config()
config.cloud_api_key = "bmc_test"
config.projects["research"] = ProjectEntry(
path="/tmp/research",
mode=ProjectMode.CLOUD,
workspace_id="team-tenant",
local_sync_path="/tmp/research",
)
config_manager.save_config(config)
monkeypatch.setattr(
project_sync_command,
"get_available_workspaces",
lambda: _async_value([_workspace("team-tenant", "organization", "team")]),
)
monkeypatch.setattr(
project_sync_command,
"get_mount_info",
lambda: pytest.fail("workspace guard should run before mount lookup"),
)
result = runner.invoke(app, argv)
assert result.exit_code == 1, result.output
assert "Local rclone sync/bisync is supported only for Personal workspaces" in result.output
assert "Team workspaces are accessed through the cloud API/MCP" in result.output
def test_require_personal_workspace_allows_personal_workspace(monkeypatch, config_manager):
"""Personal workspaces keep the existing rclone sync path available."""
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="personal-tenant",
local_sync_path="/tmp/research",
)
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_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")
monkeypatch.setattr(
project_sync_command,
"get_available_workspaces",
lambda: pytest.fail("workspace lookup requires cloud credentials"),
)
monkeypatch.setattr(
project_sync_command,
"get_project_bisync_state",
lambda _name: tmp_path / "missing-state",
)
result = runner.invoke(app, ["cloud", "bisync-reset", "research"])
assert result.exit_code == 0, result.output
assert "No bisync state found for project 'research'" in result.output
async def _async_value(value):
return value
def _workspace(tenant_id: str, workspace_type: str, slug: str) -> WorkspaceInfo:
return WorkspaceInfo(
tenant_id=tenant_id,
workspace_type=workspace_type,
slug=slug,
name=slug.title(),
role="owner",
is_default=False,
has_active_subscription=True,
)