admin/basicmachines-co-basic-memory
1
0
Fork 0
mirror of https://github.com/basicmachines-co/basic-memory synced 2026-06-21 13:47:35 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
codex/basic-machines-agent-style
basicmachines-co-basic-memory/tests/cli/cloud/test_project_sync_command.py
T
phernandez e23363606a feat(cli): add Team-safe cloud push/pull and gate sync to Personal workspaces 
Adds `bm cloud push` and `bm cloud pull` as git-style, fail-safe transfer
primitives that are usable on Team workspaces (issue #858), and restricts the
destructive `bm cloud sync` mirror to Personal workspaces.

Why
- `bm cloud sync` is a destructive local->cloud mirror; on a shared Team bucket
  it can delete a teammate's files. The only pull path was two-way `bisync`,
  which is already Personal-only (#849).
- Teams need a safe way to fetch teammates' notes and add their own without one
  stale local tree becoming authoritative for shared cloud state.

What
- push = `rclone copy` local->cloud, pull = `rclone copy` cloud->local. Both are
  additive (never delete on the destination), so neither can damage shared state.
- Conflicts (a file that differs on both sides) abort by default and list the
  paths, like git refusing to clobber local changes / rejecting a stale push.
  `--on-conflict {fail|keep-local|keep-cloud|keep-both}` lets the user decide;
  no vague --force, no silent winner.
- `sync` now requires a Personal workspace; its guard and the bisync guard point
  Team users at push/pull.

Limitations (surfaced in --help and command output, tracked by #862):
- No sync baseline yet, so deletions are not propagated and every divergence is
  treated as a conflict rather than auto-resolved. Real three-way merge needs
  the per-client manifest + Tigris snapshot baseline designed in #862.

Implementation
- rclone_commands.py: shared `_build_transfer_cmd`/`_transfer_endpoints`;
  refactor `project_sync` onto them (no behavior change); add `project_diff`
  (conflict detection via `rclone check --combined`), `project_copy`,
  `project_copy_file`, and `project_transfer` (strategy dispatch).
- project_sync.py: new `push`/`pull` commands (ungated, Team-safe) with
  `--on-conflict`/`--dry-run`; gate `sync` to Personal via a per-command guard
  message.
- Tests at the rclone-argv and CLI-command levels; existing sync/bisync tests
  updated for the new gating and messages.

Signed-off-by: phernandez <paul@basicmachines.co>
2026-06-08 11:35:13 -05:00

556 lines
21 KiB
Python
Raw Permalink Blame History

"""Tests for cloud sync and bisync command behavior."""
import importlib
from types import SimpleNamespace
import pytest
import typer
from typer.testing import CliRunner
from basic_memory.cli.app import app
from basic_memory.cli.commands.cloud.rclone_commands import TransferPlan
from basic_memory.config import ProjectEntry, ProjectMode
from basic_memory.schemas.cloud import WorkspaceInfo
runner = CliRunner()
@pytest.mark.parametrize(
"argv",
[
["cloud", "sync", "--name", "research"],
["cloud", "bisync", "--name", "research"],
# --project is an alias for --name (issue #817)
["cloud", "sync", "--project", "research"],
["cloud", "bisync", "--project", "research"],
],
)
def test_cloud_sync_commands_skip_explicit_cloud_project_sync(monkeypatch, argv, config_manager):
"""Cloud sync commands should not trigger an extra explicit cloud project sync."""
project_sync_command = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
config = config_manager.load_config()
config.set_project_mode("research", ProjectMode.CLOUD)
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, **_kwargs: None
)
monkeypatch.setattr(
project_sync_command,
"get_mount_info",
lambda: _async_value(SimpleNamespace(bucket_name="tenant-bucket")),
)
monkeypatch.setattr(
project_sync_command,
"_get_cloud_project",
lambda _name: _async_value(
SimpleNamespace(name="research", external_id="external-project-id", path="research")
),
)
monkeypatch.setattr(
project_sync_command,
"_get_sync_project",
lambda _name, _config, _project_data: (SimpleNamespace(name="research"), "/tmp/research"),
)
monkeypatch.setattr(project_sync_command, "project_sync", lambda *args, **kwargs: True)
monkeypatch.setattr(project_sync_command, "project_bisync", lambda *args, **kwargs: True)
result = runner.invoke(app, argv)
assert result.exit_code == 0, result.output
assert "Database sync initiated" not in result.output
def test_cloud_bisync_fails_fast_when_sync_entry_disappears(monkeypatch, config_manager):
"""Bisync should raise a runtime error when validated sync config vanishes before persistence."""
project_sync_command = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
config = config_manager.load_config()
config.projects.pop("research", None)
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, **_kwargs: None
)
monkeypatch.setattr(
project_sync_command,
"get_mount_info",
lambda: _async_value(SimpleNamespace(bucket_name="tenant-bucket")),
)
monkeypatch.setattr(
project_sync_command,
"_get_cloud_project",
lambda _name: _async_value(
SimpleNamespace(name="research", external_id="external-project-id", path="research")
),
)
monkeypatch.setattr(
project_sync_command,
"_get_sync_project",
lambda _name, _config, _project_data: (SimpleNamespace(name="research"), "/tmp/research"),
)
monkeypatch.setattr(project_sync_command, "project_bisync", lambda *args, **kwargs: True)
result = runner.invoke(app, ["cloud", "bisync", "--name", "research"])
assert result.exit_code == 1, result.output
assert "unexpectedly missing after validation" in result.output
@pytest.mark.parametrize(
"argv",
[
["cloud", "bisync", "--name", "research"],
["cloud", "bisync-reset", "research"],
],
)
def test_cloud_bisync_commands_block_organization_workspace(monkeypatch, argv, config_manager):
"""Bisync 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"),
)
monkeypatch.setattr(
project_sync_command,
"get_project_bisync_state",
lambda _name: pytest.fail("workspace guard should run before bisync state lookup"),
)
result = runner.invoke(app, argv)
assert result.exit_code == 1, result.output
output = " ".join(result.output.split())
assert "The bisync operation is only supported on Personal workspaces" in output
assert "bm cloud pull --name research" in output
assert "bm cloud push --name research" in output
def test_cloud_sync_blocks_organization_workspace(monkeypatch, config_manager):
"""The destructive mirror `sync` is now Personal-only and blocks 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, ["cloud", "sync", "--name", "research"])
assert result.exit_code == 1, result.output
output = " ".join(result.output.split())
assert "only supported on Personal workspaces" in output
assert "bm cloud push --name research" in output
assert "bm cloud pull --name research" in output
def test_cloud_sync_allows_personal_workspace(monkeypatch, config_manager):
"""Personal workspaces keep the one-way mirror sync available."""
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="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")]),
)
monkeypatch.setattr(
project_sync_command,
"get_mount_info",
lambda: _async_value(SimpleNamespace(bucket_name="tenant-bucket")),
)
monkeypatch.setattr(
project_sync_command,
"_get_cloud_project",
lambda _name: _async_value(
SimpleNamespace(name="research", external_id="external-project-id", path="research")
),
)
monkeypatch.setattr(
project_sync_command,
"_get_sync_project",
lambda _name, _config, _project_data: (SimpleNamespace(name="research"), "/tmp/research"),
)
monkeypatch.setattr(project_sync_command, "project_sync", lambda *args, **kwargs: True)
result = runner.invoke(app, ["cloud", "sync", "--name", "research"])
assert result.exit_code == 0, result.output
assert "research synced successfully" 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_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")
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
def _stub_transfer_env(monkeypatch, module, *, plan, transfer_result=True, recorder=None):
"""Stub the push/pull dependency chain so only diff/transfer logic is exercised."""
monkeypatch.setattr(module, "_require_cloud_credentials", lambda _config: None)
monkeypatch.setattr(
module,
"get_mount_info",
lambda: _async_value(SimpleNamespace(bucket_name="tenant-bucket")),
)
monkeypatch.setattr(
module,
"_get_cloud_project",
lambda _name: _async_value(SimpleNamespace(name="research", path="research")),
)
monkeypatch.setattr(
module,
"_get_sync_project",
lambda _name, _config, _project_data: (SimpleNamespace(name="research"), "/tmp/research"),
)
monkeypatch.setattr(module, "project_diff", lambda *args, **kwargs: plan)
def _fake_transfer(*args, **kwargs):
if recorder is not None:
recorder["args"] = args
recorder["kwargs"] = kwargs
return transfer_result
monkeypatch.setattr(module, "project_transfer", _fake_transfer)
def test_cloud_pull_aborts_on_conflict_by_default(monkeypatch, config_manager):
"""Pull refuses to clobber: it lists conflicts and exits without transferring."""
module = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
plan = TransferPlan(new=["new.md"], conflicts=["notes/dup.md"], dest_only=[], errors=[])
recorder: dict = {}
_stub_transfer_env(monkeypatch, module, plan=plan, recorder=recorder)
result = runner.invoke(app, ["cloud", "pull", "--name", "research"])
assert result.exit_code == 1, result.output
output = " ".join(result.output.split())
assert "notes/dup.md" in output
assert "--on-conflict keep-cloud" in output
assert "args" not in recorder # transfer never ran
def test_cloud_pull_clean_transfers(monkeypatch, config_manager):
"""With no conflicts, pull proceeds and reports success."""
module = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
plan = TransferPlan(new=["new.md"], conflicts=[], dest_only=["local-only.md"], errors=[])
recorder: dict = {}
_stub_transfer_env(monkeypatch, module, plan=plan, recorder=recorder)
result = runner.invoke(app, ["cloud", "pull", "--name", "research"])
assert result.exit_code == 0, result.output
output = " ".join(result.output.lower().split())
assert "research pull completed successfully" in output
# Deletions are surfaced, not propagated
assert "deletions are not propagated" in output
assert recorder["kwargs"]["strategy"] == "fail"
assert recorder["args"][2] == "pull"
def test_cloud_pull_keep_cloud_resolves_conflict(monkeypatch, config_manager):
"""An explicit --on-conflict strategy lets pull proceed through conflicts."""
module = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
plan = TransferPlan(new=[], conflicts=["notes/dup.md"], dest_only=[], errors=[])
recorder: dict = {}
_stub_transfer_env(monkeypatch, module, plan=plan, recorder=recorder)
result = runner.invoke(
app, ["cloud", "pull", "--name", "research", "--on-conflict", "keep-cloud"]
)
assert result.exit_code == 0, result.output
assert recorder["kwargs"]["strategy"] == "keep-cloud"
def test_cloud_pull_aborts_on_compare_errors(monkeypatch, config_manager):
"""If rclone cannot read/hash files, pull aborts before transferring."""
module = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
plan = TransferPlan(new=[], conflicts=[], dest_only=[], errors=["bad.md"])
recorder: dict = {}
_stub_transfer_env(monkeypatch, module, plan=plan, recorder=recorder)
result = runner.invoke(app, ["cloud", "pull", "--name", "research"])
assert result.exit_code == 1, result.output
assert "could not compare" in result.output
assert "args" not in recorder # transfer never ran
def test_cloud_push_aborts_on_conflict_by_default(monkeypatch, config_manager):
"""Push aborts on conflicts like a rejected git push (pull first)."""
module = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
plan = TransferPlan(new=["new.md"], conflicts=["notes/dup.md"], dest_only=[], errors=[])
recorder: dict = {}
_stub_transfer_env(monkeypatch, module, plan=plan, recorder=recorder)
result = runner.invoke(app, ["cloud", "push", "--name", "research"])
assert result.exit_code == 1, result.output
assert "notes/dup.md" in result.output
assert "args" not in recorder
def test_cloud_push_keep_local_resolves_conflict(monkeypatch, config_manager):
"""Push with --on-conflict keep-local overwrites cloud and reports the direction."""
module = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
plan = TransferPlan(new=[], conflicts=["notes/dup.md"], dest_only=[], errors=[])
recorder: dict = {}
_stub_transfer_env(monkeypatch, module, plan=plan, recorder=recorder)
result = runner.invoke(
app, ["cloud", "push", "--name", "research", "--on-conflict", "keep-local"]
)
assert result.exit_code == 0, result.output
assert recorder["kwargs"]["strategy"] == "keep-local"
assert recorder["args"][2] == "push"
def test_cloud_push_allows_organization_workspace(monkeypatch, config_manager):
"""push is additive and Team-safe — it must not invoke the Personal-only guard."""
module = 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)
plan = TransferPlan(new=["new.md"], conflicts=[], dest_only=[], errors=[])
_stub_transfer_env(monkeypatch, module, plan=plan)
monkeypatch.setattr(
module,
"get_available_workspaces",
lambda: pytest.fail("push/pull must not gate on workspace type"),
)
result = runner.invoke(app, ["cloud", "push", "--name", "research"])
assert result.exit_code == 0, result.output
assert "research push completed successfully" in result.output
async def _async_value(value):
return value
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=is_default,
has_active_subscription=True,
)
Reference in New Issue View Git Blame Copy Permalink