Compare commits

..

4 Commits

Author SHA1 Message Date
Drew Cain 440a2aa923 fix(cli): resolve workspace slugs and names for bm cloud share --workspace
Live prod QA found `bm cloud share ... --workspace <slug>` failed with an
opaque 400: resolve_configured_workspace returns the explicit --workspace
value verbatim, but the cloud's X-Workspace-ID resolver only accepts a
workspace/tenant UUID. Users see slugs and display names (list-workspaces
output, memory:// URLs), so the natural input never routed.

The share commands now resolve the workspace header client-side: a UUID is
forwarded verbatim (covers per-project config workspace_id and the default
chain, zero extra API calls), while any other value is treated as a human
identifier and mapped to the tenant UUID via a single get_available_workspaces
lookup, matching with slug > tenant_id > name precedence (mirroring #979).
Ambiguous and unknown identifiers fail fast with errors that name the
candidate / available workspace slugs, and the list command now re-raises
typer.Exit ahead of its broad handler so the resolution errors aren't
re-wrapped as "Unexpected error".

Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: Drew Cain <groksrc@gmail.com>
2026-06-11 15:57:29 -05:00
Drew Cain a8e452d9b4 fix(cli): route share commands to the configured workspace and encode project filter
The `bm cloud share` commands authenticated but never sent the
X-Workspace-ID header that sibling cloud calls (cloud_utils._workspace_headers)
use for workspace scoping. The cloud /api/shares endpoints resolve the target
tenant from that header (resolve_workspace in basic-memory-cloud deps.py), so
share create/list/update/revoke for a team-workspace project were evaluated
against the caller's default tenant.

Resolve the workspace via resolve_configured_workspace (explicit --workspace >
project's configured workspace_id > global default) and pass X-Workspace-ID on
all four commands, mirroring the established cloud command pattern. Adds a
--workspace option to each, matching `bm cloud pull/push`.

Also URL-encode the list `--project` filter with urllib.parse.urlencode so
project names containing query-reserved characters (&, +, #, spaces) reach the
server as a single faithful value instead of splitting into stray query params.

Extends the mocked tests to assert the header is built/routed and that a
project name with special characters is percent-encoded.

Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: Drew Cain <groksrc@gmail.com>
2026-06-11 09:39:32 -05:00
Drew Cain 22901c366d fix(cli): clean error on invalid share --expires-at and drop dead schemas
Address review findings on `bm cloud share`:

- `create` re-wrapped the `typer.Exit(1)` from `_parse_expires_at()` as a
  spurious "Unexpected error: 1" trailing line because the broad
  `except Exception` caught it (typer.Exit subclasses Exception). Parse and
  validate --expires-at before entering the try block so the error surfaces
  as a single clean message, and add the `except typer.Exit: raise` guard for
  parity with `update`/`revoke`.
- Strengthen test_create_share_invalid_expires_at to assert
  "Unexpected error" is absent, which was masking the bug.
- Drop unused PublicShareResponse/PublicShareListResponse schemas; responses
  are parsed as raw dicts, so the schemas were dead code.

Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: Drew Cain <groksrc@gmail.com>
2026-06-11 01:21:52 -05:00
Drew Cain 205196b621 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>
2026-06-11 01:12:04 -05:00
5 changed files with 1534 additions and 2059 deletions
-1333
View File
File diff suppressed because it is too large Load Diff
-726
View File
@@ -1,726 +0,0 @@
#!/usr/bin/env python3
"""Generate MCP tool reference documentation from tool docstrings.
Introspects all MCP tool definitions in src/basic_memory/mcp/tools/,
extracts names, descriptions, parameters, and docstrings, then emits
a single deterministic markdown reference to docs/mcp-tools.md.
Usage:
uv run scripts/generate_tool_docs.py
# Verify idempotency (diff should be empty):
uv run scripts/generate_tool_docs.py && uv run scripts/generate_tool_docs.py
git diff docs/mcp-tools.md
"""
from __future__ import annotations
import ast
import inspect
import re
import sys
from pathlib import Path
# ---------------------------------------------------------------------------
# Paths
# ---------------------------------------------------------------------------
ROOT = Path(__file__).resolve().parents[1]
TOOLS_DIR = ROOT / "src" / "basic_memory" / "mcp" / "tools"
OUT_FILE = ROOT / "docs" / "mcp-tools.md"
# Only files that contain public MCP tools (skip helpers/internals).
# NOTE: ui_sdk.py is intentionally omitted — its two tools are commented out
# in src/basic_memory/mcp/__init__.py and therefore not public.
TOOL_FILES = [
"build_context.py",
"canvas.py",
"chatgpt_tools.py",
"cloud_info.py",
"delete_note.py",
"edit_note.py",
"list_directory.py",
"move_note.py",
"project_management.py",
"read_content.py",
"read_note.py",
"recent_activity.py",
"release_notes.py",
"schema.py",
"search.py",
"view_note.py",
"workspaces.py",
"write_note.py",
]
# ---------------------------------------------------------------------------
# AST-based extraction (no import side-effects)
# ---------------------------------------------------------------------------
def _clean_docstring(raw: str | None) -> str:
"""Dedent and strip a raw docstring."""
if not raw:
return ""
return inspect.cleandoc(raw)
def _get_default_repr(node: ast.expr | None) -> str:
"""Return a short string representation for a default-value AST node."""
if node is None:
return ""
try:
return ast.unparse(node)
except Exception:
return "..."
def _unwrap_annotated(text: str) -> str:
"""Bracket-aware unwrap of ``Annotated[X, ...]`` → ``X``.
The naïve regex ``Annotated\\[([^,\\]]+),.*?\\]`` breaks on types whose
first argument itself contains brackets or commas, e.g.
``Annotated[Dict[str, Any] | None, ...]`` or
``Annotated[List[str] | None, BeforeValidator(...), ...]``.
This function scans character-by-character to find the first
*top-level* comma that separates the type from the metadata args,
then discards everything from that comma to the matching closing ``]``.
"""
prefix = "Annotated["
result = []
i = 0
while i < len(text):
if text[i:].startswith(prefix):
# Walk past "Annotated[" and collect the first top-level argument.
j = i + len(prefix)
depth = 0
start = j
while j < len(text):
ch = text[j]
if ch in "([{":
depth += 1
elif ch in ")]}":
if depth == 0:
# Closing bracket of Annotated[...] with no comma found
# (shouldn't happen in valid annotations, but be safe).
break
depth -= 1
elif ch == "," and depth == 0:
# Found the separator between the type arg and metadata.
break
j += 1
inner_type = text[start:j]
result.append(inner_type)
# Skip past the rest of the Annotated[...] construct.
# We need to find the matching ']' for the original 'Annotated['.
depth = 0
while j < len(text):
ch = text[j]
if ch in "([{":
depth += 1
elif ch == "]" and depth == 0:
j += 1 # consume the closing ']'
break
elif ch in ")]}":
depth -= 1
j += 1
i = j
else:
result.append(text[i])
i += 1
return "".join(result)
def _annotation_repr(node: ast.expr | None) -> str:
"""Return a readable string for a type-annotation AST node."""
if node is None:
return ""
try:
text = ast.unparse(node)
except Exception:
return ""
# Unwrap Annotated[X, ...] → X (keeps output readable).
# Use a bracket-aware unwrap so types like Dict[str, Any] | None are
# preserved correctly instead of being truncated at the first comma.
text = _unwrap_annotated(text)
return text
_SENTINEL = object() # returned when decorator IS @mcp.tool but has no description string
def _mcp_tool_description(decorator: ast.expr) -> str | None | object:
"""Extract the ``description=`` keyword from an ``@mcp.tool(...)`` decorator call.
Returns:
- A string when the decorator carries ``description="..."``
- ``_SENTINEL`` when the decorator IS an mcp.tool call but has no description
- ``None`` when the decorator is not an mcp.tool call at all
This distinction lets callers treat mcp.tool-decorated functions without a
decorator description string as still valid tools (description comes from the
docstring in that case).
"""
if not isinstance(decorator, ast.Call):
return None
func = decorator.func
# Accept both ``mcp.tool(...)`` and ``tool(...)``
if not (
(isinstance(func, ast.Attribute) and func.attr == "tool")
or (isinstance(func, ast.Name) and func.id == "tool")
):
return None
for kw in decorator.keywords:
if kw.arg == "description" and isinstance(kw.value, ast.Constant):
return kw.value.value
# Also accept a positional string as the tool name (no description)
# e.g. @mcp.tool("list_memory_projects", annotations={...})
# In this case, the description is not in the decorator; fall through to docstring.
return _SENTINEL
class ToolInfo:
"""Holds extracted metadata for a single MCP tool."""
__slots__ = (
"name",
"decorator_description",
"docstring",
"params",
"source_file",
)
def __init__(
self,
name: str,
decorator_description: str,
docstring: str,
params: list[dict[str, str]],
source_file: str,
) -> None:
self.name = name
self.decorator_description = decorator_description
self.docstring = docstring
self.params = params
self.source_file = source_file
# Parameters that are MCP/framework plumbing — not useful to document.
_SKIP_PARAMS = frozenset({"context", "self", "cls"})
def extract_tools_from_file(path: Path) -> list[ToolInfo]:
"""Parse *path* with AST and return a list of ToolInfo for every @mcp.tool function."""
source = path.read_text(encoding="utf-8")
try:
tree = ast.parse(source, filename=str(path))
except SyntaxError as exc:
print(f"WARNING: could not parse {path}: {exc}", file=sys.stderr)
return []
tools: list[ToolInfo] = []
for node in ast.walk(tree):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
# Only process functions that have an @mcp.tool (or @tool) decorator.
# decorator_desc is:
# - a non-empty string when found in the decorator
# - _SENTINEL when the decorator is mcp.tool but carries no description
# - None when no mcp.tool decorator is present
decorator_desc: str | object | None = None
for dec in node.decorator_list:
result = _mcp_tool_description(dec)
if result is not None:
decorator_desc = result
break
if decorator_desc is None:
continue
# When the decorator doesn't carry a description string, fall back to
# the first sentence of the docstring (populated below).
decorator_has_description = isinstance(decorator_desc, str)
docstring = _clean_docstring(ast.get_docstring(node))
# --- Parameters ---
params: list[dict[str, str]] = []
args = node.args
# Defaults are right-aligned against the arg list
n_defaults = len(args.defaults)
n_args = len(args.args)
defaults_padded = [None] * (n_args - n_defaults) + list(args.defaults) # type: ignore[list-item]
kw_defaults = args.kw_defaults # may contain None for "no default"
for i, arg in enumerate(args.args):
if arg.arg in _SKIP_PARAMS:
continue
default = defaults_padded[i]
params.append(
{
"name": arg.arg,
"type": _annotation_repr(arg.annotation),
"default": _get_default_repr(default) if default is not None else "",
}
)
for i, arg in enumerate(args.kwonlyargs):
if arg.arg in _SKIP_PARAMS:
continue
default = kw_defaults[i] if i < len(kw_defaults) else None
params.append(
{
"name": arg.arg,
"type": _annotation_repr(arg.annotation),
"default": _get_default_repr(default) if default is not None else "",
}
)
# Build the short one-liner description:
# - prefer the explicit decorator description=
# - fall back to the first non-blank line of the docstring
if decorator_has_description:
short_desc = decorator_desc.strip() # type: ignore[union-attr]
else:
first_line = next((ln.strip() for ln in docstring.splitlines() if ln.strip()), "")
short_desc = first_line
tools.append(
ToolInfo(
name=node.name,
decorator_description=short_desc,
docstring=docstring,
params=params,
source_file=path.name,
)
)
return tools
# ---------------------------------------------------------------------------
# Markdown rendering
# ---------------------------------------------------------------------------
def _escape_table_cell(text: str) -> str:
"""Escape ``|`` characters so they do not break GitHub-flavoured markdown tables."""
return text.replace("|", "\\|")
def _params_table(
params: list[dict[str, str]], arg_descriptions: dict[str, str] | None = None
) -> str:
"""Render parameters as a markdown table.
``arg_descriptions`` maps parameter name → one-line description extracted
from the docstring ``Args:`` block.
"""
if not params:
return ""
arg_descriptions = arg_descriptions or {}
rows = [
"| Parameter | Type | Default | Description |",
"|-----------|------|---------|-------------|",
]
for p in params:
name = f"`{p['name']}`"
# Pipe characters inside a table cell break GitHub's renderer; escape them.
raw_type = p["type"]
typ = f"`{_escape_table_cell(raw_type)}`" if raw_type else ""
default_val = p["default"]
default = f"`{_escape_table_cell(default_val)}`" if default_val else "*(required)*"
desc = _escape_table_cell(arg_descriptions.get(p["name"], ""))
rows.append(f"| {name} | {typ} | {default} | {desc} |")
return "\n".join(rows)
def _extract_examples_section(docstring: str) -> tuple[str, str]:
"""Split docstring into (body_without_examples, examples_block).
Looks for an 'Examples:' section (with or without leading '#') and
returns everything before it as the body, and everything from the
Examples header to the next top-level section (e.g. 'Raises:',
'Returns:') as examples. Trailing sections after Examples are
re-appended to the body so they are not silently dropped.
"""
# Match headings like "Examples:", "# Examples", "## Examples", etc.
example_pattern = re.compile(r"^(#{1,3}\s*)?Examples:?\s*$", re.MULTILINE | re.IGNORECASE)
match = example_pattern.search(docstring)
if not match:
return docstring.strip(), ""
body_before = docstring[: match.start()].strip()
after_header = docstring[match.end() :]
# Find the next top-level section that is NOT indented (Raises:, Returns:, etc.)
# so we do not swallow it into the examples block.
next_section = re.search(r"\n(?=[A-Z][^\n]*:$)", after_header, re.MULTILINE)
if next_section:
examples_raw = after_header[: next_section.start()].strip()
trailing = after_header[next_section.start() :].strip()
# Re-attach the trailing section to the body.
body = (body_before + "\n\n" + trailing).strip() if trailing else body_before
else:
examples_raw = after_header.strip()
body = body_before
return body, examples_raw
def _parse_args_block(args_text: str) -> dict[str, str]:
"""Parse a docstring ``Args:`` block into a ``{param_name: description}`` dict.
Handles the standard Google-style format with optional leading indentation::
param_name: First line of description.
Continuation lines are more deeply indented.
Works with both un-indented and uniformly-indented blocks (e.g. when the
raw docstring is already cleandoc'd but the Args entries still have a
consistent leading indent from the original source indentation).
Returns a dict mapping each parameter name to its single-line summary
(first line only, stripped) for use in the parameters table.
"""
result: dict[str, str] = {}
# Determine the base indent level: the minimum indent of non-empty lines.
# Parameter entries sit at this level; continuation lines are deeper.
non_empty = [ln for ln in args_text.splitlines() if ln.strip()]
if not non_empty:
return result
base_indent = min(len(ln) - len(ln.lstrip()) for ln in non_empty)
current_name: str | None = None
current_lines: list[str] = []
for line in args_text.splitlines():
if not line.strip():
continue
indent = len(line) - len(line.lstrip())
stripped = line.strip()
# A new parameter entry sits at the base indent level with "name: ..."
if indent == base_indent:
top_level = re.match(r"(\w+)\s*:(.*)", stripped)
if top_level:
if current_name is not None:
result[current_name] = " ".join(current_lines).strip()
current_name = top_level.group(1)
rest = top_level.group(2).strip()
current_lines = [rest] if rest else []
continue
# Continuation line — only capture the first continuation line to
# keep descriptions terse enough for a table cell.
if current_name is not None and not current_lines:
current_lines.append(stripped)
if current_name is not None:
result[current_name] = " ".join(current_lines).strip()
return result
def _format_args_section(docstring: str) -> tuple[str, dict[str, str]]:
"""Remove the 'Args:' block from a docstring and return (cleaned_body, arg_descriptions).
``arg_descriptions`` maps each parameter name to its one-line description
so callers can populate the Description column of the params table.
"""
pattern = re.compile(r"^Args:\s*$", re.MULTILINE)
match = pattern.search(docstring)
if not match:
return docstring.strip(), {}
before = docstring[: match.start()].strip()
after_start = match.end()
# The Args block ends at the next top-level section (line that starts
# without indentation and ends with ':'), or at end-of-string.
next_section = re.search(r"\n(?=[A-Z][^\n:]*:$)", docstring[after_start:], re.MULTILINE)
if next_section:
# Use lstrip("\n") not strip() so that leading spaces (indent) on the
# first arg line are preserved for _parse_args_block's indent detection.
args_text = docstring[after_start : after_start + next_section.start()].lstrip("\n")
remainder = docstring[after_start + next_section.start() :].strip()
body = (before + "\n\n" + remainder).strip()
else:
args_text = docstring[after_start:].lstrip("\n")
body = before.strip()
return body, _parse_args_block(args_text)
def render_tool_section(tool: ToolInfo, anchor_suffix: str = "") -> str:
"""Return a markdown section for a single tool.
``anchor_suffix`` is appended to the HTML id anchor injected before the
heading when the tool name would collide with a group-section anchor
(e.g. the ``search`` tool inside the ``## Search`` group).
"""
lines: list[str] = []
if anchor_suffix:
# Inject an explicit anchor so the TOC link resolves to the tool
# section rather than the same-named group header.
lines.append(f'<a id="{tool.name.replace("_", "-")}{anchor_suffix}"></a>')
lines.append("")
lines.append(f"### `{tool.name}`")
lines.append("")
# One-liner description from the decorator (most concise)
lines.append(tool.decorator_description)
lines.append("")
# Full docstring body (strip Args: and Examples: sub-sections which get
# their own formatting below)
doc = tool.docstring
doc, arg_descriptions = _format_args_section(doc)
doc, examples_text = _extract_examples_section(doc)
# Avoid duplicating the decorator description: if the docstring's first
# non-blank line is essentially the same sentence, drop it.
if doc:
doc_lines = doc.splitlines()
first_nonempty_idx = next(
(i for i, ln in enumerate(doc_lines) if ln.strip()),
None,
)
if first_nonempty_idx is not None:
first_line = doc_lines[first_nonempty_idx].strip()
# Normalise both sides for comparison (lowercase, strip punctuation)
norm = lambda s: re.sub(r"[^a-z0-9]", "", s.lower()) # noqa: E731
if norm(first_line) == norm(tool.decorator_description):
doc_lines.pop(first_nonempty_idx)
doc = "\n".join(doc_lines).strip()
# Demote any `###` sub-headers in the body to `####` so they do not
# collide with tool-name headers in GitHub's outline sidebar.
doc = re.sub(r"^### ", "#### ", doc, flags=re.MULTILINE)
if doc:
lines.append(doc)
lines.append("")
# Parameters table
if tool.params:
lines.append("**Parameters**")
lines.append("")
lines.append(_params_table(tool.params, arg_descriptions))
lines.append("")
# Examples block
if examples_text:
lines.append("**Examples**")
lines.append("")
# Wrap in a python code fence if not already fenced
if "```" not in examples_text:
lines.append("```python")
lines.append(examples_text)
lines.append("```")
else:
lines.append(examples_text)
lines.append("")
lines.append(f"*Source: `src/basic_memory/mcp/tools/{tool.source_file}`*")
lines.append("")
return "\n".join(lines)
# ---------------------------------------------------------------------------
# Grouping
# ---------------------------------------------------------------------------
# Stable, hand-curated grouping of tools into logical categories.
# Tools not listed here fall into "Other Tools".
TOOL_GROUPS: dict[str, list[str]] = {
"Note Management": [
"write_note",
"read_note",
"view_note",
"edit_note",
"move_note",
"delete_note",
],
"Reading & Navigation": [
"read_content",
"build_context",
"recent_activity",
"list_directory",
],
"Search": [
"search_notes",
"search",
"fetch",
],
"Project & Workspace Management": [
"list_memory_projects",
"create_memory_project",
"delete_project",
"list_workspaces",
],
"Schema Tools": [
"schema_validate",
"schema_infer",
"schema_diff",
],
"Visualization": [
"canvas",
],
"Info & Utilities": [
"cloud_info",
"release_notes",
],
}
def group_tools(tools: list[ToolInfo]) -> dict[str, list[ToolInfo]]:
"""Return an ordered dict mapping group name → list of ToolInfo."""
by_name: dict[str, ToolInfo] = {t.name: t for t in tools}
result: dict[str, list[ToolInfo]] = {}
placed: set[str] = set()
for group, names in TOOL_GROUPS.items():
members = [by_name[n] for n in names if n in by_name]
if members:
result[group] = members
placed.update(n for n in names if n in by_name)
# Anything not in TOOL_GROUPS goes to "Other Tools", sorted alphabetically
other = sorted([t for t in tools if t.name not in placed], key=lambda t: t.name)
if other:
result["Other Tools"] = other
return result
# ---------------------------------------------------------------------------
# Top-level document builder
# ---------------------------------------------------------------------------
HEADER = """\
<!--
This file is AUTO-GENERATED. Do not edit it by hand.
To regenerate:
uv run scripts/generate_tool_docs.py
Source: scripts/generate_tool_docs.py
-->
# Basic Memory MCP Tool Reference
Complete reference for all MCP tools exposed by the Basic Memory server.
Tools are grouped by function. Parameters marked *(required)* have no default value.
> **Regenerating this file**: run `uv run scripts/generate_tool_docs.py` from the
> repository root. The output is deterministic; running it twice should produce
> an identical file (zero diff).
## Table of Contents
"""
def _tool_anchor(tool_name: str, group_anchors: set[str]) -> str:
"""Return a stable GitHub anchor for a tool heading.
If the plain ``tool_name`` (with ``_`` → ``-``) would collide with a
group-section anchor (e.g. the ``search`` tool inside the ``## Search``
group), append ``-tool`` to disambiguate.
"""
base = tool_name.replace("_", "-")
if base in group_anchors:
return base + "-tool"
return base
def build_toc(groups: dict[str, list[ToolInfo]]) -> str:
lines: list[str] = []
# Collect all group anchors first so tool anchors can avoid collisions.
group_anchors: set[str] = set()
for group in groups:
anchor = re.sub(r"[^\w\s-]", "", group.lower())
anchor = re.sub(r"\s+", "-", anchor.strip())
group_anchors.add(anchor)
for group, members in groups.items():
# GitHub-style anchor: lowercase, spaces → hyphens, drop punctuation
anchor = re.sub(r"[^\w\s-]", "", group.lower())
anchor = re.sub(r"\s+", "-", anchor.strip())
lines.append(f"- [{group}](#{anchor})")
for tool in members:
t_anchor = _tool_anchor(tool.name, group_anchors)
lines.append(f" - [`{tool.name}`](#{t_anchor})")
return "\n".join(lines)
def build_document(groups: dict[str, list[ToolInfo]]) -> str:
# Pre-compute group anchors so tool-section rendering can inject explicit
# HTML id attributes where tool names would collide with group headers.
group_anchors: set[str] = set()
for group in groups:
anchor = re.sub(r"[^\w\s-]", "", group.lower())
anchor = re.sub(r"\s+", "-", anchor.strip())
group_anchors.add(anchor)
parts: list[str] = [HEADER]
parts.append(build_toc(groups))
parts.append("\n\n---\n")
for group, members in groups.items():
parts.append(f"\n## {group}\n")
for tool in members:
base = tool.name.replace("_", "-")
suffix = "-tool" if base in group_anchors else ""
parts.append(render_tool_section(tool, anchor_suffix=suffix))
parts.append("---\n")
# Trailing newline
return "\n".join(parts) + "\n"
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main() -> int:
all_tools: list[ToolInfo] = []
for filename in TOOL_FILES:
path = TOOLS_DIR / filename
if not path.exists():
print(f"WARNING: {path} does not exist — skipping", file=sys.stderr)
continue
file_tools = extract_tools_from_file(path)
all_tools.extend(file_tools)
if not all_tools:
print("ERROR: no tools found — check TOOLS_DIR path", file=sys.stderr)
return 1
# Stable sort: group order is defined by TOOL_GROUPS; within groups, order
# is defined by TOOL_GROUPS list. Global sort here is just for determinism
# of "Other Tools".
groups = group_tools(all_tools)
document = build_document(groups)
OUT_FILE.parent.mkdir(parents=True, exist_ok=True)
OUT_FILE.write_text(document, encoding="utf-8")
total = sum(len(m) for m in groups.values())
print(
f"Generated {OUT_FILE.relative_to(ROOT)} ({total} tools, {len(OUT_FILE.read_text().splitlines())} lines)"
)
return 0
if __name__ == "__main__":
sys.exit(main())
@@ -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
@@ -0,0 +1,533 @@
"""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
from urllib.parse import urlencode
from uuid import UUID
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
from basic_memory.mcp.async_client import resolve_configured_workspace
from basic_memory.schemas.cloud import WorkspaceInfo
console = Console()
share_app = typer.Typer(help="Manage public share links for notes")
# Header the cloud uses to route a request to a specific tenant's workspace.
# Mirrors basic_memory.cli.commands.cloud.cloud_utils._workspace_headers: the
# cloud /api/shares endpoints resolve the workspace from X-Workspace-ID (see
# resolve_workspace in basic-memory-cloud deps.py), so without it a team
# workspace project would be evaluated against the caller's default tenant.
WORKSPACE_ID_HEADER = "X-Workspace-ID"
def _is_uuid(value: str) -> bool:
"""Return True when value parses as a UUID in any standard textual form."""
try:
UUID(value)
except ValueError:
return False
return True
def _match_workspace_identifier(
workspaces: list[WorkspaceInfo], identifier: str
) -> Optional[WorkspaceInfo]:
"""Match a human workspace identifier with slug > tenant_id > name precedence.
Mirrors project_context._match_workspace_identifier (PR #979): try the stable
slug first (case-insensitive), then the exact tenant_id, then the display name
(case-insensitive). The first tier that yields any match wins, so a display
name colliding with another workspace's slug never shadows the slug match.
"""
slug_matches = [ws for ws in workspaces if ws.slug.casefold() == identifier.casefold()]
if slug_matches:
return slug_matches[0] if len(slug_matches) == 1 else _ambiguous(slug_matches, identifier)
tenant_matches = [ws for ws in workspaces if ws.tenant_id == identifier]
if tenant_matches:
return tenant_matches[0]
name_matches = [ws for ws in workspaces if ws.name.casefold() == identifier.casefold()]
if name_matches:
return name_matches[0] if len(name_matches) == 1 else _ambiguous(name_matches, identifier)
return None
def _ambiguous(matches: list[WorkspaceInfo], identifier: str) -> WorkspaceInfo:
"""Fail with a clear, copyable error when an identifier matches >1 workspace.
Trigger: a display name (or slug, defensively) resolves to multiple workspaces.
Why: silently picking one would route a share to the wrong tenant.
Outcome: list candidate slugs/tenant_ids and exit non-zero so the user re-runs
with an unambiguous slug or tenant_id.
"""
candidates = "\n".join(f" - {ws.slug} (tenant_id: {ws.tenant_id})" for ws in matches)
console.print(
f"[red]Workspace '{identifier}' is ambiguous; it matches multiple workspaces.[/red]\n"
"[yellow]Re-run with a unique workspace slug or tenant_id:[/yellow]\n"
f"{candidates}"
)
raise typer.Exit(1)
async def _resolve_workspace_to_tenant_id(identifier: str) -> str:
"""Resolve a human workspace identifier (slug/name/tenant_id) to a tenant UUID.
Constraint: the cloud's X-Workspace-ID resolver only accepts a workspace/tenant
UUID, but users see slugs and display names (in list-workspaces output and
memory:// URLs). So when --workspace (or the configured default) is not already
a UUID, fetch the caller's workspaces once and map it to the tenant_id here.
get_available_workspaces is the same workspace-fetch seam project.py uses for
CLI workspace resolution; it is awaited directly here because the share
commands already run inside their own event loop.
"""
from basic_memory.mcp.project_context import get_available_workspaces
workspaces = await get_available_workspaces()
match = _match_workspace_identifier(workspaces, identifier)
if match is None:
available = "\n".join(f" - {ws.slug}" for ws in workspaces)
console.print(
f"[red]Workspace '{identifier}' was not found.[/red]\n"
"[yellow]Use one of these workspace slugs (or a tenant_id):[/yellow]\n"
f"{available}"
)
raise typer.Exit(1)
return match.tenant_id
async def _workspace_headers(
*,
project_name: Optional[str] = None,
workspace: Optional[str] = None,
) -> dict[str, str]:
"""Resolve the target workspace and build the routing header, if any.
Resolution chain (see resolve_configured_workspace): explicit --workspace,
then the project's configured workspace_id, then the global default. Returns
an empty dict when nothing resolves so the request falls back to the
caller's default tenant exactly as before.
The cloud's X-Workspace-ID resolver only accepts a workspace/tenant UUID. A
UUID is forwarded verbatim (covers per-project config workspace_id values and
the default chain, zero extra API calls); any other value is treated as a
human identifier and resolved to the tenant UUID via one workspace lookup.
"""
resolved = resolve_configured_workspace(project_name=project_name, workspace=workspace)
if resolved is None:
return {}
if _is_uuid(resolved):
return {WORKSPACE_ID_HEADER: resolved}
tenant_id = await _resolve_workspace_to_tenant_id(resolved)
return {WORKSPACE_ID_HEADER: tenant_id}
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)",
),
workspace: Optional[str] = typer.Option(
None,
"--workspace",
help="Workspace to route to: a workspace slug, display name, or tenant ID",
),
) -> 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
bm cloud share create my-project notes/my-idea --workspace acme
"""
# Validate --expires-at before any async/API work so a parse error surfaces
# a single clean message and exits, rather than being re-wrapped by the broad
# handler below as "Unexpected error: 1" (typer.Exit subclasses Exception).
payload: dict = {
"project_name": project,
"note_permalink": permalink,
}
if expires_at is not None:
payload["expires_at"] = _parse_expires_at(expires_at)
async def _create():
try:
config_manager = ConfigManager()
config = config_manager.config
host_url = config.cloud_host.rstrip("/")
console.print("[blue]Creating share link...[/blue]")
response = await make_api_request(
method="POST",
url=f"{host_url}/api/shares",
json_data=payload,
headers=await _workspace_headers(project_name=project, workspace=workspace),
)
data = response.json()
console.print("[green]Share link created 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]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",
),
workspace: Optional[str] = typer.Option(
None,
"--workspace",
help="Workspace to route to: a workspace slug, display name, or tenant ID",
),
) -> None:
"""List public share links.
Examples:
bm cloud share list
bm cloud share list --project my-project
bm cloud share list --project my-project --workspace acme
"""
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:
# Encode the filter so project names with query-reserved
# characters (&, +, #, spaces) reach the server intact rather
# than being parsed as extra query parameters.
url += f"?{urlencode({'project_name': project})}"
console.print("[blue]Fetching share links...[/blue]")
response = await make_api_request(
method="GET",
url=url,
headers=await _workspace_headers(project_name=project, workspace=workspace),
)
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)
# Re-raise typer.Exit before the broad handler below: workspace resolution
# raises typer.Exit (a subclass of Exception) for ambiguous/unknown
# identifiers, and that must not be re-wrapped as "Unexpected error: 1".
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:
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.",
),
workspace: Optional[str] = typer.Option(
None,
"--workspace",
help="Workspace the share belongs to: a workspace slug, display name, or tenant ID",
),
) -> 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
bm cloud share update abc123 --disable --workspace acme
"""
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,
headers=await _workspace_headers(workspace=workspace),
)
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",
),
workspace: Optional[str] = typer.Option(
None,
"--workspace",
help="Workspace the share belongs to: a workspace slug, display name, or tenant ID",
),
) -> None:
"""Revoke (delete) a public share link.
Examples:
bm cloud share revoke abc123
bm cloud share revoke abc123 --force
bm cloud share revoke abc123 --force --workspace acme
"""
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}",
headers=await _workspace_headers(workspace=workspace),
)
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())
+999
View File
@@ -0,0 +1,999 @@
"""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 AsyncMock, 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,
)
from basic_memory.schemas.cloud import WorkspaceInfo
# A real workspace/tenant UUID is forwarded verbatim as X-Workspace-ID with no
# workspace lookup; the cloud's resolver only accepts this UUID form.
TENANT_UUID = "5ccbae40-ca03-43a2-b23d-9931eb130e22"
def _workspace(slug: str, tenant_id: str, name: str) -> WorkspaceInfo:
"""Build a WorkspaceInfo for workspace-resolution tests."""
return WorkspaceInfo(
tenant_id=tenant_id,
workspace_type="organization",
slug=slug,
name=name,
role="owner",
is_default=False,
)
def _patch_available_workspaces(workspaces):
"""Patch the workspace list fetch used when --workspace is a slug/name.
Asserts can wrap the returned mock to confirm the lookup was (or was not)
performed for a given invocation.
"""
return patch(
"basic_memory.mcp.project_context.get_available_workspaces",
new=AsyncMock(return_value=workspaces),
)
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
def _patch_workspace(resolved):
"""Patch the workspace resolver used by the share commands.
Returns whatever ``resolved`` is for every lookup, so tests can assert the
X-Workspace-ID header is built (and routed) the way the cloud expects
without depending on real config files.
"""
return patch(
"basic_memory.cli.commands.cloud.shares.resolve_configured_workspace",
return_value=resolved,
)
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")
captured["headers"] = kwargs.get("headers")
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(),
):
with _patch_workspace(TENANT_UUID):
with _patch_available_workspaces([]) as fetch:
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",
}
# Workspace routing: a resolved tenant UUID travels verbatim as the
# X-Workspace-ID header so team-workspace projects aren't evaluated
# against the caller's default tenant.
assert captured["headers"] == {"X-Workspace-ID": TENANT_UUID}
# A UUID needs no resolution: the workspace list is never fetched.
fetch.assert_not_called()
def test_create_share_slug_resolves_to_tenant_uuid(self):
"""A --workspace slug is resolved to the tenant UUID before routing."""
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["headers"] = kwargs.get("headers")
return mock_response
seen = {}
def fake_resolve(*, project_name=None, workspace=None):
seen["project_name"] = project_name
seen["workspace"] = workspace
return workspace
workspaces = [
_workspace("basic-memory-7020de4e925843c68c9056c60d101d9e", TENANT_UUID, "Acme Org"),
_workspace("other-slug", "11111111-1111-1111-1111-111111111111", "Other"),
]
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(),
):
with patch(
"basic_memory.cli.commands.cloud.shares.resolve_configured_workspace",
side_effect=fake_resolve,
):
with _patch_available_workspaces(workspaces) as fetch:
result = runner.invoke(
app,
[
"cloud",
"share",
"create",
"my-project",
"notes/my-idea",
"--workspace",
"basic-memory-7020de4e925843c68c9056c60d101d9e",
],
)
assert result.exit_code == 0
assert seen == {
"project_name": "my-project",
"workspace": "basic-memory-7020de4e925843c68c9056c60d101d9e",
}
# The slug was mapped to the workspace's tenant UUID.
assert captured["headers"] == {"X-Workspace-ID": TENANT_UUID}
fetch.assert_awaited_once()
def test_create_share_display_name_resolves_case_insensitively(self):
"""A --workspace display name resolves case-insensitively to the tenant UUID."""
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["headers"] = kwargs.get("headers")
return mock_response
workspaces = [_workspace("acme-slug", TENANT_UUID, "Acme Org")]
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(),
):
with _patch_workspace("acme org"):
with _patch_available_workspaces(workspaces):
result = runner.invoke(
app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
)
assert result.exit_code == 0
assert captured["headers"] == {"X-Workspace-ID": TENANT_UUID}
def test_create_share_tenant_id_input_passthrough(self):
"""A tenant UUID resolved from config is forwarded without a lookup."""
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["headers"] = kwargs.get("headers")
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(),
):
with _patch_workspace(TENANT_UUID):
with _patch_available_workspaces([]) as fetch:
result = runner.invoke(
app,
[
"cloud",
"share",
"create",
"my-project",
"notes/my-idea",
"--workspace",
TENANT_UUID,
],
)
assert result.exit_code == 0
assert captured["headers"] == {"X-Workspace-ID": TENANT_UUID}
fetch.assert_not_called()
def test_create_share_ambiguous_name_errors_with_candidate_slugs(self):
"""A display name matching multiple workspaces errors and lists candidates."""
runner = CliRunner()
async def mock_make_api_request(*args, **kwargs): # pragma: no cover
raise AssertionError("API should not be called when workspace is ambiguous")
workspaces = [
_workspace("acme-prod", TENANT_UUID, "Acme"),
_workspace("acme-staging", "11111111-1111-1111-1111-111111111111", "Acme"),
]
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(),
):
with _patch_workspace("Acme"):
with _patch_available_workspaces(workspaces):
result = runner.invoke(
app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
)
assert result.exit_code == 1
assert "ambiguous" in result.stdout
# Candidate slugs are listed so the user can disambiguate.
assert "acme-prod" in result.stdout
assert "acme-staging" in result.stdout
# The typer.Exit must not be re-wrapped by the broad handler.
assert "Unexpected error" not in result.stdout
def test_create_share_unknown_workspace_errors_with_available_slugs(self):
"""An unknown identifier errors and lists the available workspace slugs."""
runner = CliRunner()
async def mock_make_api_request(*args, **kwargs): # pragma: no cover
raise AssertionError("API should not be called for an unknown workspace")
workspaces = [
_workspace("acme-prod", TENANT_UUID, "Acme"),
_workspace("widget-co", "11111111-1111-1111-1111-111111111111", "Widget Co"),
]
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(),
):
with _patch_workspace("does-not-exist"):
with _patch_available_workspaces(workspaces):
result = runner.invoke(
app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
)
assert result.exit_code == 1
assert "was not found" in result.stdout
assert "acme-prod" in result.stdout
assert "widget-co" in result.stdout
assert "Unexpected error" not in result.stdout
def test_create_share_no_workspace_sends_no_header(self):
"""When nothing resolves, no routing header is added (default tenant)."""
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["headers"] = kwargs.get("headers")
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(),
):
with _patch_workspace(None):
result = runner.invoke(
app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
)
assert result.exit_code == 0
assert captured["headers"] == {}
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
# A parse error must produce a single clean message, not a
# spurious "Unexpected error: 1" from the broad handler
# re-catching typer.Exit. See issue #880 review.
assert "Unexpected error" not 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 "")
captured["headers"] = kwargs.get("headers")
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(),
):
with _patch_workspace(TENANT_UUID):
result = runner.invoke(
app, ["cloud", "share", "list", "--project", "my-project"]
)
assert result.exit_code == 0
assert "project_name=my-project" in captured["url"]
assert captured["headers"] == {"X-Workspace-ID": TENANT_UUID}
def test_list_shares_unknown_workspace_errors_without_double_error(self):
"""An unknown --workspace on list errors cleanly (no 'Unexpected error').
Exercises the list handler's typer.Exit re-raise: workspace resolution
raises typer.Exit, which must not be re-wrapped by the broad handler.
"""
runner = CliRunner()
async def mock_make_api_request(*args, **kwargs): # pragma: no cover
raise AssertionError("API should not be called for an unknown workspace")
workspaces = [_workspace("acme-prod", TENANT_UUID, "Acme")]
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(),
):
with _patch_workspace("does-not-exist"):
with _patch_available_workspaces(workspaces):
result = runner.invoke(app, ["cloud", "share", "list"])
assert result.exit_code == 1
assert "was not found" in result.stdout
assert "acme-prod" in result.stdout
assert "Unexpected error" not in result.stdout
def test_list_shares_ambiguous_slug_errors_with_candidates(self):
"""A slug colliding across workspaces errors and lists candidates.
Exercises the slug tier of _match_workspace_identifier raising on >1 match.
"""
runner = CliRunner()
async def mock_make_api_request(*args, **kwargs): # pragma: no cover
raise AssertionError("API should not be called when the slug is ambiguous")
workspaces = [
_workspace("shared-slug", TENANT_UUID, "Acme"),
_workspace("shared-slug", "11111111-1111-1111-1111-111111111111", "Widget"),
]
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(),
):
with _patch_workspace("shared-slug"):
with _patch_available_workspaces(workspaces):
result = runner.invoke(app, ["cloud", "share", "list"])
assert result.exit_code == 1
assert "ambiguous" in result.stdout
assert TENANT_UUID in result.stdout
assert "Unexpected error" not in result.stdout
def test_list_shares_project_filter_url_encoded(self):
"""Project names with query-reserved chars must be percent-encoded.
A name like "R&D+notes #1" interpolated raw would split into bogus
query params (project_name=R, plus a stray "D+notes #1" key); encoding
keeps it a single faithful project_name value.
"""
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(),
):
with _patch_workspace(None):
result = runner.invoke(
app, ["cloud", "share", "list", "--project", "R&D+notes #1"]
)
assert result.exit_code == 0
# Reserved characters are percent-encoded into a single value.
assert "project_name=R%26D%2Bnotes+%231" in captured["url"]
# And the raw, ambiguous form never reaches the wire.
assert "project_name=R&D" not 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")
captured["headers"] = kwargs.get("headers")
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(),
):
with _patch_workspace(TENANT_UUID):
result = runner.invoke(
app,
["cloud", "share", "update", "abc123", "--disable", "--workspace", "acme"],
)
assert result.exit_code == 0
assert "updated successfully" in result.stdout
assert captured["method"] == "PATCH"
assert captured["json_data"] == {"enabled": False}
assert captured["headers"] == {"X-Workspace-ID": TENANT_UUID}
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")
captured["headers"] = kwargs.get("headers")
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(),
):
with _patch_workspace(TENANT_UUID):
result = runner.invoke(
app,
["cloud", "share", "revoke", "abc123", "--force", "--workspace", "acme"],
)
assert result.exit_code == 0
assert "revoked successfully" in result.stdout
assert captured["method"] == "DELETE"
assert captured["url"].endswith("/api/shares/abc123")
assert captured["headers"] == {"X-Workspace-ID": TENANT_UUID}
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