mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
feat(cli): add bm cloud share command group (#880)
Surface the cloud /api/shares endpoints from the CLI so users can manage
public share links for notes without leaving the terminal.
New `bm cloud share` subcommand group:
- create <project> <permalink> [--expires-at] -> POST /api/shares
- list [--project] -> GET /api/shares
- update <token> [--enable|--disable|--expires-at] -> PATCH /api/shares/{token}
- revoke <token> [--force] -> DELETE /api/shares/{token}
Reuses the existing make_api_request() helper, ConfigManager cloud_host
lookup, and SubscriptionRequired/CloudAPIError handling that snapshot.py
uses. Rich table output mirrors `bm project list`. Payloads match the
cloud CreateShareRequest/UpdateShareRequest contracts (project_name,
note_permalink, expires_at, enabled).
Adds PublicShareResponse/PublicShareListResponse schemas and unit tests
with mocked HTTP following the snapshot CLI test patterns.
Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: Drew Cain <groksrc@gmail.com>
This commit is contained in:
Drew Cain
@@ -11,9 +11,11 @@ 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
|
||||
from basic_memory.cli.commands.cloud.shares import share_app
|
||||
|
||||
cloud_app.add_typer(snapshot_app, name="snapshot")
|
||||
cloud_app.add_typer(workspace_app, name="workspace")
|
||||
cloud_app.add_typer(share_app, name="share")
|
||||
|
||||
# Register restore command (directly on cloud_app via decorator)
|
||||
from basic_memory.cli.commands.cloud.restore import restore # noqa: F401, E402
|
||||
|
||||
@@ -53,3 +53,30 @@ class BucketSnapshotRestoreResponse(BaseModel):
|
||||
restored: list[str]
|
||||
snapshot_version: str
|
||||
snapshot_id: UUID
|
||||
|
||||
|
||||
class PublicShareResponse(BaseModel):
|
||||
"""Response model for a public share link.
|
||||
|
||||
Mirrors PublicShareResponse in basic-memory-cloud
|
||||
(apps/cloud/.../schemas/public_share_schemas.py).
|
||||
"""
|
||||
|
||||
id: UUID
|
||||
token: str
|
||||
project_name: str
|
||||
note_permalink: str
|
||||
note_external_id: str
|
||||
enabled: bool
|
||||
expires_at: datetime | None
|
||||
share_url: str
|
||||
view_count: int
|
||||
last_viewed_at: datetime | None
|
||||
created_at: datetime
|
||||
|
||||
|
||||
class PublicShareListResponse(BaseModel):
|
||||
"""Response from listing public shares."""
|
||||
|
||||
shares: list[PublicShareResponse]
|
||||
total: int
|
||||
|
||||
@@ -0,0 +1,377 @@
|
||||
"""Public share CLI commands for Basic Memory Cloud.
|
||||
|
||||
Surfaces the cloud `/api/shares` endpoints so users can manage public share
|
||||
links for notes without leaving the terminal:
|
||||
|
||||
- POST /api/shares -> create
|
||||
- GET /api/shares -> list
|
||||
- PATCH /api/shares/{token} -> update (enable/disable, set expiration)
|
||||
- DELETE /api/shares/{token} -> revoke
|
||||
|
||||
Auth, config lookup, and error handling reuse the shared `make_api_request()`
|
||||
helper, matching the `snapshot.py` command group.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
import typer
|
||||
from rich.console import Console
|
||||
from rich.table import Table
|
||||
|
||||
from basic_memory.cli.commands.cloud.api_client import (
|
||||
CloudAPIError,
|
||||
SubscriptionRequiredError,
|
||||
make_api_request,
|
||||
)
|
||||
from basic_memory.config import ConfigManager
|
||||
|
||||
console = Console()
|
||||
share_app = typer.Typer(help="Manage public share links for notes")
|
||||
|
||||
|
||||
def _format_timestamp(iso_timestamp: Optional[str]) -> str:
|
||||
"""Format an ISO timestamp to a human-readable form, or '-' when absent."""
|
||||
if not iso_timestamp:
|
||||
return "-"
|
||||
try:
|
||||
dt = datetime.fromisoformat(iso_timestamp.replace("Z", "+00:00"))
|
||||
return dt.strftime("%Y-%m-%d %H:%M:%S")
|
||||
except (ValueError, AttributeError):
|
||||
return iso_timestamp
|
||||
|
||||
|
||||
def _parse_expires_at(value: str) -> str:
|
||||
"""Validate an --expires-at value and normalize it to an ISO 8601 string.
|
||||
|
||||
Accepts either a full ISO timestamp ("2025-12-31T23:59:00") or a bare date
|
||||
("2025-12-31"). Exits with a clear error on anything we can't parse so the
|
||||
server never sees a malformed payload.
|
||||
"""
|
||||
try:
|
||||
dt = datetime.fromisoformat(value)
|
||||
except ValueError:
|
||||
console.print(
|
||||
f"[red]Invalid --expires-at value '{value}'. "
|
||||
"Use ISO format, e.g. 2025-12-31 or 2025-12-31T23:59:00.[/red]"
|
||||
)
|
||||
raise typer.Exit(1)
|
||||
return dt.isoformat()
|
||||
|
||||
|
||||
def _print_share_details(data: dict) -> None:
|
||||
"""Print a single share's fields in the snapshot-style detail layout."""
|
||||
console.print(f" Token: {data.get('token', 'unknown')}")
|
||||
console.print(f" URL: [blue underline]{data.get('share_url', '-')}[/blue underline]")
|
||||
console.print(f" Project: {data.get('project_name', '-')}")
|
||||
console.print(f" Note: {data.get('note_permalink', '-')}")
|
||||
console.print(f" Enabled: {'yes' if data.get('enabled', False) else 'no'}")
|
||||
console.print(f" Expires: {_format_timestamp(data.get('expires_at'))}")
|
||||
console.print(f" Views: {data.get('view_count', 0)}")
|
||||
console.print(f" Created: {_format_timestamp(data.get('created_at'))}")
|
||||
|
||||
|
||||
@share_app.command("create")
|
||||
def create(
|
||||
project: str = typer.Argument(
|
||||
...,
|
||||
help="Name of the project the note belongs to",
|
||||
),
|
||||
permalink: str = typer.Argument(
|
||||
...,
|
||||
help="Permalink of the note to share",
|
||||
),
|
||||
expires_at: Optional[str] = typer.Option(
|
||||
None,
|
||||
"--expires-at",
|
||||
"-e",
|
||||
help="Optional expiration date/time (ISO 8601, e.g. 2025-12-31)",
|
||||
),
|
||||
) -> None:
|
||||
"""Create a public share link for a note.
|
||||
|
||||
Examples:
|
||||
bm cloud share create my-project notes/my-idea
|
||||
bm cloud share create my-project notes/my-idea --expires-at 2025-12-31
|
||||
"""
|
||||
|
||||
async def _create():
|
||||
try:
|
||||
config_manager = ConfigManager()
|
||||
config = config_manager.config
|
||||
host_url = config.cloud_host.rstrip("/")
|
||||
|
||||
payload: dict = {
|
||||
"project_name": project,
|
||||
"note_permalink": permalink,
|
||||
}
|
||||
if expires_at is not None:
|
||||
payload["expires_at"] = _parse_expires_at(expires_at)
|
||||
|
||||
console.print("[blue]Creating share link...[/blue]")
|
||||
|
||||
response = await make_api_request(
|
||||
method="POST",
|
||||
url=f"{host_url}/api/shares",
|
||||
json_data=payload,
|
||||
)
|
||||
|
||||
data = response.json()
|
||||
|
||||
console.print("[green]Share link created successfully[/green]")
|
||||
_print_share_details(data)
|
||||
|
||||
except SubscriptionRequiredError as e:
|
||||
console.print("\n[red]Subscription Required[/red]\n")
|
||||
console.print(f"[yellow]{e.args[0]}[/yellow]\n")
|
||||
console.print(f"Subscribe at: [blue underline]{e.subscribe_url}[/blue underline]\n")
|
||||
raise typer.Exit(1)
|
||||
except CloudAPIError as e:
|
||||
if e.status_code == 404:
|
||||
console.print(f"[red]Note not found: {permalink} (project: {project})[/red]")
|
||||
else:
|
||||
console.print(f"[red]Failed to create share link: {e}[/red]")
|
||||
raise typer.Exit(1)
|
||||
except Exception as e:
|
||||
console.print(f"[red]Unexpected error: {e}[/red]")
|
||||
raise typer.Exit(1)
|
||||
|
||||
asyncio.run(_create())
|
||||
|
||||
|
||||
@share_app.command("list")
|
||||
def list_shares(
|
||||
project: Optional[str] = typer.Option(
|
||||
None,
|
||||
"--project",
|
||||
"-p",
|
||||
help="Filter shares by project name",
|
||||
),
|
||||
) -> None:
|
||||
"""List public share links.
|
||||
|
||||
Examples:
|
||||
bm cloud share list
|
||||
bm cloud share list --project my-project
|
||||
"""
|
||||
|
||||
async def _list():
|
||||
try:
|
||||
config_manager = ConfigManager()
|
||||
config = config_manager.config
|
||||
host_url = config.cloud_host.rstrip("/")
|
||||
|
||||
url = f"{host_url}/api/shares"
|
||||
if project:
|
||||
url += f"?project_name={project}"
|
||||
|
||||
console.print("[blue]Fetching share links...[/blue]")
|
||||
|
||||
response = await make_api_request(
|
||||
method="GET",
|
||||
url=url,
|
||||
)
|
||||
|
||||
data = response.json()
|
||||
shares = data.get("shares", [])
|
||||
total = data.get("total", len(shares))
|
||||
|
||||
if not shares:
|
||||
console.print("[yellow]No share links found[/yellow]")
|
||||
console.print(
|
||||
"\n[dim]Create a share with: bm cloud share create <project> <permalink>[/dim]"
|
||||
)
|
||||
return
|
||||
|
||||
table = Table(title=f"Public Shares ({total} total)")
|
||||
table.add_column("Token", style="cyan", no_wrap=True)
|
||||
table.add_column("Project", style="yellow")
|
||||
table.add_column("Note", style="white")
|
||||
table.add_column("Enabled", style="green")
|
||||
table.add_column("Expires", style="green")
|
||||
table.add_column("Views", style="magenta", justify="right")
|
||||
table.add_column("URL", style="blue", overflow="fold")
|
||||
|
||||
for share in shares:
|
||||
table.add_row(
|
||||
share.get("token", "unknown"),
|
||||
share.get("project_name", "-"),
|
||||
share.get("note_permalink", "-"),
|
||||
"yes" if share.get("enabled", False) else "no",
|
||||
_format_timestamp(share.get("expires_at")),
|
||||
str(share.get("view_count", 0)),
|
||||
share.get("share_url", "-"),
|
||||
)
|
||||
|
||||
console.print(table)
|
||||
|
||||
except SubscriptionRequiredError as e:
|
||||
console.print("\n[red]Subscription Required[/red]\n")
|
||||
console.print(f"[yellow]{e.args[0]}[/yellow]\n")
|
||||
console.print(f"Subscribe at: [blue underline]{e.subscribe_url}[/blue underline]\n")
|
||||
raise typer.Exit(1)
|
||||
except CloudAPIError as e:
|
||||
console.print(f"[red]Failed to list share links: {e}[/red]")
|
||||
raise typer.Exit(1)
|
||||
except Exception as e:
|
||||
console.print(f"[red]Unexpected error: {e}[/red]")
|
||||
raise typer.Exit(1)
|
||||
|
||||
asyncio.run(_list())
|
||||
|
||||
|
||||
@share_app.command("update")
|
||||
def update(
|
||||
token: str = typer.Argument(
|
||||
...,
|
||||
help="The token of the share to update",
|
||||
),
|
||||
enable: bool = typer.Option(
|
||||
False,
|
||||
"--enable",
|
||||
help="Enable the share link",
|
||||
),
|
||||
disable: bool = typer.Option(
|
||||
False,
|
||||
"--disable",
|
||||
help="Disable the share link without deleting it",
|
||||
),
|
||||
expires_at: Optional[str] = typer.Option(
|
||||
None,
|
||||
"--expires-at",
|
||||
"-e",
|
||||
help="New expiration date/time (ISO 8601). Use 'none' to clear it.",
|
||||
),
|
||||
) -> None:
|
||||
"""Update a share link: enable/disable it or change its expiration.
|
||||
|
||||
Examples:
|
||||
bm cloud share update abc123 --disable
|
||||
bm cloud share update abc123 --enable
|
||||
bm cloud share update abc123 --expires-at 2026-01-01
|
||||
bm cloud share update abc123 --expires-at none
|
||||
"""
|
||||
|
||||
async def _update():
|
||||
try:
|
||||
# --- Validate flags ---
|
||||
# Trigger: both toggles passed, or neither toggle and no expiry change.
|
||||
# Why: PATCH needs at least one concrete field, and enable/disable
|
||||
# conflict; reject up front so we don't send an empty/ambiguous body.
|
||||
if enable and disable:
|
||||
console.print("[red]Cannot use --enable and --disable together[/red]")
|
||||
raise typer.Exit(1)
|
||||
if not enable and not disable and expires_at is None:
|
||||
console.print(
|
||||
"[red]Nothing to update. Pass --enable, --disable, or --expires-at.[/red]"
|
||||
)
|
||||
raise typer.Exit(1)
|
||||
|
||||
payload: dict = {}
|
||||
if enable:
|
||||
payload["enabled"] = True
|
||||
if disable:
|
||||
payload["enabled"] = False
|
||||
if expires_at is not None:
|
||||
# "none" clears the expiration; anything else is parsed as a date.
|
||||
payload["expires_at"] = (
|
||||
None if expires_at.lower() == "none" else _parse_expires_at(expires_at)
|
||||
)
|
||||
|
||||
config_manager = ConfigManager()
|
||||
config = config_manager.config
|
||||
host_url = config.cloud_host.rstrip("/")
|
||||
|
||||
console.print("[blue]Updating share link...[/blue]")
|
||||
|
||||
response = await make_api_request(
|
||||
method="PATCH",
|
||||
url=f"{host_url}/api/shares/{token}",
|
||||
json_data=payload,
|
||||
)
|
||||
|
||||
data = response.json()
|
||||
|
||||
console.print("[green]Share link updated successfully[/green]")
|
||||
_print_share_details(data)
|
||||
|
||||
except typer.Exit:
|
||||
raise
|
||||
except SubscriptionRequiredError as e:
|
||||
console.print("\n[red]Subscription Required[/red]\n")
|
||||
console.print(f"[yellow]{e.args[0]}[/yellow]\n")
|
||||
console.print(f"Subscribe at: [blue underline]{e.subscribe_url}[/blue underline]\n")
|
||||
raise typer.Exit(1)
|
||||
except CloudAPIError as e:
|
||||
if e.status_code == 404:
|
||||
console.print(f"[red]Share not found: {token}[/red]")
|
||||
else:
|
||||
console.print(f"[red]Failed to update share link: {e}[/red]")
|
||||
raise typer.Exit(1)
|
||||
except Exception as e:
|
||||
console.print(f"[red]Unexpected error: {e}[/red]")
|
||||
raise typer.Exit(1)
|
||||
|
||||
asyncio.run(_update())
|
||||
|
||||
|
||||
@share_app.command("revoke")
|
||||
def revoke(
|
||||
token: str = typer.Argument(
|
||||
...,
|
||||
help="The token of the share to revoke",
|
||||
),
|
||||
force: bool = typer.Option(
|
||||
False,
|
||||
"--force",
|
||||
"-f",
|
||||
help="Skip confirmation prompt",
|
||||
),
|
||||
) -> None:
|
||||
"""Revoke (delete) a public share link.
|
||||
|
||||
Examples:
|
||||
bm cloud share revoke abc123
|
||||
bm cloud share revoke abc123 --force
|
||||
"""
|
||||
|
||||
async def _revoke():
|
||||
try:
|
||||
config_manager = ConfigManager()
|
||||
config = config_manager.config
|
||||
host_url = config.cloud_host.rstrip("/")
|
||||
|
||||
if not force:
|
||||
confirmed = typer.confirm(f"Are you sure you want to revoke share '{token}'?")
|
||||
if not confirmed:
|
||||
console.print("[yellow]Revocation cancelled[/yellow]")
|
||||
raise typer.Exit(0)
|
||||
|
||||
console.print("[blue]Revoking share link...[/blue]")
|
||||
|
||||
await make_api_request(
|
||||
method="DELETE",
|
||||
url=f"{host_url}/api/shares/{token}",
|
||||
)
|
||||
|
||||
console.print(f"[green]Share {token} revoked successfully[/green]")
|
||||
|
||||
except typer.Exit:
|
||||
raise
|
||||
except SubscriptionRequiredError as e:
|
||||
console.print("\n[red]Subscription Required[/red]\n")
|
||||
console.print(f"[yellow]{e.args[0]}[/yellow]\n")
|
||||
console.print(f"Subscribe at: [blue underline]{e.subscribe_url}[/blue underline]\n")
|
||||
raise typer.Exit(1)
|
||||
except CloudAPIError as e:
|
||||
if e.status_code == 404:
|
||||
console.print(f"[red]Share not found: {token}[/red]")
|
||||
else:
|
||||
console.print(f"[red]Failed to revoke share link: {e}[/red]")
|
||||
raise typer.Exit(1)
|
||||
except Exception as e:
|
||||
console.print(f"[red]Unexpected error: {e}[/red]")
|
||||
raise typer.Exit(1)
|
||||
|
||||
asyncio.run(_revoke())
|
||||
@@ -0,0 +1,595 @@
|
||||
"""Tests for cloud share CLI commands.
|
||||
|
||||
Issue #880: Tests for share create, list, update, revoke commands that surface
|
||||
the cloud /api/shares endpoints.
|
||||
"""
|
||||
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import httpx
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from basic_memory.cli.app import app
|
||||
from basic_memory.cli.commands.cloud.api_client import (
|
||||
CloudAPIError,
|
||||
SubscriptionRequiredError,
|
||||
)
|
||||
|
||||
SHARE_RESPONSE = {
|
||||
"id": "11111111-1111-1111-1111-111111111111",
|
||||
"token": "abc123",
|
||||
"project_name": "my-project",
|
||||
"note_permalink": "notes/my-idea",
|
||||
"note_external_id": "ext-1",
|
||||
"enabled": True,
|
||||
"expires_at": None,
|
||||
"share_url": "https://share.example.com/abc123",
|
||||
"view_count": 0,
|
||||
"last_viewed_at": None,
|
||||
"created_at": "2025-01-18T12:00:00Z",
|
||||
}
|
||||
|
||||
|
||||
def _mock_config_manager():
|
||||
mock_config = Mock()
|
||||
mock_config.cloud_host = "https://cloud.example.com"
|
||||
mock_config_manager = Mock()
|
||||
mock_config_manager.config = mock_config
|
||||
return mock_config_manager
|
||||
|
||||
|
||||
class TestShareCreateCommand:
|
||||
"""Tests for 'bm cloud share create' command."""
|
||||
|
||||
def test_create_share_success(self):
|
||||
runner = CliRunner()
|
||||
|
||||
mock_response = Mock(spec=httpx.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = SHARE_RESPONSE
|
||||
|
||||
captured = {}
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
captured["json_data"] = kwargs.get("json_data")
|
||||
return mock_response
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(
|
||||
app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "Share link created successfully" in result.stdout
|
||||
assert "abc123" in result.stdout
|
||||
assert "https://share.example.com/abc123" in result.stdout
|
||||
# Payload should match the cloud CreateShareRequest contract.
|
||||
assert captured["json_data"] == {
|
||||
"project_name": "my-project",
|
||||
"note_permalink": "notes/my-idea",
|
||||
}
|
||||
|
||||
def test_create_share_with_expires_at(self):
|
||||
runner = CliRunner()
|
||||
|
||||
mock_response = Mock(spec=httpx.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = SHARE_RESPONSE
|
||||
|
||||
captured = {}
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
captured["json_data"] = kwargs.get("json_data")
|
||||
return mock_response
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(
|
||||
app,
|
||||
[
|
||||
"cloud",
|
||||
"share",
|
||||
"create",
|
||||
"my-project",
|
||||
"notes/my-idea",
|
||||
"--expires-at",
|
||||
"2025-12-31",
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert captured["json_data"]["expires_at"].startswith("2025-12-31")
|
||||
|
||||
def test_create_share_invalid_expires_at(self):
|
||||
runner = CliRunner()
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs): # pragma: no cover
|
||||
raise AssertionError("API should not be called on invalid input")
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(
|
||||
app,
|
||||
[
|
||||
"cloud",
|
||||
"share",
|
||||
"create",
|
||||
"my-project",
|
||||
"notes/my-idea",
|
||||
"--expires-at",
|
||||
"not-a-date",
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "Invalid --expires-at" in result.stdout
|
||||
|
||||
def test_create_share_note_not_found(self):
|
||||
runner = CliRunner()
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
raise CloudAPIError("Not found", status_code=404)
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(
|
||||
app, ["cloud", "share", "create", "my-project", "notes/missing"]
|
||||
)
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "Note not found" in result.stdout
|
||||
|
||||
def test_create_share_subscription_required(self):
|
||||
runner = CliRunner()
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
raise SubscriptionRequiredError(
|
||||
message="Active subscription required",
|
||||
subscribe_url="https://basicmemory.com/subscribe",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(
|
||||
app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
|
||||
)
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "Subscription Required" in result.stdout
|
||||
|
||||
def test_create_share_api_error(self):
|
||||
runner = CliRunner()
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
raise CloudAPIError("Server error", status_code=500)
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(
|
||||
app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
|
||||
)
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "Failed to create share link" in result.stdout
|
||||
|
||||
|
||||
class TestShareListCommand:
|
||||
"""Tests for 'bm cloud share list' command."""
|
||||
|
||||
def test_list_shares_success(self):
|
||||
# Wide terminal so the rich table doesn't truncate cell contents.
|
||||
runner = CliRunner(env={"COLUMNS": "200"})
|
||||
|
||||
mock_response = Mock(spec=httpx.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {
|
||||
"shares": [
|
||||
SHARE_RESPONSE,
|
||||
{
|
||||
**SHARE_RESPONSE,
|
||||
"token": "def456",
|
||||
"note_permalink": "notes/second",
|
||||
"enabled": False,
|
||||
"expires_at": "2025-12-31T00:00:00Z",
|
||||
"view_count": 7,
|
||||
},
|
||||
],
|
||||
"total": 2,
|
||||
}
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
return mock_response
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(app, ["cloud", "share", "list"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "abc123" in result.stdout
|
||||
assert "def456" in result.stdout
|
||||
assert "notes/second" in result.stdout
|
||||
|
||||
def test_list_shares_empty(self):
|
||||
runner = CliRunner()
|
||||
|
||||
mock_response = Mock(spec=httpx.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {"shares": [], "total": 0}
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
return mock_response
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(app, ["cloud", "share", "list"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "No share links found" in result.stdout
|
||||
|
||||
def test_list_shares_with_project_filter(self):
|
||||
runner = CliRunner()
|
||||
|
||||
mock_response = Mock(spec=httpx.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {"shares": [SHARE_RESPONSE], "total": 1}
|
||||
|
||||
captured = {}
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
captured["url"] = kwargs.get("url", args[1] if len(args) > 1 else "")
|
||||
return mock_response
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(app, ["cloud", "share", "list", "--project", "my-project"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "project_name=my-project" in captured["url"]
|
||||
|
||||
def test_list_shares_api_error(self):
|
||||
runner = CliRunner()
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
raise CloudAPIError("Server error", status_code=500)
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(app, ["cloud", "share", "list"])
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "Failed to list share links" in result.stdout
|
||||
|
||||
|
||||
class TestShareUpdateCommand:
|
||||
"""Tests for 'bm cloud share update' command."""
|
||||
|
||||
def test_update_disable(self):
|
||||
runner = CliRunner()
|
||||
|
||||
mock_response = Mock(spec=httpx.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {**SHARE_RESPONSE, "enabled": False}
|
||||
|
||||
captured = {}
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
captured["json_data"] = kwargs.get("json_data")
|
||||
captured["method"] = kwargs.get("method")
|
||||
return mock_response
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(app, ["cloud", "share", "update", "abc123", "--disable"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "updated successfully" in result.stdout
|
||||
assert captured["method"] == "PATCH"
|
||||
assert captured["json_data"] == {"enabled": False}
|
||||
|
||||
def test_update_enable(self):
|
||||
runner = CliRunner()
|
||||
|
||||
mock_response = Mock(spec=httpx.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = SHARE_RESPONSE
|
||||
|
||||
captured = {}
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
captured["json_data"] = kwargs.get("json_data")
|
||||
return mock_response
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(app, ["cloud", "share", "update", "abc123", "--enable"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert captured["json_data"] == {"enabled": True}
|
||||
|
||||
def test_update_expires_at(self):
|
||||
runner = CliRunner()
|
||||
|
||||
mock_response = Mock(spec=httpx.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = SHARE_RESPONSE
|
||||
|
||||
captured = {}
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
captured["json_data"] = kwargs.get("json_data")
|
||||
return mock_response
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(
|
||||
app,
|
||||
["cloud", "share", "update", "abc123", "--expires-at", "2026-01-01"],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert captured["json_data"]["expires_at"].startswith("2026-01-01")
|
||||
|
||||
def test_update_clear_expires_at(self):
|
||||
runner = CliRunner()
|
||||
|
||||
mock_response = Mock(spec=httpx.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = SHARE_RESPONSE
|
||||
|
||||
captured = {}
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
captured["json_data"] = kwargs.get("json_data")
|
||||
return mock_response
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(
|
||||
app, ["cloud", "share", "update", "abc123", "--expires-at", "none"]
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert captured["json_data"] == {"expires_at": None}
|
||||
|
||||
def test_update_enable_and_disable_conflict(self):
|
||||
runner = CliRunner()
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs): # pragma: no cover
|
||||
raise AssertionError("API should not be called on conflicting flags")
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(
|
||||
app,
|
||||
["cloud", "share", "update", "abc123", "--enable", "--disable"],
|
||||
)
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "Cannot use --enable and --disable together" in result.stdout
|
||||
|
||||
def test_update_nothing_to_change(self):
|
||||
runner = CliRunner()
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs): # pragma: no cover
|
||||
raise AssertionError("API should not be called with empty update")
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(app, ["cloud", "share", "update", "abc123"])
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "Nothing to update" in result.stdout
|
||||
|
||||
def test_update_not_found(self):
|
||||
runner = CliRunner()
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
raise CloudAPIError("Not found", status_code=404)
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(app, ["cloud", "share", "update", "missing", "--disable"])
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "Share not found" in result.stdout
|
||||
|
||||
|
||||
class TestShareRevokeCommand:
|
||||
"""Tests for 'bm cloud share revoke' command."""
|
||||
|
||||
def test_revoke_success_with_force(self):
|
||||
runner = CliRunner()
|
||||
|
||||
mock_response = Mock(spec=httpx.Response)
|
||||
mock_response.status_code = 204
|
||||
mock_response.json.return_value = {}
|
||||
|
||||
captured = {}
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
captured["method"] = kwargs.get("method")
|
||||
captured["url"] = kwargs.get("url")
|
||||
return mock_response
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(app, ["cloud", "share", "revoke", "abc123", "--force"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "revoked successfully" in result.stdout
|
||||
assert captured["method"] == "DELETE"
|
||||
assert captured["url"].endswith("/api/shares/abc123")
|
||||
|
||||
def test_revoke_cancelled(self):
|
||||
runner = CliRunner()
|
||||
|
||||
call_count = 0
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
return Mock(spec=httpx.Response)
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(app, ["cloud", "share", "revoke", "abc123"], input="n\n")
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "cancelled" in result.stdout
|
||||
assert call_count == 0
|
||||
|
||||
def test_revoke_not_found(self):
|
||||
runner = CliRunner()
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
raise CloudAPIError("Not found", status_code=404)
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(app, ["cloud", "share", "revoke", "missing", "--force"])
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "Share not found" in result.stdout
|
||||
|
||||
def test_revoke_subscription_required(self):
|
||||
runner = CliRunner()
|
||||
|
||||
async def mock_make_api_request(*args, **kwargs):
|
||||
raise SubscriptionRequiredError(
|
||||
message="Active subscription required",
|
||||
subscribe_url="https://basicmemory.com/subscribe",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.make_api_request",
|
||||
side_effect=mock_make_api_request,
|
||||
):
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.shares.ConfigManager",
|
||||
return_value=_mock_config_manager(),
|
||||
):
|
||||
result = runner.invoke(app, ["cloud", "share", "revoke", "abc123", "--force"])
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "Subscription Required" in result.stdout
|
||||
Reference in New Issue
Block a user