mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
Drew Cain
482 lines
16 KiB
Python
482 lines
16 KiB
Python
"""Test project-scoped rclone commands."""
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from basic_memory.cli.commands.cloud.rclone_commands import (
|
|
RcloneError,
|
|
SyncProject,
|
|
bisync_initialized,
|
|
check_rclone_installed,
|
|
get_project_bisync_state,
|
|
get_project_remote,
|
|
project_bisync,
|
|
project_check,
|
|
project_ls,
|
|
project_sync,
|
|
)
|
|
|
|
|
|
def test_sync_project_dataclass():
|
|
"""Test SyncProject dataclass."""
|
|
project = SyncProject(
|
|
name="research",
|
|
path="app/data/research",
|
|
local_sync_path="/Users/test/research",
|
|
)
|
|
|
|
assert project.name == "research"
|
|
assert project.path == "app/data/research"
|
|
assert project.local_sync_path == "/Users/test/research"
|
|
|
|
|
|
def test_sync_project_optional_local_path():
|
|
"""Test SyncProject with optional local_sync_path."""
|
|
project = SyncProject(
|
|
name="research",
|
|
path="app/data/research",
|
|
)
|
|
|
|
assert project.name == "research"
|
|
assert project.path == "app/data/research"
|
|
assert project.local_sync_path is None
|
|
|
|
|
|
def test_get_project_remote():
|
|
"""Test building rclone remote path with normalized path."""
|
|
# Path comes from API already normalized (no /app/data/ prefix)
|
|
project = SyncProject(name="research", path="/research")
|
|
|
|
remote = get_project_remote(project, "my-bucket")
|
|
|
|
assert remote == "basic-memory-cloud:my-bucket/research"
|
|
|
|
|
|
def test_get_project_remote_strips_app_data_prefix():
|
|
"""Test that /app/data/ prefix is stripped from cloud path."""
|
|
# If API returns path with /app/data/, it should be stripped
|
|
project = SyncProject(name="research", path="/app/data/research")
|
|
|
|
remote = get_project_remote(project, "my-bucket")
|
|
|
|
# Should strip /app/data/ prefix to get actual S3 path
|
|
assert remote == "basic-memory-cloud:my-bucket/research"
|
|
|
|
|
|
def test_get_project_bisync_state():
|
|
"""Test getting bisync state directory path."""
|
|
state_path = get_project_bisync_state("research")
|
|
|
|
expected = Path.home() / ".basic-memory" / "bisync-state" / "research"
|
|
assert state_path == expected
|
|
|
|
|
|
def test_bisync_initialized_false_when_not_exists(tmp_path, monkeypatch):
|
|
"""Test bisync_initialized returns False when state doesn't exist."""
|
|
# Patch to use tmp directory
|
|
monkeypatch.setattr(
|
|
"basic_memory.cli.commands.cloud.rclone_commands.get_project_bisync_state",
|
|
lambda project_name: tmp_path / project_name,
|
|
)
|
|
|
|
assert bisync_initialized("research") is False
|
|
|
|
|
|
def test_bisync_initialized_false_when_empty(tmp_path, monkeypatch):
|
|
"""Test bisync_initialized returns False when state directory is empty."""
|
|
state_dir = tmp_path / "research"
|
|
state_dir.mkdir()
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.cli.commands.cloud.rclone_commands.get_project_bisync_state",
|
|
lambda project_name: tmp_path / project_name,
|
|
)
|
|
|
|
assert bisync_initialized("research") is False
|
|
|
|
|
|
def test_bisync_initialized_true_when_has_files(tmp_path, monkeypatch):
|
|
"""Test bisync_initialized returns True when state has files."""
|
|
state_dir = tmp_path / "research"
|
|
state_dir.mkdir()
|
|
(state_dir / "state.lst").touch()
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.cli.commands.cloud.rclone_commands.get_project_bisync_state",
|
|
lambda project_name: tmp_path / project_name,
|
|
)
|
|
|
|
assert bisync_initialized("research") is True
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
|
|
def test_project_sync_success(mock_run, mock_is_installed):
|
|
"""Test successful project sync."""
|
|
mock_is_installed.return_value = True
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
|
|
project = SyncProject(
|
|
name="research",
|
|
path="/research", # Normalized path from API
|
|
local_sync_path="/tmp/research",
|
|
)
|
|
|
|
result = project_sync(project, "my-bucket", dry_run=True)
|
|
|
|
assert result is True
|
|
mock_run.assert_called_once()
|
|
|
|
# Check command arguments
|
|
cmd = mock_run.call_args[0][0]
|
|
assert cmd[0] == "rclone"
|
|
assert cmd[1] == "sync"
|
|
# Use Path for cross-platform comparison (Windows uses backslashes)
|
|
assert Path(cmd[2]) == Path("/tmp/research")
|
|
assert cmd[3] == "basic-memory-cloud:my-bucket/research"
|
|
assert "--dry-run" in cmd
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
|
|
def test_project_sync_with_verbose(mock_run, mock_is_installed):
|
|
"""Test project sync with verbose flag."""
|
|
mock_is_installed.return_value = True
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
|
|
project = SyncProject(
|
|
name="research",
|
|
path="app/data/research",
|
|
local_sync_path="/tmp/research",
|
|
)
|
|
|
|
project_sync(project, "my-bucket", verbose=True)
|
|
|
|
cmd = mock_run.call_args[0][0]
|
|
assert "--verbose" in cmd
|
|
assert "--progress" not in cmd
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
|
|
def test_project_sync_with_progress(mock_run, mock_is_installed):
|
|
"""Test project sync with progress (default)."""
|
|
mock_is_installed.return_value = True
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
|
|
project = SyncProject(
|
|
name="research",
|
|
path="app/data/research",
|
|
local_sync_path="/tmp/research",
|
|
)
|
|
|
|
project_sync(project, "my-bucket")
|
|
|
|
cmd = mock_run.call_args[0][0]
|
|
assert "--progress" in cmd
|
|
assert "--verbose" not in cmd
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
def test_project_sync_no_local_path(mock_is_installed):
|
|
"""Test project sync raises error when local_sync_path not configured."""
|
|
mock_is_installed.return_value = True
|
|
project = SyncProject(name="research", path="app/data/research")
|
|
|
|
with pytest.raises(RcloneError) as exc_info:
|
|
project_sync(project, "my-bucket")
|
|
|
|
assert "no local_sync_path configured" in str(exc_info.value)
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.bisync_initialized")
|
|
def test_project_bisync_success(mock_bisync_init, mock_run, mock_is_installed):
|
|
"""Test successful project bisync."""
|
|
mock_is_installed.return_value = True
|
|
mock_bisync_init.return_value = True # Already initialized
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
|
|
project = SyncProject(
|
|
name="research",
|
|
path="app/data/research",
|
|
local_sync_path="/tmp/research",
|
|
)
|
|
|
|
result = project_bisync(project, "my-bucket")
|
|
|
|
assert result is True
|
|
mock_run.assert_called_once()
|
|
|
|
# Check command arguments
|
|
cmd = mock_run.call_args[0][0]
|
|
assert cmd[0] == "rclone"
|
|
assert cmd[1] == "bisync"
|
|
assert "--conflict-resolve=newer" in cmd
|
|
assert "--max-delete=25" in cmd
|
|
assert "--resilient" in cmd
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.bisync_initialized")
|
|
def test_project_bisync_requires_resync_first_time(mock_bisync_init, mock_run, mock_is_installed):
|
|
"""Test that first bisync requires --resync flag."""
|
|
mock_is_installed.return_value = True
|
|
mock_bisync_init.return_value = False # Not initialized
|
|
|
|
project = SyncProject(
|
|
name="research",
|
|
path="app/data/research",
|
|
local_sync_path="/tmp/research",
|
|
)
|
|
|
|
with pytest.raises(RcloneError) as exc_info:
|
|
project_bisync(project, "my-bucket")
|
|
|
|
assert "requires --resync" in str(exc_info.value)
|
|
mock_run.assert_not_called()
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.bisync_initialized")
|
|
def test_project_bisync_with_resync_flag(mock_bisync_init, mock_run, mock_is_installed):
|
|
"""Test bisync with --resync flag for first time."""
|
|
mock_is_installed.return_value = True
|
|
mock_bisync_init.return_value = False # Not initialized
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
|
|
project = SyncProject(
|
|
name="research",
|
|
path="app/data/research",
|
|
local_sync_path="/tmp/research",
|
|
)
|
|
|
|
result = project_bisync(project, "my-bucket", resync=True)
|
|
|
|
assert result is True
|
|
cmd = mock_run.call_args[0][0]
|
|
assert "--resync" in cmd
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.bisync_initialized")
|
|
def test_project_bisync_dry_run_skips_init_check(mock_bisync_init, mock_run, mock_is_installed):
|
|
"""Test that dry-run skips initialization check."""
|
|
mock_is_installed.return_value = True
|
|
mock_bisync_init.return_value = False # Not initialized
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
|
|
project = SyncProject(
|
|
name="research",
|
|
path="app/data/research",
|
|
local_sync_path="/tmp/research",
|
|
)
|
|
|
|
# Should not raise error even though not initialized
|
|
result = project_bisync(project, "my-bucket", dry_run=True)
|
|
|
|
assert result is True
|
|
cmd = mock_run.call_args[0][0]
|
|
assert "--dry-run" in cmd
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
def test_project_bisync_no_local_path(mock_is_installed):
|
|
"""Test project bisync raises error when local_sync_path not configured."""
|
|
mock_is_installed.return_value = True
|
|
project = SyncProject(name="research", path="app/data/research")
|
|
|
|
with pytest.raises(RcloneError) as exc_info:
|
|
project_bisync(project, "my-bucket")
|
|
|
|
assert "no local_sync_path configured" in str(exc_info.value)
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
|
|
def test_project_check_success(mock_run, mock_is_installed):
|
|
"""Test successful project check."""
|
|
mock_is_installed.return_value = True
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
|
|
project = SyncProject(
|
|
name="research",
|
|
path="app/data/research",
|
|
local_sync_path="/tmp/research",
|
|
)
|
|
|
|
result = project_check(project, "my-bucket")
|
|
|
|
assert result is True
|
|
cmd = mock_run.call_args[0][0]
|
|
assert cmd[0] == "rclone"
|
|
assert cmd[1] == "check"
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
|
|
def test_project_check_with_one_way(mock_run, mock_is_installed):
|
|
"""Test project check with one-way flag."""
|
|
mock_is_installed.return_value = True
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
|
|
project = SyncProject(
|
|
name="research",
|
|
path="app/data/research",
|
|
local_sync_path="/tmp/research",
|
|
)
|
|
|
|
project_check(project, "my-bucket", one_way=True)
|
|
|
|
cmd = mock_run.call_args[0][0]
|
|
assert "--one-way" in cmd
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
def test_project_check_no_local_path(mock_is_installed):
|
|
"""Test project check raises error when local_sync_path not configured."""
|
|
mock_is_installed.return_value = True
|
|
project = SyncProject(name="research", path="app/data/research")
|
|
|
|
with pytest.raises(RcloneError) as exc_info:
|
|
project_check(project, "my-bucket")
|
|
|
|
assert "no local_sync_path configured" in str(exc_info.value)
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
|
|
def test_project_ls_success(mock_run, mock_is_installed):
|
|
"""Test successful project ls."""
|
|
mock_is_installed.return_value = True
|
|
mock_run.return_value = MagicMock(returncode=0, stdout="file1.md\nfile2.md\nsubdir/file3.md\n")
|
|
|
|
project = SyncProject(name="research", path="app/data/research")
|
|
|
|
files = project_ls(project, "my-bucket")
|
|
|
|
assert len(files) == 3
|
|
assert "file1.md" in files
|
|
assert "file2.md" in files
|
|
assert "subdir/file3.md" in files
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.subprocess.run")
|
|
def test_project_ls_with_subpath(mock_run, mock_is_installed):
|
|
"""Test project ls with subdirectory."""
|
|
mock_is_installed.return_value = True
|
|
mock_run.return_value = MagicMock(returncode=0, stdout="")
|
|
|
|
project = SyncProject(name="research", path="/research") # Normalized path
|
|
|
|
project_ls(project, "my-bucket", path="subdir")
|
|
|
|
cmd = mock_run.call_args[0][0]
|
|
assert cmd[-1] == "basic-memory-cloud:my-bucket/research/subdir"
|
|
|
|
|
|
# Tests for rclone installation check
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
def test_check_rclone_installed_success(mock_is_installed):
|
|
"""Test check_rclone_installed when rclone is installed."""
|
|
mock_is_installed.return_value = True
|
|
|
|
# Should not raise any error
|
|
check_rclone_installed()
|
|
|
|
mock_is_installed.assert_called_once()
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
def test_check_rclone_installed_not_found(mock_is_installed):
|
|
"""Test check_rclone_installed raises error when rclone not installed."""
|
|
mock_is_installed.return_value = False
|
|
|
|
with pytest.raises(RcloneError) as exc_info:
|
|
check_rclone_installed()
|
|
|
|
error_msg = str(exc_info.value)
|
|
assert "rclone is not installed" in error_msg
|
|
assert "bm cloud setup" in error_msg
|
|
assert "https://rclone.org/downloads/" in error_msg
|
|
mock_is_installed.assert_called_once()
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
def test_project_sync_checks_rclone_installed(mock_is_installed):
|
|
"""Test project_sync checks rclone is installed before running."""
|
|
mock_is_installed.return_value = False
|
|
|
|
project = SyncProject(
|
|
name="research",
|
|
path="app/data/research",
|
|
local_sync_path="/tmp/research",
|
|
)
|
|
|
|
with pytest.raises(RcloneError) as exc_info:
|
|
project_sync(project, "my-bucket")
|
|
|
|
assert "rclone is not installed" in str(exc_info.value)
|
|
mock_is_installed.assert_called_once()
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.bisync_initialized")
|
|
def test_project_bisync_checks_rclone_installed(mock_bisync_init, mock_is_installed):
|
|
"""Test project_bisync checks rclone is installed before running."""
|
|
mock_is_installed.return_value = False
|
|
mock_bisync_init.return_value = True
|
|
|
|
project = SyncProject(
|
|
name="research",
|
|
path="app/data/research",
|
|
local_sync_path="/tmp/research",
|
|
)
|
|
|
|
with pytest.raises(RcloneError) as exc_info:
|
|
project_bisync(project, "my-bucket")
|
|
|
|
assert "rclone is not installed" in str(exc_info.value)
|
|
mock_is_installed.assert_called_once()
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
def test_project_check_checks_rclone_installed(mock_is_installed):
|
|
"""Test project_check checks rclone is installed before running."""
|
|
mock_is_installed.return_value = False
|
|
|
|
project = SyncProject(
|
|
name="research",
|
|
path="app/data/research",
|
|
local_sync_path="/tmp/research",
|
|
)
|
|
|
|
with pytest.raises(RcloneError) as exc_info:
|
|
project_check(project, "my-bucket")
|
|
|
|
assert "rclone is not installed" in str(exc_info.value)
|
|
mock_is_installed.assert_called_once()
|
|
|
|
|
|
@patch("basic_memory.cli.commands.cloud.rclone_commands.is_rclone_installed")
|
|
def test_project_ls_checks_rclone_installed(mock_is_installed):
|
|
"""Test project_ls checks rclone is installed before running."""
|
|
mock_is_installed.return_value = False
|
|
|
|
project = SyncProject(name="research", path="app/data/research")
|
|
|
|
with pytest.raises(RcloneError) as exc_info:
|
|
project_ls(project, "my-bucket")
|
|
|
|
assert "rclone is not installed" in str(exc_info.value)
|
|
mock_is_installed.assert_called_once()
|