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

clean up cli commands

Browse Source 
Signed-off-by: phernandez <paul@basicmachines.co>
This commit is contained in:
phernandez
2026-02-21 16:06:43 -06:00
parent 9515130b2a
commit 2cde8d2659
30 changed files with 1974 additions and 1966 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/SPEC-PER-PROJECT-ROUTING.md
+2 -2
View File
@@ -45,14 +45,14 @@ When explicit routing is active, project mode does not override the selected rou
"main": {
"path": "/Users/me/basic-memory",
"mode": "local",
"cloud_sync_path": null,
"local_sync_path": null,
"bisync_initialized": false,
"last_sync": null
},
"specs": {
"path": "specs",
"mode": "cloud",
"cloud_sync_path": "/Users/me/dev/specs",
"local_sync_path": "/Users/me/dev/specs",
"bisync_initialized": true,
"last_sync": "2026-02-06T17:36:38.544153"
}
docs/cloud-cli.md
+1 -1
View File
@@ -120,7 +120,7 @@ bm project sync-setup research ~/Documents/research
When you add a project with `--local-path`:
1. Project created on cloud at `/app/data/research`
2. Local path stored in config for that project (`cloud_sync_path`)
2. Local path stored in config for that project (`local_sync_path`)
3. Local directory created if it doesn't exist
4. Bisync state directory created at `~/.basic-memory/bisync-state/research/`
src/basic_memory/cli/commands/__init__.py
-4
View File
@@ -8,8 +8,6 @@ from . import (
project,
format,
schema,
watch,
workspace,
)
__all__ = [
@@ -25,6 +23,4 @@ __all__ = [
"project",
"format",
"schema",
"watch",
"workspace",
]
src/basic_memory/cli/commands/cloud/__init__.py
+3
View File
@@ -6,11 +6,14 @@ from basic_memory.cli.app import cloud_app
from basic_memory.cli.commands.cloud.core_commands import * # noqa: F401,F403
from basic_memory.cli.commands.cloud.api_client import get_authenticated_headers, get_cloud_config # noqa: F401
from basic_memory.cli.commands.cloud.upload_command import * # noqa: F401,F403
from basic_memory.cli.commands.cloud.project_sync import * # noqa: F401,F403
# Register snapshot sub-command group
from basic_memory.cli.commands.cloud.snapshot import snapshot_app
from basic_memory.cli.commands.cloud.workspace import workspace_app
cloud_app.add_typer(snapshot_app, name="snapshot")
cloud_app.add_typer(workspace_app, name="workspace")
# Register restore command (directly on cloud_app via decorator)
from basic_memory.cli.commands.cloud.restore import restore # noqa: F401, E402
src/basic_memory/cli/commands/cloud/core_commands.py
+17 -13
View File
@@ -113,7 +113,7 @@ def status() -> None:
has_credentials = bool(config.cloud_api_key) or tokens is not None
if not has_credentials:
console.print(
"\n[dim]No cloud credentials found. Run: bm cloud login or bm cloud set-key <key>[/dim]"
"\n[dim]No cloud credentials found. Run: bm cloud login or bm cloud api-key save <key>[/dim]"
)
return
@@ -140,7 +140,7 @@ def status() -> None:
except CloudAPIError as e:
console.print(f"[yellow]Cloud health check failed: {e}[/yellow]")
console.print(
"[dim]Try re-authenticating with 'bm cloud login' or setting API key with 'bm cloud set-key'.[/dim]"
"[dim]Try re-authenticating with 'bm cloud login' or setting API key with 'bm cloud api-key save'.[/dim]"
)
except Exception as e:
console.print(f"[yellow]Unexpected health check error: {e}[/yellow]")
@@ -219,17 +219,22 @@ def promo(enabled: bool = typer.Option(True, "--on/--off", help="Enable or disab
console.print("[yellow]Cloud promo messages disabled[/yellow]")
@cloud_app.command("set-key")
def set_key(
# --- API key management subcommand group ---
api_key_app = typer.Typer(help="Manage cloud API keys")
cloud_app.add_typer(api_key_app, name="api-key")
@api_key_app.command("save")
def api_key_save(
api_key: str = typer.Argument(..., help="API key (bmc_ prefixed) for cloud access"),
) -> None:
"""Save a cloud API key for per-project cloud routing.
"""Save an existing API key to local config.
The API key is account-level and used by projects set to cloud mode.
Create a key in the web app or use 'bm cloud create-key'.
Use when you already have an API key (e.g., from the web app).
Example:
bm cloud set-key bmc_abc123...
bm cloud api-key save bmc_abc123...
"""
if not api_key.startswith("bmc_"):
console.print("[red]Error: API key must start with 'bmc_'[/red]")
@@ -245,17 +250,16 @@ def set_key(
console.print("[dim]Set a project to cloud mode: bm project set-cloud <name>[/dim]")
@cloud_app.command("create-key")
def create_key(
@api_key_app.command("create")
def api_key_create(
name: str = typer.Argument(..., help="Human-readable name for the API key"),
) -> None:
"""Create a new cloud API key and save it locally.
"""Create a new API key via the cloud API and save it locally.
Requires active OAuth session (run 'bm cloud login' first).
The key is created via the cloud API and saved to local config.
Example:
bm cloud create-key "my-laptop"
bm cloud api-key create "my-laptop"
"""
async def _create_key():
src/basic_memory/cli/commands/cloud/project_sync.py
+368
View File
@@ -0,0 +1,368 @@
"""Cloud sync commands for Basic Memory projects.
Commands for syncing, bisyncing, and checking integrity between local and cloud
project instances. These were previously in project.py but belong here since
they are cloud-specific operations.
"""
from datetime import datetime
import typer
from rich.console import Console
from basic_memory.cli.app import cloud_app
from basic_memory.cli.auth import CLIAuth
from basic_memory.cli.commands.cloud.bisync_commands import get_mount_info
from basic_memory.cli.commands.cloud.rclone_commands import (
RcloneError,
SyncProject,
get_project_bisync_state,
project_bisync,
project_check,
project_sync,
)
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.mcp.async_client import get_client
from basic_memory.mcp.tools.utils import call_get, call_post
from basic_memory.schemas.project_info import ProjectItem, ProjectList
from basic_memory.utils import generate_permalink, normalize_project_path
console = Console()
# --- Shared helpers ---
def _has_cloud_credentials(config) -> bool:
"""Return whether cloud credentials are available (API key or OAuth token)."""
if config.cloud_api_key:
return True
auth = CLIAuth(client_id=config.cloud_client_id, authkit_domain=config.cloud_domain)
return auth.load_tokens() is not None
def _require_cloud_credentials(config) -> None:
"""Exit with actionable guidance when cloud credentials are missing."""
if _has_cloud_credentials(config):
return
console.print("[red]Error: cloud credentials are required for this command[/red]")
console.print("[dim]Run 'bm cloud login' or 'bm cloud api-key save <key>' first[/dim]")
raise typer.Exit(1)
async def _get_cloud_project(name: str) -> ProjectItem | None:
"""Fetch a project by name from the cloud API."""
async with get_client() as client:
response = await call_get(client, "/v2/projects/")
projects_list = ProjectList.model_validate(response.json())
for proj in projects_list.projects:
if generate_permalink(proj.name) == generate_permalink(name):
return proj
return None
def _get_sync_project(
name: str, config, project_data: ProjectItem
) -> tuple[SyncProject, str | None]:
"""Build a SyncProject and resolve local_sync_path from config.
Returns (sync_project, local_sync_path). Exits if no local_sync_path configured.
"""
sync_entry = config.projects.get(name)
local_sync_path = sync_entry.local_sync_path if sync_entry else None
if not local_sync_path:
console.print(f"[red]Error: Project '{name}' has no local_sync_path configured[/red]")
console.print(f"\nConfigure sync with: bm cloud sync-setup {name} ~/path/to/local")
raise typer.Exit(1)
sync_project = SyncProject(
name=project_data.name,
path=normalize_project_path(project_data.path),
local_sync_path=local_sync_path,
)
return sync_project, local_sync_path
# --- Commands ---
@cloud_app.command("sync")
def sync_project_command(
name: str = typer.Option(..., "--name", help="Project name to sync"),
dry_run: bool = typer.Option(False, "--dry-run", help="Preview changes without syncing"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed output"),
) -> None:
"""One-way sync: local -> cloud (make cloud identical to local).
Example:
bm cloud sync --name research
bm cloud sync --name research --dry-run
"""
config = ConfigManager().config
_require_cloud_credentials(config)
try:
# Get tenant info for bucket name
tenant_info = run_with_cleanup(get_mount_info())
bucket_name = tenant_info.bucket_name
# Get project info
with force_routing(cloud=True):
project_data = run_with_cleanup(_get_cloud_project(name))
if not project_data:
console.print(f"[red]Error: Project '{name}' not found[/red]")
raise typer.Exit(1)
sync_project, local_sync_path = _get_sync_project(name, config, project_data)
# Run sync
console.print(f"[blue]Syncing {name} (local -> cloud)...[/blue]")
success = project_sync(sync_project, bucket_name, dry_run=dry_run, verbose=verbose)
if success:
console.print(f"[green]{name} synced successfully[/green]")
# Trigger database sync if not a dry run
if not dry_run:
async def _trigger_db_sync():
async with get_client() as client:
response = await call_post(
client,
f"/v2/projects/{project_data.external_id}/sync?force_full=true",
json={},
)
return response.json()
try:
with force_routing(cloud=True):
result = run_with_cleanup(_trigger_db_sync())
console.print(f"[dim]Database sync initiated: {result.get('message')}[/dim]")
except Exception as e:
console.print(f"[yellow]Warning: Could not trigger database sync: {e}[/yellow]")
else:
console.print(f"[red]{name} sync failed[/red]")
raise typer.Exit(1)
except RcloneError as e:
console.print(f"[red]Sync error: {e}[/red]")
raise typer.Exit(1)
except Exception as e:
console.print(f"[red]Error: {e}[/red]")
raise typer.Exit(1)
@cloud_app.command("bisync")
def bisync_project_command(
name: str = typer.Option(..., "--name", help="Project name to bisync"),
dry_run: bool = typer.Option(False, "--dry-run", help="Preview changes without syncing"),
resync: bool = typer.Option(False, "--resync", help="Force new baseline"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed output"),
) -> None:
"""Two-way sync: local <-> cloud (bidirectional sync).
Examples:
bm cloud bisync --name research --resync # First time
bm cloud bisync --name research # Subsequent syncs
bm cloud bisync --name research --dry-run # Preview changes
"""
config = ConfigManager().config
_require_cloud_credentials(config)
try:
# Get tenant info for bucket name
tenant_info = run_with_cleanup(get_mount_info())
bucket_name = tenant_info.bucket_name
# Get project info
with force_routing(cloud=True):
project_data = run_with_cleanup(_get_cloud_project(name))
if not project_data:
console.print(f"[red]Error: Project '{name}' not found[/red]")
raise typer.Exit(1)
sync_project, local_sync_path = _get_sync_project(name, config, project_data)
# Run bisync
console.print(f"[blue]Bisync {name} (local <-> cloud)...[/blue]")
success = project_bisync(
sync_project, bucket_name, dry_run=dry_run, resync=resync, verbose=verbose
)
if success:
console.print(f"[green]{name} bisync completed successfully[/green]")
# Update config — sync_entry is guaranteed non-None because
# _get_sync_project validated local_sync_path (which comes from sync_entry)
sync_entry = config.projects.get(name)
assert sync_entry is not None
sync_entry.last_sync = datetime.now()
sync_entry.bisync_initialized = True
ConfigManager().save_config(config)
# Trigger database sync if not a dry run
if not dry_run:
async def _trigger_db_sync():
async with get_client() as client:
response = await call_post(
client,
f"/v2/projects/{project_data.external_id}/sync?force_full=true",
json={},
)
return response.json()
try:
with force_routing(cloud=True):
result = run_with_cleanup(_trigger_db_sync())
console.print(f"[dim]Database sync initiated: {result.get('message')}[/dim]")
except Exception as e:
console.print(f"[yellow]Warning: Could not trigger database sync: {e}[/yellow]")
else:
console.print(f"[red]{name} bisync failed[/red]")
raise typer.Exit(1)
except RcloneError as e:
console.print(f"[red]Bisync error: {e}[/red]")
raise typer.Exit(1)
except Exception as e:
console.print(f"[red]Error: {e}[/red]")
raise typer.Exit(1)
@cloud_app.command("check")
def check_project_command(
name: str = typer.Option(..., "--name", help="Project name to check"),
one_way: bool = typer.Option(False, "--one-way", help="Check one direction only (faster)"),
) -> None:
"""Verify file integrity between local and cloud.
Example:
bm cloud check --name research
"""
config = ConfigManager().config
_require_cloud_credentials(config)
try:
# Get tenant info for bucket name
tenant_info = run_with_cleanup(get_mount_info())
bucket_name = tenant_info.bucket_name
# Get project info
with force_routing(cloud=True):
project_data = run_with_cleanup(_get_cloud_project(name))
if not project_data:
console.print(f"[red]Error: Project '{name}' not found[/red]")
raise typer.Exit(1)
sync_project, local_sync_path = _get_sync_project(name, config, project_data)
# Run check
console.print(f"[blue]Checking {name} integrity...[/blue]")
match = project_check(sync_project, bucket_name, one_way=one_way)
if match:
console.print(f"[green]{name} files match[/green]")
else:
console.print(f"[yellow]!{name} has differences[/yellow]")
except RcloneError as e:
console.print(f"[red]Check error: {e}[/red]")
raise typer.Exit(1)
except Exception as e:
console.print(f"[red]Error: {e}[/red]")
raise typer.Exit(1)
@cloud_app.command("bisync-reset")
def bisync_reset(
name: str = typer.Argument(..., help="Project name to reset bisync state for"),
) -> None:
"""Clear bisync state for a project.
This removes the bisync metadata files, forcing a fresh --resync on next bisync.
Useful when bisync gets into an inconsistent state or when remote path changes.
"""
import shutil
try:
state_path = get_project_bisync_state(name)
if not state_path.exists():
console.print(f"[yellow]No bisync state found for project '{name}'[/yellow]")
return
# Remove the entire state directory
shutil.rmtree(state_path)
console.print(f"[green]Cleared bisync state for project '{name}'[/green]")
console.print("\nNext steps:")
console.print(f" 1. Preview: bm cloud bisync --name {name} --resync --dry-run")
console.print(f" 2. Sync: bm cloud bisync --name {name} --resync")
except Exception as e:
console.print(f"[red]Error clearing bisync state: {str(e)}[/red]")
raise typer.Exit(1)
@cloud_app.command("sync-setup")
def setup_project_sync(
name: str = typer.Argument(..., help="Project name"),
local_path: str = typer.Argument(..., help="Local sync directory"),
) -> None:
"""Configure local sync for an existing cloud project.
Example:
bm cloud sync-setup research ~/Documents/research
"""
import os
from pathlib import Path
config_manager = ConfigManager()
config = config_manager.config
_require_cloud_credentials(config)
async def _verify_project_exists():
"""Verify the project exists on cloud by listing all projects."""
async with get_client() as client:
response = await call_get(client, "/v2/projects/")
project_list = response.json()
project_names = [p["name"] for p in project_list["projects"]]
if name not in project_names:
raise ValueError(f"Project '{name}' not found on cloud")
return True
try:
# Verify project exists on cloud
with force_routing(cloud=True):
run_with_cleanup(_verify_project_exists())
# Resolve and create local path
resolved_path = Path(os.path.abspath(os.path.expanduser(local_path)))
resolved_path.mkdir(parents=True, exist_ok=True)
# Update project entry with sync path
entry = config.projects.get(name)
if entry:
entry.local_sync_path = resolved_path.as_posix()
entry.bisync_initialized = False
entry.last_sync = None
else:
config.projects[name] = ProjectEntry(
path=resolved_path.as_posix(),
local_sync_path=resolved_path.as_posix(),
)
config_manager.save_config(config)
console.print(f"[green]Sync configured for project '{name}'[/green]")
console.print(f"\nLocal sync path: {resolved_path}")
console.print("\nNext steps:")
console.print(f" 1. Preview: bm cloud bisync --name {name} --resync --dry-run")
console.print(f" 2. Sync: bm cloud bisync --name {name} --resync")
except Exception as e:
console.print(f"[red]Error configuring sync: {str(e)}[/red]")
raise typer.Exit(1)
src/basic_memory/cli/commands/workspace.py → src/basic_memory/cli/commands/cloud/workspace.py
+1 -8
View File
@@ -4,14 +4,13 @@ import typer
from rich.console import Console
from rich.table import Table
from basic_memory.cli.app import app
from basic_memory.cli.app import cloud_app
from basic_memory.cli.commands.command_utils import run_with_cleanup
from basic_memory.mcp.project_context import get_available_workspaces
console = Console()
workspace_app = typer.Typer(help="Manage cloud workspaces")
app.add_typer(workspace_app, name="workspace")
@workspace_app.command("list")
@@ -49,9 +48,3 @@ def list_workspaces() -> None:
)
console.print(table)
@app.command("workspaces")
def workspaces_alias() -> None:
"""Alias for `bm workspace list`."""
list_workspaces()
src/basic_memory/cli/commands/project.py
+44 -396
View File
@@ -12,6 +12,15 @@ from rich.table import Table
from basic_memory.cli.app import app
from basic_memory.cli.auth import CLIAuth
from basic_memory.cli.commands.cloud.bisync_commands import get_mount_info
from basic_memory.cli.commands.cloud.project_sync import (
_has_cloud_credentials,
_require_cloud_credentials,
)
from basic_memory.cli.commands.cloud.rclone_commands import (
SyncProject,
project_ls,
)
from basic_memory.cli.commands.command_utils import get_project_info, run_with_cleanup
from basic_memory.cli.commands.routing import force_routing, validate_routing_flags
from basic_memory.config import ConfigManager, ProjectEntry, ProjectMode
@@ -21,17 +30,6 @@ from basic_memory.schemas.project_info import ProjectItem, ProjectList, ProjectS
from basic_memory.schemas.v2 import ProjectResolveResponse
from basic_memory.utils import generate_permalink, normalize_project_path
# Import rclone commands for project sync
from basic_memory.cli.commands.cloud.rclone_commands import (
RcloneError,
SyncProject,
project_bisync,
project_check,
project_ls,
project_sync,
)
from basic_memory.cli.commands.cloud.bisync_commands import get_mount_info
console = Console()
# Create a project subcommand
@@ -47,25 +45,6 @@ def format_path(path: str) -> str:
return path
def _has_cloud_credentials(config) -> bool:
"""Return whether cloud credentials are available (API key or OAuth token)."""
if config.cloud_api_key:
return True
auth = CLIAuth(client_id=config.cloud_client_id, authkit_domain=config.cloud_domain)
return auth.load_tokens() is not None
def _require_cloud_credentials(config) -> None:
"""Exit with actionable guidance when cloud credentials are missing."""
if _has_cloud_credentials(config):
return
console.print("[red]Error: cloud credentials are required for this command[/red]")
console.print("[dim]Run 'bm cloud login' or 'bm cloud set-key <key>' first[/dim]")
raise typer.Exit(1)
@project_app.command("list")
def list_projects(
local: bool = typer.Option(False, "--local", help="Force local routing for this command"),
@@ -141,8 +120,8 @@ def list_projects(
local_path = ""
if local_project is not None:
local_path = format_path(normalize_project_path(local_project.path))
elif entry and entry.cloud_sync_path:
local_path = format_path(entry.cloud_sync_path)
elif entry and entry.local_sync_path:
local_path = format_path(entry.local_sync_path)
elif entry and entry.mode == ProjectMode.LOCAL and entry.path:
local_path = format_path(normalize_project_path(entry.path))
@@ -169,7 +148,7 @@ def list_projects(
if cloud_project is not None and cloud_project.is_default:
is_default = "[X]"
has_sync = "[X]" if entry and entry.cloud_sync_path else ""
has_sync = "[X]" if entry and entry.local_sync_path else ""
mcp_stdio_target = "local" if local_project is not None else "n/a"
row = [
@@ -188,7 +167,7 @@ def list_projects(
if cloud_error is not None:
console.print(
"[yellow]Cloud project discovery failed. "
"Showing local projects only. Run 'bm cloud login' or 'bm cloud set-key <key>'.[/yellow]"
"Showing local projects only. Run 'bm cloud login' or 'bm cloud api-key save <key>'.[/yellow]"
)
except Exception as e:
console.print(f"[red]Error listing projects: {str(e)}[/red]")
@@ -281,80 +260,24 @@ def add_project(
# Update project entry with sync path
entry = config.projects.get(name)
if entry:
entry.cloud_sync_path = local_sync_path
entry.local_sync_path = local_sync_path
else:
# Project may not be in local config yet (cloud-only add)
config.projects[name] = ProjectEntry(
path=local_sync_path,
cloud_sync_path=local_sync_path,
local_sync_path=local_sync_path,
)
ConfigManager().save_config(config)
console.print(f"\n[green]Local sync path configured: {local_sync_path}[/green]")
console.print("\nNext steps:")
console.print(f" 1. Preview: bm project bisync --name {name} --resync --dry-run")
console.print(f" 2. Sync: bm project bisync --name {name} --resync")
console.print(f" 1. Preview: bm cloud bisync --name {name} --resync --dry-run")
console.print(f" 2. Sync: bm cloud bisync --name {name} --resync")
except Exception as e:
console.print(f"[red]Error adding project: {str(e)}[/red]")
raise typer.Exit(1)
@project_app.command("sync-setup")
def setup_project_sync(
name: str = typer.Argument(..., help="Project name"),
local_path: str = typer.Argument(..., help="Local sync directory"),
) -> None:
"""Configure local sync for an existing cloud project.
Example:
bm project sync-setup research ~/Documents/research
"""
config_manager = ConfigManager()
config = config_manager.config
_require_cloud_credentials(config)
async def _verify_project_exists():
"""Verify the project exists on cloud by listing all projects."""
async with get_client() as client:
response = await call_get(client, "/v2/projects/")
project_list = response.json()
project_names = [p["name"] for p in project_list["projects"]]
if name not in project_names:
raise ValueError(f"Project '{name}' not found on cloud")
return True
try:
# Verify project exists on cloud
with force_routing(cloud=True):
run_with_cleanup(_verify_project_exists())
# Resolve and create local path
resolved_path = Path(os.path.abspath(os.path.expanduser(local_path)))
resolved_path.mkdir(parents=True, exist_ok=True)
# Update project entry with sync path
entry = config.projects.get(name)
if entry:
entry.cloud_sync_path = resolved_path.as_posix()
entry.bisync_initialized = False
entry.last_sync = None
else:
config.projects[name] = ProjectEntry(
path=resolved_path.as_posix(),
cloud_sync_path=resolved_path.as_posix(),
)
config_manager.save_config(config)
console.print(f"[green]Sync configured for project '{name}'[/green]")
console.print(f"\nLocal sync path: {resolved_path}")
console.print("\nNext steps:")
console.print(f" 1. Preview: bm project bisync --name {name} --resync --dry-run")
console.print(f" 2. Sync: bm project bisync --name {name} --resync")
except Exception as e:
console.print(f"[red]Error configuring sync: {str(e)}[/red]")
raise typer.Exit(1)
@project_app.command("remove")
def remove_project(
name: str = typer.Argument(..., help="Name of the project to remove"),
@@ -400,8 +323,8 @@ def remove_project(
has_bisync_state = False
entry = config.projects.get(name)
if cloud and entry and entry.cloud_sync_path:
local_path_config = entry.cloud_sync_path
if cloud and entry and entry.local_sync_path:
local_path_config = entry.local_sync_path
# Check for bisync state
from basic_memory.cli.commands.cloud.rclone_commands import get_project_bisync_state
@@ -434,8 +357,8 @@ def remove_project(
console.print("[green]Removed bisync state[/green]")
# Clean up cloud sync fields on the project entry
if cloud and entry and entry.cloud_sync_path:
entry.cloud_sync_path = None
if cloud and entry and entry.local_sync_path:
entry.local_sync_path = None
entry.bisync_initialized = False
entry.last_sync = None
ConfigManager().save_config(config)
@@ -516,13 +439,11 @@ def synchronize_projects(
def move_project(
name: str = typer.Argument(..., help="Name of the project to move"),
new_path: str = typer.Argument(..., help="New absolute path for the project"),
local: bool = typer.Option(
False, "--local", help="Force local API routing (required in cloud mode)"
),
) -> None:
"""Move a project to a new location.
"""Move a local project to a new filesystem location.
In cloud mode, use --local to modify local project paths.
This command only applies to local projects — it updates the project's
configured path in the local database.
"""
# Resolve to absolute path
resolved_path = Path(os.path.abspath(os.path.expanduser(new_path))).as_posix()
@@ -542,7 +463,7 @@ def move_project(
return ProjectStatusResponse.model_validate(response.json())
try:
with force_routing(local=local):
with force_routing(local=True):
result = run_with_cleanup(_move_project())
console.print(f"[green]{result.message}[/green]")
@@ -574,13 +495,12 @@ def set_cloud(
Requires either an API key or an active OAuth session.
Examples:
bm cloud set-key bmc_abc123... # save API key, then:
bm cloud api-key save bmc_abc123... # save API key, then:
bm project set-cloud research # route "research" through cloud
bm cloud login # OAuth login, then:
bm project set-cloud research # route "research" through cloud
"""
from basic_memory.cli.auth import CLIAuth
config_manager = ConfigManager()
config = config_manager.config
@@ -599,7 +519,7 @@ def set_cloud(
if not has_api_key and not has_oauth:
console.print("[red]Error: No cloud credentials found[/red]")
console.print("[dim]Run 'bm cloud set-key <key>' or 'bm cloud login' first[/dim]")
console.print("[dim]Run 'bm cloud api-key save <key>' or 'bm cloud login' first[/dim]")
raise typer.Exit(1)
config.set_project_mode(name, ProjectMode.CLOUD)
@@ -633,293 +553,6 @@ def set_local(
console.print("[dim]MCP tools and CLI commands for this project will use local transport[/dim]")
@project_app.command("sync")
def sync_project_command(
name: str = typer.Option(..., "--name", help="Project name to sync"),
dry_run: bool = typer.Option(False, "--dry-run", help="Preview changes without syncing"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed output"),
) -> None:
"""One-way sync: local -> cloud (make cloud identical to local).
Example:
bm project sync --name research
bm project sync --name research --dry-run
"""
config = ConfigManager().config
_require_cloud_credentials(config)
try:
# Get tenant info for bucket name
tenant_info = run_with_cleanup(get_mount_info())
bucket_name = tenant_info.bucket_name
# Get project info
async def _get_project():
async with get_client() as client:
response = await call_get(client, "/v2/projects/")
projects_list = ProjectList.model_validate(response.json())
for proj in projects_list.projects:
if generate_permalink(proj.name) == generate_permalink(name):
return proj
return None
with force_routing(cloud=True):
project_data = run_with_cleanup(_get_project())
if not project_data:
console.print(f"[red]Error: Project '{name}' not found[/red]")
raise typer.Exit(1)
# Get local_sync_path from project entry
sync_entry = config.projects.get(name)
local_sync_path = sync_entry.cloud_sync_path if sync_entry else None
if not local_sync_path:
console.print(f"[red]Error: Project '{name}' has no local_sync_path configured[/red]")
console.print(f"\nConfigure sync with: bm project sync-setup {name} ~/path/to/local")
raise typer.Exit(1)
# Create SyncProject
sync_project = SyncProject(
name=project_data.name,
path=normalize_project_path(project_data.path),
local_sync_path=local_sync_path,
)
# Run sync
console.print(f"[blue]Syncing {name} (local -> cloud)...[/blue]")
success = project_sync(sync_project, bucket_name, dry_run=dry_run, verbose=verbose)
if success:
console.print(f"[green]{name} synced successfully[/green]")
# Trigger database sync if not a dry run
if not dry_run:
async def _trigger_db_sync():
async with get_client() as client:
response = await call_post(
client,
f"/v2/projects/{project_data.external_id}/sync?force_full=true",
json={},
)
return response.json()
try:
with force_routing(cloud=True):
result = run_with_cleanup(_trigger_db_sync())
console.print(f"[dim]Database sync initiated: {result.get('message')}[/dim]")
except Exception as e:
console.print(f"[yellow]Warning: Could not trigger database sync: {e}[/yellow]")
else:
console.print(f"[red]{name} sync failed[/red]")
raise typer.Exit(1)
except RcloneError as e:
console.print(f"[red]Sync error: {e}[/red]")
raise typer.Exit(1)
except Exception as e:
console.print(f"[red]Error: {e}[/red]")
raise typer.Exit(1)
@project_app.command("bisync")
def bisync_project_command(
name: str = typer.Option(..., "--name", help="Project name to bisync"),
dry_run: bool = typer.Option(False, "--dry-run", help="Preview changes without syncing"),
resync: bool = typer.Option(False, "--resync", help="Force new baseline"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed output"),
) -> None:
"""Two-way sync: local <-> cloud (bidirectional sync).
Examples:
bm project bisync --name research --resync # First time
bm project bisync --name research # Subsequent syncs
bm project bisync --name research --dry-run # Preview changes
"""
config = ConfigManager().config
_require_cloud_credentials(config)
try:
# Get tenant info for bucket name
tenant_info = run_with_cleanup(get_mount_info())
bucket_name = tenant_info.bucket_name
# Get project info
async def _get_project():
async with get_client() as client:
response = await call_get(client, "/v2/projects/")
projects_list = ProjectList.model_validate(response.json())
for proj in projects_list.projects:
if generate_permalink(proj.name) == generate_permalink(name):
return proj
return None
with force_routing(cloud=True):
project_data = run_with_cleanup(_get_project())
if not project_data:
console.print(f"[red]Error: Project '{name}' not found[/red]")
raise typer.Exit(1)
# Get local_sync_path from project entry
sync_entry = config.projects.get(name)
local_sync_path = sync_entry.cloud_sync_path if sync_entry else None
if not local_sync_path:
console.print(f"[red]Error: Project '{name}' has no local_sync_path configured[/red]")
console.print(f"\nConfigure sync with: bm project sync-setup {name} ~/path/to/local")
raise typer.Exit(1)
# Create SyncProject
sync_project = SyncProject(
name=project_data.name,
path=normalize_project_path(project_data.path),
local_sync_path=local_sync_path,
)
# Run bisync
console.print(f"[blue]Bisync {name} (local <-> cloud)...[/blue]")
success = project_bisync(
sync_project, bucket_name, dry_run=dry_run, resync=resync, verbose=verbose
)
if success:
console.print(f"[green]{name} bisync completed successfully[/green]")
# Update config — sync_entry is guaranteed non-None because
# we checked local_sync_path above (which comes from sync_entry)
assert sync_entry is not None
sync_entry.last_sync = datetime.now()
sync_entry.bisync_initialized = True
ConfigManager().save_config(config)
# Trigger database sync if not a dry run
if not dry_run:
async def _trigger_db_sync():
async with get_client() as client:
response = await call_post(
client,
f"/v2/projects/{project_data.external_id}/sync?force_full=true",
json={},
)
return response.json()
try:
with force_routing(cloud=True):
result = run_with_cleanup(_trigger_db_sync())
console.print(f"[dim]Database sync initiated: {result.get('message')}[/dim]")
except Exception as e:
console.print(f"[yellow]Warning: Could not trigger database sync: {e}[/yellow]")
else:
console.print(f"[red]{name} bisync failed[/red]")
raise typer.Exit(1)
except RcloneError as e:
console.print(f"[red]Bisync error: {e}[/red]")
raise typer.Exit(1)
except Exception as e:
console.print(f"[red]Error: {e}[/red]")
raise typer.Exit(1)
@project_app.command("check")
def check_project_command(
name: str = typer.Option(..., "--name", help="Project name to check"),
one_way: bool = typer.Option(False, "--one-way", help="Check one direction only (faster)"),
) -> None:
"""Verify file integrity between local and cloud.
Example:
bm project check --name research
"""
config = ConfigManager().config
_require_cloud_credentials(config)
try:
# Get tenant info for bucket name
tenant_info = run_with_cleanup(get_mount_info())
bucket_name = tenant_info.bucket_name
# Get project info
async def _get_project():
async with get_client() as client:
response = await call_get(client, "/v2/projects/")
projects_list = ProjectList.model_validate(response.json())
for proj in projects_list.projects:
if generate_permalink(proj.name) == generate_permalink(name):
return proj
return None
with force_routing(cloud=True):
project_data = run_with_cleanup(_get_project())
if not project_data:
console.print(f"[red]Error: Project '{name}' not found[/red]")
raise typer.Exit(1)
# Get local_sync_path from project entry
check_entry = config.projects.get(name)
local_sync_path = check_entry.cloud_sync_path if check_entry else None
if not local_sync_path:
console.print(f"[red]Error: Project '{name}' has no local_sync_path configured[/red]")
console.print(f"\nConfigure sync with: bm project sync-setup {name} ~/path/to/local")
raise typer.Exit(1)
# Create SyncProject
sync_project = SyncProject(
name=project_data.name,
path=normalize_project_path(project_data.path),
local_sync_path=local_sync_path,
)
# Run check
console.print(f"[blue]Checking {name} integrity...[/blue]")
match = project_check(sync_project, bucket_name, one_way=one_way)
if match:
console.print(f"[green]{name} files match[/green]")
else:
console.print(f"[yellow]!{name} has differences[/yellow]")
except RcloneError as e:
console.print(f"[red]Check error: {e}[/red]")
raise typer.Exit(1)
except Exception as e:
console.print(f"[red]Error: {e}[/red]")
raise typer.Exit(1)
@project_app.command("bisync-reset")
def bisync_reset(
name: str = typer.Argument(..., help="Project name to reset bisync state for"),
) -> None:
"""Clear bisync state for a project.
This removes the bisync metadata files, forcing a fresh --resync on next bisync.
Useful when bisync gets into an inconsistent state or when remote path changes.
"""
from basic_memory.cli.commands.cloud.rclone_commands import get_project_bisync_state
import shutil
try:
state_path = get_project_bisync_state(name)
if not state_path.exists():
console.print(f"[yellow]No bisync state found for project '{name}'[/yellow]")
return
# Remove the entire state directory
shutil.rmtree(state_path)
console.print(f"[green]Cleared bisync state for project '{name}'[/green]")
console.print("\nNext steps:")
console.print(f" 1. Preview: bm project bisync --name {name} --resync --dry-run")
console.print(f" 2. Sync: bm project bisync --name {name} --resync")
except Exception as e:
console.print(f"[red]Error clearing bisync state: {str(e)}[/red]")
raise typer.Exit(1)
@project_app.command("ls")
def ls_project_command(
name: str = typer.Option(..., "--name", help="Project name to list files from"),
@@ -941,7 +574,13 @@ def ls_project_command(
console.print(f"[red]Error: {e}[/red]")
raise typer.Exit(1)
use_cloud_route = cloud and not local
# Determine routing: explicit flags take precedence, otherwise check project mode
if cloud or local:
use_cloud_route = cloud and not local
else:
config = ConfigManager().config
project_mode = config.get_project_mode(name)
use_cloud_route = project_mode == ProjectMode.CLOUD
def _list_local_files(project_path: str, subpath: str | None = None) -> list[str]:
project_root = Path(normalize_project_path(project_path)).expanduser().resolve()
@@ -1005,7 +644,16 @@ def ls_project_command(
if not project_data:
console.print(f"[red]Error: Project '{name}' not found[/red]")
raise typer.Exit(1)
files = _list_local_files(project_data.path, path)
# For cloud-mode projects accessed with --local, use local_sync_path
# (the actual local directory) instead of project_data.path from the API
local_dir = project_data.path
if local:
entry = ConfigManager().config.projects.get(name)
if entry and entry.local_sync_path:
local_dir = entry.local_sync_path
files = _list_local_files(local_dir, path)
target_label = "LOCAL"
if files:
src/basic_memory/cli/commands/schema.py
+188 -188
View File
@@ -2,6 +2,10 @@
Provides CLI access to schema validation, inference, and drift detection.
Registered as a subcommand group: `bm schema validate`, `bm schema infer`, `bm schema diff`.
Each command calls the corresponding MCP tool with output_format="json" and
renders the result as Rich tables — same code path as `bm tool schema-*` but
with human-friendly formatting.
"""
import json
@@ -16,8 +20,9 @@ from basic_memory.cli.app import app
from basic_memory.cli.commands.command_utils import run_with_cleanup
from basic_memory.cli.commands.routing import force_routing, validate_routing_flags
from basic_memory.config import ConfigManager
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.project_context import get_active_project
from basic_memory.mcp.tools import schema_diff as mcp_schema_diff
from basic_memory.mcp.tools import schema_infer as mcp_schema_infer
from basic_memory.mcp.tools import schema_validate as mcp_schema_validate
console = Console()
@@ -37,77 +42,124 @@ def _resolve_project_name(project: Optional[str]) -> Optional[str]:
return config_manager.default_project
# --- Validate ---
# --- Rendering helpers ---
async def _run_validate(
target: Optional[str] = None,
project: Optional[str] = None,
strict: bool = False,
):
"""Run schema validation via the API."""
from basic_memory.mcp.clients.schema import SchemaClient
def _render_validate_table(data: dict) -> None:
"""Render a validation report dict as a Rich table."""
entity_type = data.get("entity_type")
title_label = entity_type or "all"
async with get_client(project_name=project) as client:
active_project = await get_active_project(client, project, None)
schema_client = SchemaClient(client, active_project.external_id)
table = Table(title=f"Schema Validation: {title_label}")
table.add_column("Note", style="cyan")
table.add_column("Status", justify="center")
table.add_column("Warnings", justify="right")
table.add_column("Errors", justify="right")
# Determine if target is a note identifier or note type
# Heuristic: if target contains / or ., treat as identifier
entity_type = None
identifier = None
if target:
if "/" in target or "." in target:
identifier = target
else:
entity_type = target
for result in data.get("results", []):
warnings = result.get("warnings", [])
errors = result.get("errors", [])
passed = result.get("passed", True)
report = await schema_client.validate(
entity_type=entity_type,
identifier=identifier,
if passed and not warnings:
status = "[green]pass[/green]"
elif passed:
status = "[yellow]warn[/yellow]"
else:
status = "[red]fail[/red]"
table.add_row(
result.get("note_identifier", ""),
status,
str(len(warnings)),
str(len(errors)),
)
# --- Display results ---
if report.total_notes == 0:
if report.total_entities == 0:
console.print(f"[yellow]No notes of type '{entity_type}' found.[/yellow]")
else:
console.print(
f"[yellow]Found {report.total_entities} notes but no schema "
f"defined for '{entity_type}'.[/yellow]"
)
return
console.print(table)
console.print(
f"\nSummary: {data.get('valid_count', 0)}/{data.get('total_notes', 0)} valid, "
f"{data.get('warning_count', 0)} warnings, {data.get('error_count', 0)} errors"
)
table = Table(title=f"Schema Validation: {entity_type or identifier or 'all'}")
table.add_column("Note", style="cyan")
table.add_column("Status", justify="center")
table.add_column("Warnings", justify="right")
table.add_column("Errors", justify="right")
for result in report.results:
if result.passed and not result.warnings:
status = "[green]pass[/green]"
elif result.passed:
status = "[yellow]warn[/yellow]"
else:
status = "[red]fail[/red]"
def _render_infer_table(data: dict) -> None:
"""Render an inference report dict as a Rich table."""
entity_type = data.get("entity_type", "")
notes_analyzed = data.get("notes_analyzed", 0)
suggested_required = data.get("suggested_required", [])
suggested_optional = data.get("suggested_optional", [])
table.add_row(
result.note_identifier,
status,
str(len(result.warnings)),
str(len(result.errors)),
console.print(f"\n[bold]Analyzing {notes_analyzed} notes with type: {entity_type}...[/bold]\n")
table = Table(title="Field Frequencies")
table.add_column("Field", style="cyan")
table.add_column("Source")
table.add_column("Count", justify="right")
table.add_column("Percentage", justify="right")
table.add_column("Suggested")
for freq in data.get("field_frequencies", []):
pct = f"{freq.get('percentage', 0):.0%}"
name = freq.get("name", "")
if name in suggested_required:
suggested = "[green]required[/green]"
elif name in suggested_optional:
suggested = "[yellow]optional[/yellow]"
else:
suggested = "[dim]excluded[/dim]"
table.add_row(
name,
freq.get("source", ""),
str(freq.get("count", 0)),
pct,
suggested,
)
console.print(table)
suggested_schema = data.get("suggested_schema", {})
if suggested_schema:
console.print("\n[bold]Suggested schema:[/bold]")
console.print(json.dumps(suggested_schema, indent=2))
def _render_diff_output(data: dict) -> None:
"""Render a drift report dict as Rich output."""
entity_type = data.get("entity_type", "")
new_fields = data.get("new_fields", [])
dropped_fields = data.get("dropped_fields", [])
cardinality_changes = data.get("cardinality_changes", [])
has_drift = new_fields or dropped_fields or cardinality_changes
if not has_drift:
console.print(f"[green]No drift detected for {entity_type} schema.[/green]")
return
console.print(f"\n[bold]Schema drift detected for {entity_type}:[/bold]\n")
if new_fields:
console.print("[green]+ New fields (common in notes, not in schema):[/green]")
for f in new_fields:
console.print(
f" + {f['name']}: {f.get('percentage', 0):.0%} of notes ({f.get('source', '')})"
)
console.print(table)
console.print(
f"\nSummary: {report.valid_count}/{report.total_notes} valid, "
f"{report.warning_count} warnings, {report.error_count} errors"
)
if dropped_fields:
console.print("[red]- Dropped fields (in schema, rare in notes):[/red]")
for f in dropped_fields:
console.print(
f" - {f['name']}: {f.get('percentage', 0):.0%} of notes ({f.get('source', '')})"
)
# Exit with error code in strict mode if there are failures
if strict and report.error_count > 0:
raise typer.Exit(1)
if cardinality_changes:
console.print("[yellow]~ Cardinality changes:[/yellow]")
for change in cardinality_changes:
console.print(f" ~ {change}")
# --- Commands ---
@schema_app.command()
@@ -138,8 +190,36 @@ def validate(
try:
validate_routing_flags(local, cloud)
project_name = _resolve_project_name(project)
# Heuristic: if target contains / or ., treat as identifier; otherwise as note type
note_type, identifier = None, None
if target:
if "/" in target or "." in target:
identifier = target
else:
note_type = target
with force_routing(local=local, cloud=cloud):
run_with_cleanup(_run_validate(target, project_name, strict))
result = run_with_cleanup(
mcp_schema_validate(
note_type=note_type,
identifier=identifier,
project=project_name,
output_format="json",
)
)
# Handle error responses
if isinstance(result, dict) and "error" in result:
console.print(f"[yellow]{result['error']}[/yellow]")
return
# output_format="json" guarantees a dict return
assert isinstance(result, dict)
_render_validate_table(result)
if strict and result.get("error_count", 0) > 0:
raise typer.Exit(1)
except ValueError as e:
console.print(f"[red]Error: {e}[/red]")
raise typer.Exit(1)
@@ -151,91 +231,6 @@ def validate(
raise
# --- Infer ---
async def _run_infer(
entity_type: str,
project: Optional[str] = None,
threshold: float = 0.25,
save: bool = False,
):
"""Run schema inference via the API."""
from basic_memory.mcp.clients.schema import SchemaClient
async with get_client(project_name=project) as client:
active_project = await get_active_project(client, project, None)
schema_client = SchemaClient(client, active_project.external_id)
report = await schema_client.infer(entity_type, threshold=threshold)
if report.notes_analyzed == 0:
console.print(f"[yellow]No notes found with type: {entity_type}[/yellow]")
return
# --- Empty schema guard ---
# Trigger: notes were analyzed but no fields met the threshold
# Why: dumping hundreds of excluded fields is not useful output
# Outcome: show count and suggest a more specific type
if not report.suggested_schema:
console.print(
f"\n[yellow]Analyzed {report.notes_analyzed} notes of type '{entity_type}', "
f"but no fields met the {threshold:.0%} threshold.[/yellow]\n"
)
console.print(
f"This usually means '{entity_type}' is too broad — "
f"the notes don't share a consistent structure.\n"
)
console.print("[bold]Suggestions:[/bold]")
console.print(" 1. Use a more specific type")
console.print(
f" 2. Lower the threshold: bm schema infer {entity_type} --threshold 0.1"
)
console.print(" 3. Create typed notes with write_note using a specific note_type")
return
# --- Display frequency analysis ---
console.print(
f"\n[bold]Analyzing {report.notes_analyzed} notes with type: {entity_type}...[/bold]\n"
)
table = Table(title="Field Frequencies")
table.add_column("Field", style="cyan")
table.add_column("Source")
table.add_column("Count", justify="right")
table.add_column("Percentage", justify="right")
table.add_column("Suggested")
for freq in report.field_frequencies:
pct = f"{freq.percentage:.0%}"
if freq.name in report.suggested_required:
suggested = "[green]required[/green]"
elif freq.name in report.suggested_optional:
suggested = "[yellow]optional[/yellow]"
else:
suggested = "[dim]excluded[/dim]"
table.add_row(
freq.name,
freq.source,
str(freq.count),
pct,
suggested,
)
console.print(table)
# --- Display suggested schema ---
console.print("\n[bold]Suggested schema:[/bold]")
console.print(json.dumps(report.suggested_schema, indent=2))
if save:
console.print(
f"\n[yellow]--save not yet implemented. "
f"Copy the schema above into schema/{entity_type}.md[/yellow]"
)
@schema_app.command()
def infer(
entity_type: Annotated[
@@ -269,8 +264,37 @@ def infer(
try:
validate_routing_flags(local, cloud)
project_name = _resolve_project_name(project)
with force_routing(local=local, cloud=cloud):
run_with_cleanup(_run_infer(entity_type, project_name, threshold, save))
result = run_with_cleanup(
mcp_schema_infer(
note_type=entity_type,
threshold=threshold,
project=project_name,
output_format="json",
)
)
# Handle error responses
if isinstance(result, dict) and "error" in result:
console.print(f"[yellow]{result['error']}[/yellow]")
return
# output_format="json" guarantees a dict return
assert isinstance(result, dict)
# Handle zero notes
if result.get("notes_analyzed", 0) == 0:
console.print(f"[yellow]No notes found with type: {entity_type}[/yellow]")
return
_render_infer_table(result)
if save:
console.print(
f"\n[yellow]--save not yet implemented. "
f"Copy the schema above into schema/{entity_type}.md[/yellow]"
)
except ValueError as e:
console.print(f"[red]Error: {e}[/red]")
raise typer.Exit(1)
@@ -282,46 +306,6 @@ def infer(
raise
# --- Diff ---
async def _run_diff(
entity_type: str,
project: Optional[str] = None,
):
"""Run schema drift detection via the API."""
from basic_memory.mcp.clients.schema import SchemaClient
async with get_client(project_name=project) as client:
active_project = await get_active_project(client, project, None)
schema_client = SchemaClient(client, active_project.external_id)
report = await schema_client.diff(entity_type)
has_drift = report.new_fields or report.dropped_fields or report.cardinality_changes
if not has_drift:
console.print(f"[green]No drift detected for {entity_type} schema.[/green]")
return
console.print(f"\n[bold]Schema drift detected for {entity_type}:[/bold]\n")
if report.new_fields:
console.print("[green]+ New fields (common in notes, not in schema):[/green]")
for f in report.new_fields:
console.print(f" + {f.name}: {f.percentage:.0%} of notes ({f.source})")
if report.dropped_fields:
console.print("[red]- Dropped fields (in schema, rare in notes):[/red]")
for f in report.dropped_fields:
console.print(f" - {f.name}: {f.percentage:.0%} of notes ({f.source})")
if report.cardinality_changes:
console.print("[yellow]~ Cardinality changes:[/yellow]")
for change in report.cardinality_changes:
console.print(f" ~ {change}")
@schema_app.command()
def diff(
entity_type: Annotated[
@@ -349,8 +333,24 @@ def diff(
try:
validate_routing_flags(local, cloud)
project_name = _resolve_project_name(project)
with force_routing(local=local, cloud=cloud):
run_with_cleanup(_run_diff(entity_type, project_name))
result = run_with_cleanup(
mcp_schema_diff(
note_type=entity_type,
project=project_name,
output_format="json",
)
)
# Handle error responses
if isinstance(result, dict) and "error" in result:
console.print(f"[yellow]{result['error']}[/yellow]")
return
# output_format="json" guarantees a dict return
assert isinstance(result, dict)
_render_diff_output(result)
except ValueError as e:
console.print(f"[red]Error: {e}[/red]")
raise typer.Exit(1)
src/basic_memory/cli/commands/tool.py
+332 -538
View File
File diff suppressed because it is too large Load Diff
src/basic_memory/cli/commands/watch.py
-97
View File
@@ -1,97 +0,0 @@
"""Watch command - run file watcher as a standalone long-running process."""
import asyncio
import os
import signal
import sys
from typing import Optional
import typer
from loguru import logger
from basic_memory import db
from basic_memory.cli.app import app
from basic_memory.cli.container import get_container
from basic_memory.config import ConfigManager
from basic_memory.services.initialization import initialize_app
from basic_memory.sync.coordinator import SyncCoordinator
async def run_watch(project: Optional[str] = None) -> None:
"""Run the file watcher as a long-running process.
This is the async core of the watch command. It:
1. Initializes the app (DB migrations + project reconciliation)
2. Validates and sets project constraint if --project given
3. Creates a SyncCoordinator with quiet=False for Rich console output
4. Blocks until SIGINT/SIGTERM, then shuts down cleanly
"""
container = get_container()
config = container.config
# --- Initialization ---
# Wrapped in try/finally so DB resources are cleaned up on all exit paths,
# including early exits from invalid --project names.
await initialize_app(config)
sync_coordinator = None
try:
# --- Project constraint ---
if project:
config_manager = ConfigManager()
project_name, _ = config_manager.get_project(project)
if not project_name:
typer.echo(f"No project found named: {project}", err=True)
raise typer.Exit(1)
os.environ["BASIC_MEMORY_MCP_PROJECT"] = project_name
logger.info(f"Watch constrained to project: {project_name}")
# --- Sync coordinator ---
# quiet=False so file change events are printed to the terminal
sync_coordinator = SyncCoordinator(config=config, should_sync=True, quiet=False)
# --- Signal handling ---
shutdown_event = asyncio.Event()
def _signal_handler() -> None:
logger.info("Shutdown signal received")
shutdown_event.set()
loop = asyncio.get_running_loop()
# Windows ProactorEventLoop does not support add_signal_handler;
# fall back to the stdlib signal module which works cross-platform.
try:
for sig in (signal.SIGINT, signal.SIGTERM):
loop.add_signal_handler(sig, _signal_handler)
except NotImplementedError:
for sig in (signal.SIGINT, signal.SIGTERM):
signal.signal(sig, lambda _signum, _frame: _signal_handler())
# --- Run ---
await sync_coordinator.start()
logger.info("Watch service running, press Ctrl+C to stop")
await shutdown_event.wait()
finally:
if sync_coordinator is not None:
await sync_coordinator.stop()
await db.shutdown_db()
logger.info("Watch service stopped")
@app.command()
def watch(
project: Optional[str] = typer.Option(None, help="Restrict watcher to a single project"),
) -> None:
"""Run file watcher as a long-running process (no MCP server).
Watches for file changes in project directories and syncs them to the
database. Useful for running Basic Memory sync alongside external tools
that don't use the MCP server.
"""
# On Windows, use SelectorEventLoop to avoid ProactorEventLoop cleanup issues
if sys.platform == "win32": # pragma: no cover
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
asyncio.run(run_watch(project=project))
src/basic_memory/cli/main.py
-1
View File
@@ -28,7 +28,6 @@ if not _version_only_invocation(sys.argv[1:]):
schema,
status,
tool,
workspace,
)
warnings.filterwarnings("ignore") # pragma: no cover
src/basic_memory/config.py
+11 -9
View File
@@ -10,7 +10,7 @@ from typing import Any, Dict, Literal, Optional, List, Tuple
from enum import Enum
from loguru import logger
from pydantic import BaseModel, Field, model_validator
from pydantic import AliasChoices, BaseModel, Field, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
from basic_memory.utils import setup_logging, generate_permalink
@@ -97,9 +97,10 @@ class ProjectEntry(BaseModel):
description="Routing mode: local (in-process ASGI) or cloud (remote API)",
)
# Cloud sync state (replaces CloudProjectConfig)
cloud_sync_path: Optional[str] = Field(
local_sync_path: Optional[str] = Field(
default=None,
description="Local working directory for bisync (formerly CloudProjectConfig.local_path)",
description="Local working directory for bisync",
validation_alias=AliasChoices("local_sync_path", "cloud_sync_path"),
)
bisync_initialized: bool = Field(
default=False,
@@ -362,26 +363,27 @@ class BasicMemoryConfig(BaseSettings):
if name in cloud_projects:
cp = cloud_projects[name]
if isinstance(cp, dict):
entry["cloud_sync_path"] = cp.get("local_path")
entry["local_sync_path"] = cp.get("local_path")
entry["bisync_initialized"] = cp.get("bisync_initialized", False)
entry["last_sync"] = cp.get("last_sync")
else:
# Already a CloudProjectConfig-like object
entry["cloud_sync_path"] = getattr(cp, "local_path", None)
entry["local_sync_path"] = getattr(cp, "local_path", None)
entry["bisync_initialized"] = getattr(cp, "bisync_initialized", False)
entry["last_sync"] = getattr(cp, "last_sync", None)
new_projects[name] = entry
# Pick up cloud_projects entries not already in projects
# These are cloud-only projects — path is the cloud permalink,
# local_path goes into cloud_sync_path for bisync
# These are cloud-only projects — path should be the local working
# directory (if one exists), local_path goes into local_sync_path for bisync
for name, cp in cloud_projects.items():
if name not in new_projects:
if isinstance(cp, dict):
local_path = cp.get("local_path", "")
new_projects[name] = {
"path": generate_permalink(name),
"path": local_path or "",
"mode": project_modes.get(name, "cloud"),
"cloud_sync_path": cp.get("local_path"),
"local_sync_path": local_path,
"bisync_initialized": cp.get("bisync_initialized", False),
"last_sync": cp.get("last_sync"),
}
src/basic_memory/mcp/async_client.py
+2 -2
View File
@@ -56,7 +56,7 @@ async def _resolve_cloud_token(config) -> str:
raise RuntimeError(
"Cloud routing requested but no credentials found. "
"Run 'bm cloud set-key <key>' or 'bm cloud login' first."
"Run 'bm cloud api-key save <key>' or 'bm cloud login' first."
)
@@ -158,7 +158,7 @@ async def get_client(
except RuntimeError as exc:
raise RuntimeError(
f"Project '{project_name}' is set to cloud mode but no credentials found. "
"Run 'bm cloud set-key <key>' or 'bm cloud login' first."
"Run 'bm cloud api-key save <key>' or 'bm cloud login' first."
) from exc
return
src/basic_memory/mcp/tools/schema.py
+35 -4
View File
@@ -4,7 +4,7 @@ Provides tools for schema validation, inference, and drift detection through the
These tools call the schema API endpoints via the typed SchemaClient.
"""
from typing import Optional
from typing import Literal, Optional
from loguru import logger
from fastmcp import Context
@@ -78,8 +78,9 @@ async def schema_validate(
identifier: Optional[str] = None,
project: Optional[str] = None,
workspace: Optional[str] = None,
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> ValidationReport | str:
) -> ValidationReport | str | dict:
"""Validate notes against their resolved schema.
Validates a specific note (by identifier) or all notes of a given type.
@@ -142,6 +143,8 @@ async def schema_validate(
# Why: can't validate notes that don't exist yet
# Outcome: return guidance on creating notes of this type
if note_type and result.total_entities == 0:
if output_format == "json":
return {"error": f"No notes found of type '{note_type}'"}
return _no_notes_guidance(note_type, "schema_validate")
# --- No schema guard ---
@@ -149,12 +152,19 @@ async def schema_validate(
# Why: notes of this type exist but no schema was found, so none were validated
# Outcome: return guidance on how to create a schema
if note_type and result.total_notes == 0:
if output_format == "json":
return {"error": f"No schema found for type '{note_type}'"}
return _no_schema_guidance(note_type, "schema_validate")
if output_format == "json":
return result.model_dump(mode="json", exclude_none=True)
return result
except Exception as e:
logger.error(f"Schema validation failed: {e}, project: {active_project.name}")
if output_format == "json":
return {"error": f"Schema validation failed: {e}"}
return (
f"# Schema Validation Failed\n\n"
f"Error validating schemas: {e}\n\n"
@@ -174,8 +184,9 @@ async def schema_infer(
threshold: float = 0.25,
project: Optional[str] = None,
workspace: Optional[str] = None,
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> InferenceReport | str:
) -> InferenceReport | str | dict:
"""Analyze existing notes and suggest a schema definition.
Examines observation categories and relation types across all notes
@@ -235,6 +246,13 @@ async def schema_infer(
# Why: returning hundreds of excluded fields overwhelms the LLM context
# Outcome: return actionable guidance instead of a massive empty result
if result.notes_analyzed > 0 and not result.suggested_schema:
if output_format == "json":
return {
"error": (
f"No schema pattern found for '{note_type}' "
f"(threshold: {threshold:.0%})"
)
}
return (
f"# No Schema Pattern Found\n\n"
f"Analyzed {result.notes_analyzed} notes of type '{note_type}', "
@@ -253,10 +271,15 @@ async def schema_infer(
f"structure\n"
)
if output_format == "json":
return result.model_dump(mode="json", exclude_none=True)
return result
except Exception as e:
logger.error(f"Schema inference failed: {e}, project: {active_project.name}")
if output_format == "json":
return {"error": f"Schema inference failed: {e}"}
return (
f"# Schema Inference Failed\n\n"
f"Error inferring schema for type '{note_type}': {e}\n\n"
@@ -275,8 +298,9 @@ async def schema_diff(
note_type: str,
project: Optional[str] = None,
workspace: Optional[str] = None,
output_format: Literal["text", "json"] = "text",
context: Context | None = None,
) -> DriftReport | str:
) -> DriftReport | str | dict:
"""Detect drift between a schema definition and actual note usage.
Compares the existing schema for a note type against how notes of
@@ -331,12 +355,19 @@ async def schema_diff(
# Why: diff requires a schema to compare against
# Outcome: return guidance on how to create a schema
if not result.schema_found:
if output_format == "json":
return {"error": f"No schema found for type '{note_type}'"}
return _no_schema_guidance(note_type, "schema_diff")
if output_format == "json":
return result.model_dump(mode="json", exclude_none=True)
return result
except Exception as e:
logger.error(f"Schema diff failed: {e}, project: {active_project.name}")
if output_format == "json":
return {"error": f"Schema diff failed: {e}"}
return (
f"# Schema Diff Failed\n\n"
f"Error detecting drift for type '{note_type}': {e}\n\n"
src/basic_memory/mcp/tools/utils.py
+1 -1
View File
@@ -125,7 +125,7 @@ def _resolve_error_message(
return (
"Authentication failed: the configured cloud API key was rejected by the server. "
"Basic Memory prioritizes cloud_api_key over OAuth for cloud routing. "
"Fix by running `bm cloud set-key <valid-key>` "
"Fix by running `bm cloud api-key save <valid-key>` "
"or remove `cloud_api_key` and use `bm cloud login`."
)
test-int/cli/test_cli_tool_edit_note_integration.py
+4 -12
View File
@@ -19,8 +19,6 @@ def _write_note(title: str, folder: str, content: str, project: str | None = Non
folder,
"--content",
content,
"--format",
"json",
]
if project:
args.extend(["--project", project])
@@ -31,7 +29,7 @@ def _write_note(title: str, folder: str, content: str, project: str | None = Non
def _read_note(identifier: str, project: str | None = None) -> dict:
args = ["tool", "read-note", identifier, "--format", "json"]
args = ["tool", "read-note", identifier]
if project:
args.extend(["--project", project])
@@ -214,7 +212,7 @@ def test_edit_note_replace_section_fails_without_section(
def test_edit_note_json_format_contract(app, app_config, test_project, config_manager):
"""JSON format returns only metadata keys required by contract."""
"""JSON output returns metadata keys required by contract."""
note = _write_note(
"Edit JSON Note",
"edit-tests",
@@ -231,8 +229,6 @@ def test_edit_note_json_format_contract(app, app_config, test_project, config_ma
"append",
"--content",
"\nJSON_MARKER",
"--format",
"json",
],
)
@@ -243,10 +239,8 @@ def test_edit_note_json_format_contract(app, app_config, test_project, config_ma
assert data["title"] == "Edit JSON Note"
def test_edit_note_text_backend_failure_returns_nonzero(
app, app_config, test_project, config_manager
):
"""Text mode should return non-zero when backend edit operation fails."""
def test_edit_note_backend_failure_returns_nonzero(app, app_config, test_project, config_manager):
"""Edit should return non-zero when backend edit operation fails."""
note = _write_note(
"Edit Backend Failure Note",
"edit-tests",
@@ -271,8 +265,6 @@ def test_edit_note_text_backend_failure_returns_nonzero(
)
assert result.exit_code != 0
assert "# Edit Failed - Wrong Replacement Count" in result.output
assert "Expected 2 occurrences of 'Gamma' but found 1" in result.output
def test_edit_note_project_and_routing_flag_parity(app, app_config, test_project, config_manager):
test-int/cli/test_cli_tool_json_failure_integration.py
+15 -24
View File
@@ -1,4 +1,4 @@
"""Failure-path integration tests for CLI tool --format json output.
"""Failure-path integration tests for CLI tool JSON output.
Verifies that error conditions return proper exit codes and that
error messages go to stderr, not stdout (which would break JSON parsing).
@@ -13,25 +13,22 @@ from basic_memory.cli.main import app as cli_app
runner = CliRunner()
def test_read_note_not_found_json(app, app_config, test_project, config_manager):
"""read-note with non-existent identifier returns error exit code."""
def test_read_note_not_found(app, app_config, test_project, config_manager):
"""read-note with non-existent identifier returns JSON with null fields."""
result = runner.invoke(
cli_app,
["tool", "read-note", "nonexistent-note-that-does-not-exist", "--format", "json"],
["tool", "read-note", "nonexistent-note-that-does-not-exist"],
)
assert result.exit_code != 0, "Should fail for non-existent note"
# stdout should NOT contain valid JSON with data (it's an error)
# The error message should be informative
output = result.stdout + (result.stderr if hasattr(result, "stderr") and result.stderr else "")
assert (
"error" in output.lower()
or "not found" in output.lower()
or "could not find" in output.lower()
)
assert result.exit_code == 0
data = json.loads(result.stdout)
# MCP tool returns a valid JSON payload with null fields for not-found
assert data["title"] is None
assert data["permalink"] is None
assert data["content"] is None
def test_write_note_missing_content_json(app, app_config, test_project, config_manager):
def test_write_note_missing_content(app, app_config, test_project, config_manager):
"""write-note without content or stdin returns error exit code."""
result = runner.invoke(
cli_app,
@@ -42,8 +39,6 @@ def test_write_note_missing_content_json(app, app_config, test_project, config_m
"No Content Note",
"--folder",
"test",
"--format",
"json",
],
input="", # Empty stdin
)
@@ -52,7 +47,7 @@ def test_write_note_missing_content_json(app, app_config, test_project, config_m
assert result.exit_code != 0, "Should fail when no content is provided"
def test_write_note_json_then_read_json_roundtrip(app, app_config, test_project, config_manager):
def test_write_note_then_read_note_roundtrip(app, app_config, test_project, config_manager):
"""write-note JSON output can be used to read-note by permalink."""
# Write a note
write_result = runner.invoke(
@@ -66,8 +61,6 @@ def test_write_note_json_then_read_json_roundtrip(app, app_config, test_project,
"test-roundtrip",
"--content",
"# Roundtrip Test\n\nContent for roundtrip.",
"--format",
"json",
],
)
assert write_result.exit_code == 0
@@ -77,7 +70,7 @@ def test_write_note_json_then_read_json_roundtrip(app, app_config, test_project,
# Read it back using the permalink from the write response
read_result = runner.invoke(
cli_app,
["tool", "read-note", write_data["permalink"], "--format", "json"],
["tool", "read-note", write_data["permalink"]],
)
assert read_result.exit_code == 0
read_data = json.loads(read_result.stdout)
@@ -85,15 +78,13 @@ def test_write_note_json_then_read_json_roundtrip(app, app_config, test_project,
assert read_data["permalink"] == write_data["permalink"]
def test_recent_activity_empty_project_json(
app, app_config, test_project, config_manager, monkeypatch
):
def test_recent_activity_empty_project(app, app_config, test_project, config_manager, monkeypatch):
"""recent-activity on empty project returns valid empty JSON list."""
monkeypatch.setenv("BASIC_MEMORY_MCP_PROJECT", test_project.name)
result = runner.invoke(
cli_app,
["tool", "recent-activity", "--format", "json"],
["tool", "recent-activity"],
)
# Should succeed even if empty
test-int/cli/test_cli_tool_json_integration.py
+16 -76
View File
@@ -1,4 +1,4 @@
"""Integration tests for CLI tool --format json output."""
"""Integration tests for CLI tool JSON output."""
import json
@@ -9,8 +9,8 @@ from basic_memory.cli.main import app as cli_app
runner = CliRunner()
def test_write_note_json_format(app, app_config, test_project, config_manager):
"""Test write-note --format json returns valid JSON with expected keys."""
def test_write_note_json_output(app, app_config, test_project, config_manager):
"""write-note returns valid JSON with expected keys."""
result = runner.invoke(
cli_app,
[
@@ -22,8 +22,6 @@ def test_write_note_json_format(app, app_config, test_project, config_manager):
"test-notes",
"--content",
"# Test\n\nThis is test content.",
"--format",
"json",
],
)
@@ -36,12 +34,11 @@ def test_write_note_json_format(app, app_config, test_project, config_manager):
data = json.loads(result.stdout)
assert data["title"] == "Integration Test Note"
assert "permalink" in data
assert data["content"] == "# Test\n\nThis is test content."
assert "file_path" in data
def test_read_note_json_format(app, app_config, test_project, config_manager):
"""Test read-note --format json returns valid JSON with expected keys."""
def test_read_note_json_output(app, app_config, test_project, config_manager):
"""read-note returns valid JSON with expected keys."""
# First, write a note
write_result = runner.invoke(
cli_app,
@@ -54,8 +51,6 @@ def test_read_note_json_format(app, app_config, test_project, config_manager):
"test-notes",
"--content",
"# Read Test\n\nContent to read back.",
"--format",
"json",
],
)
assert write_result.exit_code == 0
@@ -65,7 +60,7 @@ def test_read_note_json_format(app, app_config, test_project, config_manager):
# Now read it back
result = runner.invoke(
cli_app,
["tool", "read-note", permalink, "--format", "json"],
["tool", "read-note", permalink],
)
if result.exit_code != 0:
@@ -78,25 +73,21 @@ def test_read_note_json_format(app, app_config, test_project, config_manager):
assert data["permalink"] == permalink
assert "content" in data
assert "file_path" in data
assert "frontmatter" in data
assert isinstance(data["frontmatter"], dict)
def test_read_note_json_strip_frontmatter_permalink(app, app_config, test_project, config_manager):
"""read-note strips frontmatter in JSON mode for permalink lookup."""
def test_read_note_include_frontmatter(app, app_config, test_project, config_manager):
"""read-note --include-frontmatter includes frontmatter in output."""
write_result = runner.invoke(
cli_app,
[
"tool",
"write-note",
"--title",
"Read Strip Permalink Note",
"Read Frontmatter Note",
"--folder",
"test-notes",
"--content",
"# Read Strip Permalink Note\n\nPermalink lookup content.",
"--format",
"json",
"# Read Frontmatter Note\n\nFrontmatter test content.",
],
)
assert write_result.exit_code == 0
@@ -108,68 +99,19 @@ def test_read_note_json_strip_frontmatter_permalink(app, app_config, test_projec
"tool",
"read-note",
write_data["permalink"],
"--format",
"json",
"--strip-frontmatter",
"--include-frontmatter",
],
)
assert result.exit_code == 0
data = json.loads(result.stdout)
assert data["title"] == "Read Strip Permalink Note"
assert data["title"] == "Read Frontmatter Note"
assert data["permalink"] == write_data["permalink"]
assert not data["content"].startswith("---")
assert "# Read Strip Permalink Note" in data["content"]
assert isinstance(data["frontmatter"], dict)
assert data["frontmatter"].get("title") == "Read Strip Permalink Note"
assert "content" in data
def test_read_note_json_strip_frontmatter_title(app, app_config, test_project, config_manager):
"""read-note strips frontmatter in JSON mode for title-based lookup."""
write_result = runner.invoke(
cli_app,
[
"tool",
"write-note",
"--title",
"Read Strip Title Note",
"--folder",
"test-notes",
"--content",
"# Read Strip Title Note\n\nTitle lookup content.",
"--format",
"json",
],
)
assert write_result.exit_code == 0
write_data = json.loads(write_result.stdout)
result = runner.invoke(
cli_app,
[
"tool",
"read-note",
"Read Strip Title Note",
"--format",
"json",
"--strip-frontmatter",
],
)
assert result.exit_code == 0
data = json.loads(result.stdout)
assert data["title"] == "Read Strip Title Note"
assert data["permalink"] == write_data["permalink"]
assert not data["content"].startswith("---")
assert "# Read Strip Title Note" in data["content"]
assert isinstance(data["frontmatter"], dict)
assert data["frontmatter"].get("title") == "Read Strip Title Note"
def test_recent_activity_json_format(app, app_config, test_project, config_manager, monkeypatch):
"""Test recent-activity --format json returns valid JSON list."""
# _recent_activity_json uses resolve_project_parameter which requires either
# default_project set or BASIC_MEMORY_MCP_PROJECT to resolve a project
def test_recent_activity_json_output(app, app_config, test_project, config_manager, monkeypatch):
"""recent-activity returns valid JSON list."""
monkeypatch.setenv("BASIC_MEMORY_MCP_PROJECT", test_project.name)
# Write a note to ensure there's recent activity
@@ -184,8 +126,6 @@ def test_recent_activity_json_format(app, app_config, test_project, config_manag
"test-notes",
"--content",
"# Activity\n\nTest content for activity.",
"--format",
"json",
],
)
assert write_result.exit_code == 0
@@ -193,7 +133,7 @@ def test_recent_activity_json_format(app, app_config, test_project, config_manag
# Get recent activity
result = runner.invoke(
cli_app,
["tool", "recent-activity", "--format", "json"],
["tool", "recent-activity"],
)
if result.exit_code != 0:
test-int/cli/test_routing_integration.py
-2
View File
@@ -189,7 +189,6 @@ class TestToolCommandsAcceptFlags:
("read-note", ["test"]),
("edit-note", ["test", "--operation", "append", "--content", "test"]),
("build-context", ["memory://test"]),
("continue-conversation", []),
],
)
def test_tool_commands_accept_local_flag(self, command, args, app_config):
@@ -207,7 +206,6 @@ class TestToolCommandsAcceptFlags:
("read-note", ["test"]),
("edit-note", ["test", "--operation", "append", "--content", "test"]),
("build-context", ["memory://test"]),
("continue-conversation", []),
],
)
def test_tool_commands_accept_cloud_flag(self, command, args, app_config):
test-int/cli/test_search_notes_meta_integration.py
-4
View File
@@ -25,8 +25,6 @@ def test_search_notes_query_plus_meta_filter(app, app_config, test_project, conf
"meta-tests",
"--content",
active_content,
"--format",
"json",
],
)
assert active_write.exit_code == 0, active_write.output
@@ -43,8 +41,6 @@ def test_search_notes_query_plus_meta_filter(app, app_config, test_project, conf
"meta-tests",
"--content",
inactive_content,
"--format",
"json",
],
)
assert inactive_write.exit_code == 0, inactive_write.output
tests/cli/test_cli_schema.py
+302
View File
@@ -0,0 +1,302 @@
"""Tests for CLI schema commands (Rich output).
Tests mock the MCP tool functions and verify Rich-formatted output.
"""
from unittest.mock import AsyncMock, MagicMock, patch
from typer.testing import CliRunner
from basic_memory.cli.main import app as cli_app
runner = CliRunner()
# --- Shared mock data ---
VALIDATE_REPORT = {
"entity_type": "person",
"total_notes": 2,
"total_entities": 2,
"valid_count": 1,
"warning_count": 1,
"error_count": 1,
"results": [
{
"note_identifier": "people/alice",
"schema_entity": "person",
"passed": True,
"warnings": [],
"errors": [],
},
{
"note_identifier": "people/bob",
"schema_entity": "person",
"passed": False,
"warnings": ["Missing optional field: role"],
"errors": ["Missing required field: name"],
},
],
}
INFER_REPORT = {
"entity_type": "person",
"notes_analyzed": 5,
"field_frequencies": [
{"name": "name", "source": "observation", "count": 5, "total": 5, "percentage": 1.0},
{"name": "role", "source": "observation", "count": 3, "total": 5, "percentage": 0.6},
],
"suggested_schema": {"name": "string, full name", "role?": "string, job title"},
"suggested_required": ["name"],
"suggested_optional": ["role"],
"excluded": [],
}
DIFF_REPORT_WITH_DRIFT = {
"entity_type": "person",
"schema_found": True,
"new_fields": [
{"name": "email", "source": "observation", "count": 3, "total": 5, "percentage": 0.6}
],
"dropped_fields": [
{"name": "phone", "source": "observation", "count": 0, "total": 5, "percentage": 0.0}
],
"cardinality_changes": ["role: single -> array"],
}
DIFF_REPORT_NO_DRIFT = {
"entity_type": "person",
"schema_found": True,
"new_fields": [],
"dropped_fields": [],
"cardinality_changes": [],
}
def _mock_config_manager():
"""Create a mock ConfigManager that avoids reading real config."""
mock_cm = MagicMock()
mock_cm.config = MagicMock()
mock_cm.default_project = "test-project"
mock_cm.get_project.return_value = ("test-project", "/tmp/test")
return mock_cm
# --- validate ---
@patch("basic_memory.cli.commands.schema.ConfigManager")
@patch(
"basic_memory.cli.commands.schema.mcp_schema_validate",
new_callable=AsyncMock,
return_value=VALIDATE_REPORT,
)
def test_validate_renders_table(mock_mcp, mock_config_cls):
"""bm schema validate renders a Rich table with results."""
mock_config_cls.return_value = _mock_config_manager()
result = runner.invoke(cli_app, ["schema", "validate", "person"])
assert result.exit_code == 0, f"CLI failed: {result.output}"
assert "Schema Validation" in result.output
assert "people/alice" in result.output
assert "people/bob" in result.output
assert "1/2 valid" in result.output
mock_mcp.assert_called_once()
assert mock_mcp.call_args.kwargs["output_format"] == "json"
@patch("basic_memory.cli.commands.schema.ConfigManager")
@patch(
"basic_memory.cli.commands.schema.mcp_schema_validate",
new_callable=AsyncMock,
return_value=VALIDATE_REPORT,
)
def test_validate_strict_exits_on_errors(mock_mcp, mock_config_cls):
"""bm schema validate --strict exits with code 1 when errors exist."""
mock_config_cls.return_value = _mock_config_manager()
result = runner.invoke(cli_app, ["schema", "validate", "person", "--strict"])
assert result.exit_code == 1
@patch("basic_memory.cli.commands.schema.ConfigManager")
@patch(
"basic_memory.cli.commands.schema.mcp_schema_validate",
new_callable=AsyncMock,
return_value={"error": "No notes found of type 'person'"},
)
def test_validate_error_response(mock_mcp, mock_config_cls):
"""bm schema validate shows error message from MCP tool."""
mock_config_cls.return_value = _mock_config_manager()
result = runner.invoke(cli_app, ["schema", "validate", "person"])
assert result.exit_code == 0
assert "No notes found" in result.output
@patch("basic_memory.cli.commands.schema.ConfigManager")
@patch(
"basic_memory.cli.commands.schema.mcp_schema_validate",
new_callable=AsyncMock,
return_value=VALIDATE_REPORT,
)
def test_validate_identifier_heuristic(mock_mcp, mock_config_cls):
"""bm schema validate treats target with / as identifier."""
mock_config_cls.return_value = _mock_config_manager()
result = runner.invoke(cli_app, ["schema", "validate", "people/alice.md"])
assert result.exit_code == 0, f"CLI failed: {result.output}"
assert mock_mcp.call_args.kwargs["identifier"] == "people/alice.md"
assert mock_mcp.call_args.kwargs["note_type"] is None
# --- infer ---
@patch("basic_memory.cli.commands.schema.ConfigManager")
@patch(
"basic_memory.cli.commands.schema.mcp_schema_infer",
new_callable=AsyncMock,
return_value=INFER_REPORT,
)
def test_infer_renders_table(mock_mcp, mock_config_cls):
"""bm schema infer renders frequency table and suggested schema."""
mock_config_cls.return_value = _mock_config_manager()
result = runner.invoke(cli_app, ["schema", "infer", "person"])
assert result.exit_code == 0, f"CLI failed: {result.output}"
assert "Field Frequencies" in result.output
assert "name" in result.output
assert "Suggested schema" in result.output
mock_mcp.assert_called_once()
assert mock_mcp.call_args.kwargs["output_format"] == "json"
@patch("basic_memory.cli.commands.schema.ConfigManager")
@patch(
"basic_memory.cli.commands.schema.mcp_schema_infer",
new_callable=AsyncMock,
return_value=INFER_REPORT,
)
def test_infer_threshold_passthrough(mock_mcp, mock_config_cls):
"""bm schema infer passes --threshold through to MCP tool."""
mock_config_cls.return_value = _mock_config_manager()
result = runner.invoke(cli_app, ["schema", "infer", "person", "--threshold", "0.5"])
assert result.exit_code == 0, f"CLI failed: {result.output}"
assert mock_mcp.call_args.kwargs["threshold"] == 0.5
@patch("basic_memory.cli.commands.schema.ConfigManager")
@patch(
"basic_memory.cli.commands.schema.mcp_schema_infer",
new_callable=AsyncMock,
return_value={"error": "No schema pattern found for 'person' (threshold: 25%)"},
)
def test_infer_error_response(mock_mcp, mock_config_cls):
"""bm schema infer shows error message from MCP tool."""
mock_config_cls.return_value = _mock_config_manager()
result = runner.invoke(cli_app, ["schema", "infer", "person"])
assert result.exit_code == 0
assert "No schema pattern found" in result.output
@patch("basic_memory.cli.commands.schema.ConfigManager")
@patch(
"basic_memory.cli.commands.schema.mcp_schema_infer",
new_callable=AsyncMock,
return_value={
"entity_type": "person",
"notes_analyzed": 0,
"field_frequencies": [],
"suggested_schema": {},
"suggested_required": [],
"suggested_optional": [],
"excluded": [],
},
)
def test_infer_zero_notes(mock_mcp, mock_config_cls):
"""bm schema infer shows message when zero notes found."""
mock_config_cls.return_value = _mock_config_manager()
result = runner.invoke(cli_app, ["schema", "infer", "person"])
assert result.exit_code == 0
assert "No notes found" in result.output
# --- diff ---
@patch("basic_memory.cli.commands.schema.ConfigManager")
@patch(
"basic_memory.cli.commands.schema.mcp_schema_diff",
new_callable=AsyncMock,
return_value=DIFF_REPORT_WITH_DRIFT,
)
def test_diff_renders_drift(mock_mcp, mock_config_cls):
"""bm schema diff shows new/dropped fields and cardinality changes."""
mock_config_cls.return_value = _mock_config_manager()
result = runner.invoke(cli_app, ["schema", "diff", "person"])
assert result.exit_code == 0, f"CLI failed: {result.output}"
assert "drift detected" in result.output
assert "email" in result.output
assert "phone" in result.output
assert "role: single -> array" in result.output
mock_mcp.assert_called_once()
assert mock_mcp.call_args.kwargs["output_format"] == "json"
@patch("basic_memory.cli.commands.schema.ConfigManager")
@patch(
"basic_memory.cli.commands.schema.mcp_schema_diff",
new_callable=AsyncMock,
return_value=DIFF_REPORT_NO_DRIFT,
)
def test_diff_no_drift(mock_mcp, mock_config_cls):
"""bm schema diff shows success message when no drift found."""
mock_config_cls.return_value = _mock_config_manager()
result = runner.invoke(cli_app, ["schema", "diff", "person"])
assert result.exit_code == 0, f"CLI failed: {result.output}"
assert "No drift detected" in result.output
@patch("basic_memory.cli.commands.schema.ConfigManager")
@patch(
"basic_memory.cli.commands.schema.mcp_schema_diff",
new_callable=AsyncMock,
return_value={"error": "No schema found for type 'person'"},
)
def test_diff_error_response(mock_mcp, mock_config_cls):
"""bm schema diff shows error message from MCP tool."""
mock_config_cls.return_value = _mock_config_manager()
result = runner.invoke(cli_app, ["schema", "diff", "person"])
assert result.exit_code == 0
assert "No schema found" in result.output
# --- Routing flags ---
def test_schema_routing_both_flags_error():
"""Schema commands exit with error when both --local and --cloud are specified."""
result = runner.invoke(
cli_app,
["schema", "validate", "person", "--local", "--cloud"],
)
assert result.exit_code == 1
tests/cli/test_cli_tool_json_output.py
+558 -286
View File
File diff suppressed because it is too large Load Diff
tests/cli/test_project_add_with_local_path.py
+4 -4
View File
@@ -109,7 +109,7 @@ def test_project_add_with_local_path_saves_to_config(
assert "test-project" in config_data["projects"]
entry = config_data["projects"]["test-project"]
# Use as_posix() for cross-platform compatibility (Windows uses backslashes)
assert entry["cloud_sync_path"] == local_sync_dir.as_posix()
assert entry["local_sync_path"] == local_sync_dir.as_posix()
assert entry.get("last_sync") is None
assert entry.get("bisync_initialized", False) is False
@@ -131,10 +131,10 @@ def test_project_add_without_local_path_no_config_entry(runner, mock_config, moc
# Verify config was NOT updated with cloud sync path
config_data = json.loads(mock_config.read_text())
# Project may or may not be in config, but if it is, cloud_sync_path should be null
# Project may or may not be in config, but if it is, local_sync_path should be null
entry = config_data.get("projects", {}).get("test-project")
if entry:
assert entry.get("cloud_sync_path") is None
assert entry.get("local_sync_path") is None
def test_project_add_local_path_expands_tilde(runner, mock_config, mock_api_client):
@@ -148,7 +148,7 @@ def test_project_add_local_path_expands_tilde(runner, mock_config, mock_api_clie
# Verify config has expanded path
config_data = json.loads(mock_config.read_text())
local_path = config_data["projects"]["test-project"]["cloud_sync_path"]
local_path = config_data["projects"]["test-project"]["local_sync_path"]
# Path should be absolute (starts with / on Unix or drive letter on Windows)
assert Path(local_path).is_absolute()
assert "~" not in local_path
tests/cli/test_project_list_and_ls.py
+58 -4
View File
@@ -64,7 +64,7 @@ def test_project_list_shows_local_cloud_presence_and_routes(
"beta": {
"path": beta_local_sync,
"mode": "cloud",
"cloud_sync_path": beta_local_sync,
"local_sync_path": beta_local_sync,
},
},
"default_project": "alpha",
@@ -140,10 +140,10 @@ def test_project_list_shows_local_cloud_presence_and_routes(
assert "/beta" in result.stdout
def test_project_ls_defaults_to_local_route(
def test_project_ls_local_mode_defaults_to_local_route(
runner: CliRunner, write_config, mock_client, tmp_path, monkeypatch
):
"""project ls without flags should list local files and not require cloud credentials."""
"""project ls without flags for a local-mode project should list local files."""
project_dir = tmp_path / "alpha-files"
(project_dir / "docs").mkdir(parents=True, exist_ok=True)
(project_dir / "notes.md").write_text("# local note")
@@ -152,7 +152,7 @@ def test_project_ls_defaults_to_local_route(
write_config(
{
"env": "dev",
"projects": {"alpha": {"path": project_dir.as_posix(), "mode": "cloud"}},
"projects": {"alpha": {"path": project_dir.as_posix(), "mode": "local"}},
"default_project": "alpha",
}
)
@@ -193,6 +193,60 @@ def test_project_ls_defaults_to_local_route(
assert "docs/spec.md" in result.stdout
def test_project_ls_cloud_mode_defaults_to_cloud_route(
runner: CliRunner, write_config, mock_client, tmp_path, monkeypatch
):
"""project ls without flags for a cloud-mode project should list cloud files."""
write_config(
{
"env": "dev",
"projects": {"alpha": {"path": str(tmp_path / "alpha"), "mode": "cloud"}},
"default_project": "alpha",
"cloud_api_key": "bmc_test_key_123",
}
)
cloud_payload = {
"projects": [
{
"id": 1,
"external_id": "11111111-1111-1111-1111-111111111111",
"name": "alpha",
"path": "/alpha",
"is_default": True,
}
],
"default_project": "alpha",
}
class _Resp:
def json(self):
return cloud_payload
class _TenantInfo:
bucket_name = "tenant-bucket"
async def fake_call_get(client, path: str, **kwargs):
assert path == "/v2/projects/"
# Cloud routing should be active when project mode is cloud
assert os.getenv("BASIC_MEMORY_FORCE_CLOUD", "").lower() in ("true", "1", "yes")
return _Resp()
async def fake_get_mount_info():
return _TenantInfo()
monkeypatch.setattr(project_cmd, "call_get", fake_call_get)
monkeypatch.setattr(project_cmd, "get_mount_info", fake_get_mount_info)
monkeypatch.setattr(project_cmd, "project_ls", lambda *args, **kwargs: [" 42 cloud.md"])
# No --cloud flag: project mode should determine route
result = runner.invoke(app, ["project", "ls", "--name", "alpha"], env={"COLUMNS": "200"})
assert result.exit_code == 0, f"Exit code: {result.exit_code}, output: {result.stdout}"
assert "Files in alpha (CLOUD)" in result.stdout
assert "cloud.md" in result.stdout
def test_project_ls_cloud_route_uses_cloud_listing(
runner: CliRunner, write_config, mock_client, tmp_path, monkeypatch
):
tests/cli/test_watch.py
-278
View File
@@ -1,278 +0,0 @@
"""Tests for the watch CLI command."""
import asyncio
import os
import signal
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import typer
from basic_memory.cli.commands.watch import run_watch
from basic_memory.config import BasicMemoryConfig
@pytest.fixture
def mock_config():
"""Create a mock config for testing."""
return BasicMemoryConfig()
@pytest.fixture
def mock_container(mock_config):
"""Create a mock CLI container."""
container = MagicMock()
container.config = mock_config
return container
class TestRunWatch:
"""Tests for run_watch async function."""
@pytest.mark.asyncio
async def test_initializes_app(self, mock_container):
"""run_watch calls initialize_app with the container's config."""
mock_init = AsyncMock()
with (
patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
patch("basic_memory.cli.commands.watch.initialize_app", mock_init),
patch("basic_memory.cli.commands.watch.SyncCoordinator") as mock_coordinator_cls,
patch("basic_memory.cli.commands.watch.db") as mock_db,
):
# Make coordinator.start() set the shutdown event so we don't block
mock_coordinator = AsyncMock()
mock_coordinator_cls.return_value = mock_coordinator
async def start_then_shutdown():
# Simulate immediate shutdown after start
pass
mock_coordinator.start = start_then_shutdown
mock_coordinator.stop = AsyncMock()
mock_db.shutdown_db = AsyncMock()
# Patch signal handlers and make shutdown_event fire immediately
with patch("asyncio.get_running_loop") as mock_loop:
mock_loop_instance = MagicMock()
mock_loop.return_value = mock_loop_instance
# Capture the signal handler so we can trigger it
signal_handlers = {}
def capture_handler(sig, handler):
signal_handlers[sig] = handler
mock_loop_instance.add_signal_handler.side_effect = capture_handler
# Run in a task so we can trigger shutdown
async def run_and_shutdown():
task = asyncio.create_task(run_watch())
# Give it a moment to start
await asyncio.sleep(0.01)
# Trigger shutdown via captured signal handler
import signal
if signal.SIGINT in signal_handlers:
signal_handlers[signal.SIGINT]()
await task
await run_and_shutdown()
mock_init.assert_called_once_with(mock_container.config)
@pytest.mark.asyncio
async def test_creates_coordinator_with_quiet_false(self, mock_container):
"""SyncCoordinator is created with should_sync=True and quiet=False."""
with (
patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
patch("basic_memory.cli.commands.watch.initialize_app", AsyncMock()),
patch("basic_memory.cli.commands.watch.SyncCoordinator") as mock_coordinator_cls,
patch("basic_memory.cli.commands.watch.db") as mock_db,
):
mock_coordinator = AsyncMock()
mock_coordinator_cls.return_value = mock_coordinator
mock_db.shutdown_db = AsyncMock()
with patch("asyncio.get_running_loop") as mock_loop:
mock_loop_instance = MagicMock()
mock_loop.return_value = mock_loop_instance
signal_handlers = {}
def capture_handler(sig, handler):
signal_handlers[sig] = handler
mock_loop_instance.add_signal_handler.side_effect = capture_handler
async def run_and_shutdown():
task = asyncio.create_task(run_watch())
await asyncio.sleep(0.01)
import signal
if signal.SIGINT in signal_handlers:
signal_handlers[signal.SIGINT]()
await task
await run_and_shutdown()
mock_coordinator_cls.assert_called_once_with(
config=mock_container.config,
should_sync=True,
quiet=False,
)
@pytest.mark.asyncio
async def test_project_sets_env_var(self, mock_container):
"""--project validates and sets BASIC_MEMORY_MCP_PROJECT env var."""
mock_config_manager = MagicMock()
mock_config_manager.get_project.return_value = ("my-project", "/some/path")
with (
patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
patch("basic_memory.cli.commands.watch.initialize_app", AsyncMock()),
patch(
"basic_memory.cli.commands.watch.ConfigManager",
return_value=mock_config_manager,
),
patch("basic_memory.cli.commands.watch.SyncCoordinator") as mock_coordinator_cls,
patch("basic_memory.cli.commands.watch.db") as mock_db,
patch.dict(os.environ, {}, clear=False),
):
mock_coordinator = AsyncMock()
mock_coordinator_cls.return_value = mock_coordinator
mock_db.shutdown_db = AsyncMock()
with patch("asyncio.get_running_loop") as mock_loop:
mock_loop_instance = MagicMock()
mock_loop.return_value = mock_loop_instance
signal_handlers = {}
def capture_handler(sig, handler):
signal_handlers[sig] = handler
mock_loop_instance.add_signal_handler.side_effect = capture_handler
async def run_and_shutdown():
task = asyncio.create_task(run_watch(project="my-project"))
await asyncio.sleep(0.01)
import signal
if signal.SIGINT in signal_handlers:
signal_handlers[signal.SIGINT]()
await task
await run_and_shutdown()
assert os.environ.get("BASIC_MEMORY_MCP_PROJECT") == "my-project"
# Clean up env var
os.environ.pop("BASIC_MEMORY_MCP_PROJECT", None)
@pytest.mark.asyncio
async def test_invalid_project_exits_with_error(self, mock_container):
"""--project with unknown name exits with error and still cleans up DB."""
mock_config_manager = MagicMock()
mock_config_manager.get_project.return_value = (None, None)
with (
patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
patch("basic_memory.cli.commands.watch.initialize_app", AsyncMock()),
patch(
"basic_memory.cli.commands.watch.ConfigManager",
return_value=mock_config_manager,
),
patch("basic_memory.cli.commands.watch.db") as mock_db,
):
mock_db.shutdown_db = AsyncMock()
with pytest.raises(typer.Exit) as exc_info:
await run_watch(project="nonexistent")
assert exc_info.value.exit_code == 1
# DB should still be cleaned up even on early exit
mock_db.shutdown_db.assert_called_once()
@pytest.mark.asyncio
async def test_shutdown_stops_coordinator_and_db(self, mock_container):
"""On shutdown, coordinator.stop() and db.shutdown_db() are called."""
with (
patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
patch("basic_memory.cli.commands.watch.initialize_app", AsyncMock()),
patch("basic_memory.cli.commands.watch.SyncCoordinator") as mock_coordinator_cls,
patch("basic_memory.cli.commands.watch.db") as mock_db,
):
mock_coordinator = AsyncMock()
mock_coordinator_cls.return_value = mock_coordinator
mock_db.shutdown_db = AsyncMock()
with patch("asyncio.get_running_loop") as mock_loop:
mock_loop_instance = MagicMock()
mock_loop.return_value = mock_loop_instance
signal_handlers = {}
def capture_handler(sig, handler):
signal_handlers[sig] = handler
mock_loop_instance.add_signal_handler.side_effect = capture_handler
async def run_and_shutdown():
task = asyncio.create_task(run_watch())
await asyncio.sleep(0.01)
import signal
if signal.SIGINT in signal_handlers:
signal_handlers[signal.SIGINT]()
await task
await run_and_shutdown()
mock_coordinator.stop.assert_called_once()
mock_db.shutdown_db.assert_called_once()
@pytest.mark.asyncio
async def test_falls_back_to_signal_module_on_windows(self, mock_container):
"""When add_signal_handler raises NotImplementedError, falls back to signal.signal()."""
with (
patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
patch("basic_memory.cli.commands.watch.initialize_app", AsyncMock()),
patch("basic_memory.cli.commands.watch.SyncCoordinator") as mock_coordinator_cls,
patch("basic_memory.cli.commands.watch.db") as mock_db,
):
mock_coordinator = AsyncMock()
mock_coordinator_cls.return_value = mock_coordinator
mock_db.shutdown_db = AsyncMock()
with patch("asyncio.get_running_loop") as mock_loop:
mock_loop_instance = MagicMock()
mock_loop.return_value = mock_loop_instance
# Simulate Windows: add_signal_handler raises NotImplementedError
mock_loop_instance.add_signal_handler.side_effect = NotImplementedError
with patch("basic_memory.cli.commands.watch.signal.signal") as mock_signal:
# Track calls to signal.signal for the fallback path
registered_handlers = {}
def capture_signal(sig, handler):
registered_handlers[sig] = handler
mock_signal.side_effect = capture_signal
async def run_and_shutdown():
task = asyncio.create_task(run_watch())
await asyncio.sleep(0.01)
# Trigger shutdown via the fallback handler
if signal.SIGINT in registered_handlers:
registered_handlers[signal.SIGINT](signal.SIGINT, None)
await task
await run_and_shutdown()
# Verify fallback signal.signal was called for both signals
assert mock_signal.call_count == 2
called_signals = {call.args[0] for call in mock_signal.call_args_list}
assert signal.SIGINT in called_signals
assert signal.SIGTERM in called_signals
tests/cli/test_workspace_commands.py
+1 -1
View File
@@ -7,7 +7,7 @@ from basic_memory.cli.app import app
from basic_memory.schemas.cloud import WorkspaceInfo
# Importing registers workspace commands on the shared app instance.
import basic_memory.cli.commands.workspace as workspace_cmd # noqa: F401
import basic_memory.cli.commands.cloud.workspace as workspace_cmd # noqa: F401
@pytest.fixture
tests/mcp/test_tool_contracts.py
+3 -3
View File
@@ -69,9 +69,9 @@ EXPECTED_TOOL_SIGNATURES: dict[str, list[str]] = {
"workspace",
"output_format",
],
"schema_diff": ["note_type", "project", "workspace"],
"schema_infer": ["note_type", "threshold", "project", "workspace"],
"schema_validate": ["note_type", "identifier", "project", "workspace"],
"schema_diff": ["note_type", "project", "workspace", "output_format"],
"schema_infer": ["note_type", "threshold", "project", "workspace", "output_format"],
"schema_validate": ["note_type", "identifier", "project", "workspace", "output_format"],
"search": ["query"],
"search_by_metadata": ["filters", "project", "workspace", "limit", "offset"],
"search_notes": [
tests/mcp/test_tool_utils_cloud_auth.py
+1 -1
View File
@@ -52,5 +52,5 @@ async def test_call_post_401_with_cloud_key_shows_actionable_remediation(config_
message = str(exc.value)
assert "configured cloud API key was rejected" in message
assert "bm cloud set-key <valid-key>" in message
assert "bm cloud api-key save <valid-key>" in message
assert "cloud_api_key" in message
tests/test_config.py
+7 -7
View File
@@ -335,7 +335,7 @@ class TestConfigManager:
config = config_manager.load_config()
entry = config.projects["main"]
assert entry.cloud_sync_path is None
assert entry.local_sync_path is None
assert entry.bisync_initialized is False
assert entry.last_sync is None
@@ -357,7 +357,7 @@ class TestConfigManager:
"research": {
"path": str(temp_path / "research"),
"mode": "cloud",
"cloud_sync_path": str(temp_path / "research-local"),
"local_sync_path": str(temp_path / "research-local"),
"last_sync": now.isoformat(),
"bisync_initialized": True,
},
@@ -369,7 +369,7 @@ class TestConfigManager:
loaded_config = config_manager.load_config()
assert "research" in loaded_config.projects
entry = loaded_config.projects["research"]
assert entry.cloud_sync_path == str(temp_path / "research-local")
assert entry.local_sync_path == str(temp_path / "research-local")
assert entry.bisync_initialized is True
assert entry.last_sync == now
@@ -389,14 +389,14 @@ class TestConfigManager:
# Load, modify, and save
config = config_manager.load_config()
assert config.projects["main"].cloud_sync_path is None
assert config.projects["main"].local_sync_path is None
config.projects["main"].cloud_sync_path = str(temp_path / "work-local")
config.projects["main"].local_sync_path = str(temp_path / "work-local")
config_manager.save_config(config)
# Reload and verify persistence
reloaded_config = config_manager.load_config()
assert reloaded_config.projects["main"].cloud_sync_path == str(temp_path / "work-local")
assert reloaded_config.projects["main"].local_sync_path == str(temp_path / "work-local")
assert reloaded_config.projects["main"].bisync_initialized is False
def test_backward_compatibility_loading_old_format_config(self):
@@ -469,7 +469,7 @@ class TestConfigManager:
# Verify migration
assert config.projects["research"].mode == ProjectMode.CLOUD
assert config.projects["research"].cloud_sync_path == str(temp_path / "research-local")
assert config.projects["research"].local_sync_path == str(temp_path / "research-local")
assert config.projects["research"].bisync_initialized is True
assert config.projects["main"].mode == ProjectMode.LOCAL