mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 8476a16757 | |||
| cacfdc4adc |
+1333
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,726 @@
|
||||
#!/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,11 +11,9 @@ 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
|
||||
|
||||
@@ -1,533 +0,0 @@
|
||||
"""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())
|
||||
@@ -1,999 +0,0 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user