basicmachines-co-basic-memory/tests/cli/test_cli_tool_json_output.py

"""Tests for CLI tool commands.

All commands return JSON via MCP tool output_format="json".
Tests mock the MCP tool functions directly.
"""

import json
from unittest.mock import AsyncMock, patch

from typer.testing import CliRunner

from basic_memory.cli.main import app as cli_app

runner = CliRunner()

# --- Shared mock data ---

WRITE_NOTE_RESULT = {
    "title": "Test Note",
    "permalink": "notes/test-note",
    "file_path": "notes/Test Note.md",
    "checksum": "abc123",
    "action": "created",
}

READ_NOTE_RESULT = {
    "title": "Test Note",
    "permalink": "notes/test-note",
    "file_path": "notes/Test Note.md",
    "content": "# Test Note\n\nhello world",
    "frontmatter": {"title": "Test Note", "tags": ["test"]},
}

EDIT_NOTE_RESULT = {
    "title": "Test Note",
    "permalink": "notes/test-note",
    "file_path": "notes/Test Note.md",
    "checksum": "def456",
    "operation": "append",
}

BUILD_CONTEXT_RESULT = {
    "results": [],
    "metadata": {"uri": "test/topic", "depth": 1},
    "page": 1,
    "page_size": 10,
}

RECENT_ACTIVITY_RESULT = [
    {
        "type": "entity",
        "title": "Note A",
        "permalink": "notes/note-a",
        "file_path": "notes/Note A.md",
        "created_at": "2025-01-01 00:00:00",
    },
    {
        "type": "entity",
        "title": "Note B",
        "permalink": "notes/note-b",
        "file_path": "notes/Note B.md",
        "created_at": "2025-01-02 00:00:00",
    },
]

SEARCH_RESULT = {
    "query": "test",
    "total": 1,
    "page": 1,
    "page_size": 10,
    "results": [
        {
            "type": "entity",
            "title": "Test Note",
            "permalink": "notes/test-note",
            "file_path": "notes/Test Note.md",
        }
    ],
}


# --- write-note ---


@patch(
    "basic_memory.cli.commands.tool.mcp_write_note",
    new_callable=AsyncMock,
    return_value=WRITE_NOTE_RESULT,
)
def test_write_note_json_output(mock_mcp_write):
    """write-note outputs valid JSON from MCP tool."""
    result = runner.invoke(
        cli_app,
        [
            "tool",
            "write-note",
            "--title",
            "Test Note",
            "--folder",
            "notes",
            "--content",
            "hello world",
        ],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    data = json.loads(result.output)
    assert data["title"] == "Test Note"
    assert data["permalink"] == "notes/test-note"
    assert data["file_path"] == "notes/Test Note.md"
    mock_mcp_write.assert_called_once()
    # Verify output_format="json" was passed
    assert mock_mcp_write.call_args.kwargs["output_format"] == "json"


@patch(
    "basic_memory.cli.commands.tool.mcp_write_note",
    new_callable=AsyncMock,
    return_value=WRITE_NOTE_RESULT,
)
def test_write_note_with_tags(mock_mcp_write):
    """write-note passes tags through to MCP tool."""
    result = runner.invoke(
        cli_app,
        [
            "tool",
            "write-note",
            "--title",
            "Test Note",
            "--folder",
            "notes",
            "--content",
            "hello",
            "--tags",
            "python",
            "--tags",
            "async",
        ],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    assert mock_mcp_write.call_args.kwargs["tags"] == ["python", "async"]


# --- read-note ---


@patch(
    "basic_memory.cli.commands.tool.mcp_read_note",
    new_callable=AsyncMock,
    return_value=READ_NOTE_RESULT,
)
def test_read_note_json_output(mock_mcp_read):
    """read-note outputs valid JSON from MCP tool."""
    result = runner.invoke(
        cli_app,
        ["tool", "read-note", "test-note"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    data = json.loads(result.output)
    assert data["title"] == "Test Note"
    assert data["permalink"] == "notes/test-note"
    assert data["content"] == "# Test Note\n\nhello world"
    assert data["frontmatter"] == {"title": "Test Note", "tags": ["test"]}
    mock_mcp_read.assert_called_once()
    assert mock_mcp_read.call_args.kwargs["output_format"] == "json"


@patch(
    "basic_memory.cli.commands.tool.mcp_read_note",
    new_callable=AsyncMock,
    return_value=READ_NOTE_RESULT,
)
def test_read_note_workspace_passthrough(mock_mcp_read):
    """read-note --workspace passes workspace through to the MCP tool call."""
    result = runner.invoke(
        cli_app,
        ["tool", "read-note", "test-note", "--workspace", "tenant-123"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    assert mock_mcp_read.call_args.kwargs["workspace"] == "tenant-123"


@patch(
    "basic_memory.cli.commands.tool.mcp_read_note",
    new_callable=AsyncMock,
    return_value=READ_NOTE_RESULT,
)
def test_read_note_include_frontmatter(mock_mcp_read):
    """read-note --include-frontmatter passes flag through to MCP tool."""
    result = runner.invoke(
        cli_app,
        ["tool", "read-note", "test-note", "--include-frontmatter"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    assert mock_mcp_read.call_args.kwargs["include_frontmatter"] is True


@patch(
    "basic_memory.cli.commands.tool.mcp_read_note",
    new_callable=AsyncMock,
    return_value=READ_NOTE_RESULT,
)
def test_read_note_pagination(mock_mcp_read):
    """read-note --page and --page-size are passed through."""
    result = runner.invoke(
        cli_app,
        ["tool", "read-note", "test-note", "--page", "2", "--page-size", "5"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    assert mock_mcp_read.call_args.kwargs["page"] == 2
    assert mock_mcp_read.call_args.kwargs["page_size"] == 5


# --- edit-note ---


@patch(
    "basic_memory.cli.commands.tool.mcp_edit_note",
    new_callable=AsyncMock,
    return_value=EDIT_NOTE_RESULT,
)
def test_edit_note_json_output(mock_mcp_edit):
    """edit-note outputs valid JSON from MCP tool."""
    result = runner.invoke(
        cli_app,
        [
            "tool",
            "edit-note",
            "test-note",
            "--operation",
            "append",
            "--content",
            "new content",
        ],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    data = json.loads(result.output)
    assert data["title"] == "Test Note"
    assert data["operation"] == "append"
    mock_mcp_edit.assert_called_once()
    assert mock_mcp_edit.call_args.kwargs["output_format"] == "json"


@patch(
    "basic_memory.cli.commands.tool.mcp_edit_note",
    new_callable=AsyncMock,
    return_value={"title": "Test", "permalink": "test", "error": "Edit failed: not found"},
)
def test_edit_note_error_response(mock_mcp_edit):
    """edit-note exits with code 1 when MCP tool returns error field."""
    result = runner.invoke(
        cli_app,
        [
            "tool",
            "edit-note",
            "test-note",
            "--operation",
            "append",
            "--content",
            "content",
        ],
    )

    assert result.exit_code == 1


# --- build-context ---


@patch(
    "basic_memory.cli.commands.tool.mcp_build_context",
    new_callable=AsyncMock,
    return_value=BUILD_CONTEXT_RESULT,
)
def test_build_context_json_output(mock_build_ctx):
    """build-context outputs valid JSON from MCP tool."""
    result = runner.invoke(
        cli_app,
        ["tool", "build-context", "memory://test/topic"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    data = json.loads(result.output)
    assert "results" in data
    mock_build_ctx.assert_called_once()
    assert mock_build_ctx.call_args.kwargs["output_format"] == "json"


@patch(
    "basic_memory.cli.commands.tool.mcp_build_context",
    new_callable=AsyncMock,
    return_value=BUILD_CONTEXT_RESULT,
)
def test_build_context_with_options(mock_build_ctx):
    """build-context passes depth, timeframe, pagination through."""
    result = runner.invoke(
        cli_app,
        [
            "tool",
            "build-context",
            "memory://test/topic",
            "--depth",
            "2",
            "--timeframe",
            "30d",
            "--page",
            "3",
            "--max-related",
            "5",
        ],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    kwargs = mock_build_ctx.call_args.kwargs
    assert kwargs["depth"] == 2
    assert kwargs["timeframe"] == "30d"
    assert kwargs["page"] == 3
    assert kwargs["max_related"] == 5


# --- recent-activity ---


@patch(
    "basic_memory.cli.commands.tool.mcp_recent_activity",
    new_callable=AsyncMock,
    return_value=RECENT_ACTIVITY_RESULT,
)
def test_recent_activity_json_output(mock_mcp_recent):
    """recent-activity outputs valid JSON list from MCP tool."""
    result = runner.invoke(
        cli_app,
        ["tool", "recent-activity"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    data = json.loads(result.output)
    assert isinstance(data, list)
    assert len(data) == 2
    assert data[0]["title"] == "Note A"
    assert data[1]["title"] == "Note B"
    mock_mcp_recent.assert_called_once()
    assert mock_mcp_recent.call_args.kwargs["output_format"] == "json"


@patch(
    "basic_memory.cli.commands.tool.mcp_recent_activity",
    new_callable=AsyncMock,
    return_value=RECENT_ACTIVITY_RESULT,
)
def test_recent_activity_pagination(mock_mcp_recent):
    """recent-activity passes --page and --page-size through."""
    result = runner.invoke(
        cli_app,
        ["tool", "recent-activity", "--page", "2", "--page-size", "10"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    kwargs = mock_mcp_recent.call_args.kwargs
    assert kwargs["page"] == 2
    assert kwargs["page_size"] == 10


@patch(
    "basic_memory.cli.commands.tool.mcp_recent_activity",
    new_callable=AsyncMock,
    return_value=[],
)
def test_recent_activity_empty(mock_mcp_recent):
    """recent-activity handles empty results."""
    result = runner.invoke(
        cli_app,
        ["tool", "recent-activity"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    data = json.loads(result.output)
    assert data == []


# --- search-notes ---


@patch(
    "basic_memory.cli.commands.tool.mcp_search",
    new_callable=AsyncMock,
    return_value=SEARCH_RESULT,
)
def test_search_notes_json_output(mock_mcp_search):
    """search-notes outputs valid JSON from MCP tool."""
    result = runner.invoke(
        cli_app,
        ["tool", "search-notes", "test query"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    data = json.loads(result.output)
    assert data["total"] == 1
    assert data["results"][0]["title"] == "Test Note"
    mock_mcp_search.assert_called_once()
    assert mock_mcp_search.call_args.kwargs["output_format"] == "json"


@patch(
    "basic_memory.cli.commands.tool.mcp_search",
    new_callable=AsyncMock,
    return_value=SEARCH_RESULT,
)
def test_search_notes_with_meta_filter(mock_mcp_search):
    """search-notes --meta key=value builds metadata filters."""
    result = runner.invoke(
        cli_app,
        ["tool", "search-notes", "query", "--meta", "status=draft"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    assert mock_mcp_search.call_args.kwargs["metadata_filters"] == {"status": "draft"}


@patch(
    "basic_memory.cli.commands.tool.mcp_search",
    new_callable=AsyncMock,
    return_value=SEARCH_RESULT,
)
def test_search_notes_permalink_mode(mock_mcp_search):
    """search-notes --permalink sets search_type."""
    result = runner.invoke(
        cli_app,
        ["tool", "search-notes", "specs/*", "--permalink"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    assert mock_mcp_search.call_args.kwargs["search_type"] == "permalink"


@patch(
    "basic_memory.cli.commands.tool.mcp_search",
    new_callable=AsyncMock,
    return_value="Error: search failed",
)
def test_search_notes_string_error(mock_mcp_search):
    """search-notes exits with code 1 when MCP returns string error."""
    result = runner.invoke(
        cli_app,
        ["tool", "search-notes", "query"],
    )

    assert result.exit_code == 1


# --- Routing flags ---


def test_routing_both_flags_error():
    """Commands exit with error when both --local and --cloud are specified."""
    result = runner.invoke(
        cli_app,
        [
            "tool",
            "recent-activity",
            "--local",
            "--cloud",
        ],
    )

    assert result.exit_code == 1


# --- schema-validate ---

SCHEMA_VALIDATE_RESULT = {
    "entity_type": "person",
    "total_notes": 2,
    "total_entities": 2,
    "valid_count": 1,
    "warning_count": 1,
    "error_count": 1,
    "results": [
        {
            "note_identifier": "people/alice",
            "schema_entity": "person",
            "passed": True,
            "warnings": [],
            "errors": [],
        },
        {
            "note_identifier": "people/bob",
            "schema_entity": "person",
            "passed": False,
            "warnings": ["Missing optional field: role"],
            "errors": ["Missing required field: name"],
        },
    ],
}


@patch(
    "basic_memory.cli.commands.tool.mcp_schema_validate",
    new_callable=AsyncMock,
    return_value=SCHEMA_VALIDATE_RESULT,
)
def test_schema_validate_json_output(mock_mcp):
    """schema-validate outputs valid JSON from MCP tool."""
    result = runner.invoke(
        cli_app,
        ["tool", "schema-validate", "person"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    data = json.loads(result.output)
    assert data["entity_type"] == "person"
    assert data["total_notes"] == 2
    mock_mcp.assert_called_once()
    assert mock_mcp.call_args.kwargs["output_format"] == "json"


@patch(
    "basic_memory.cli.commands.tool.mcp_schema_validate",
    new_callable=AsyncMock,
    return_value=SCHEMA_VALIDATE_RESULT,
)
def test_schema_validate_identifier_heuristic(mock_mcp):
    """schema-validate treats target with / as identifier, not note_type."""
    result = runner.invoke(
        cli_app,
        ["tool", "schema-validate", "people/alice.md"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    assert mock_mcp.call_args.kwargs["identifier"] == "people/alice.md"
    assert mock_mcp.call_args.kwargs["note_type"] is None


@patch(
    "basic_memory.cli.commands.tool.mcp_schema_validate",
    new_callable=AsyncMock,
    return_value={"error": "No notes found of type 'person'"},
)
def test_schema_validate_error_response(mock_mcp):
    """schema-validate outputs error JSON when MCP returns error dict."""
    result = runner.invoke(
        cli_app,
        ["tool", "schema-validate", "person"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    data = json.loads(result.output)
    assert "error" in data


# --- schema-infer ---

SCHEMA_INFER_RESULT = {
    "entity_type": "person",
    "notes_analyzed": 5,
    "field_frequencies": [
        {"name": "name", "source": "observation", "count": 5, "total": 5, "percentage": 1.0},
        {"name": "role", "source": "observation", "count": 3, "total": 5, "percentage": 0.6},
    ],
    "suggested_schema": {"name": "string, full name", "role?": "string, job title"},
    "suggested_required": ["name"],
    "suggested_optional": ["role"],
    "excluded": [],
}


@patch(
    "basic_memory.cli.commands.tool.mcp_schema_infer",
    new_callable=AsyncMock,
    return_value=SCHEMA_INFER_RESULT,
)
def test_schema_infer_json_output(mock_mcp):
    """schema-infer outputs valid JSON from MCP tool."""
    result = runner.invoke(
        cli_app,
        ["tool", "schema-infer", "person"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    data = json.loads(result.output)
    assert data["entity_type"] == "person"
    assert data["notes_analyzed"] == 5
    mock_mcp.assert_called_once()
    assert mock_mcp.call_args.kwargs["output_format"] == "json"


@patch(
    "basic_memory.cli.commands.tool.mcp_schema_infer",
    new_callable=AsyncMock,
    return_value=SCHEMA_INFER_RESULT,
)
def test_schema_infer_threshold_passthrough(mock_mcp):
    """schema-infer passes --threshold through to MCP tool."""
    result = runner.invoke(
        cli_app,
        ["tool", "schema-infer", "person", "--threshold", "0.5"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    assert mock_mcp.call_args.kwargs["threshold"] == 0.5


# --- schema-diff ---

SCHEMA_DIFF_RESULT = {
    "entity_type": "person",
    "schema_found": True,
    "new_fields": [
        {"name": "email", "source": "observation", "count": 3, "total": 5, "percentage": 0.6}
    ],
    "dropped_fields": [],
    "cardinality_changes": [],
}


@patch(
    "basic_memory.cli.commands.tool.mcp_schema_diff",
    new_callable=AsyncMock,
    return_value=SCHEMA_DIFF_RESULT,
)
def test_schema_diff_json_output(mock_mcp):
    """schema-diff outputs valid JSON from MCP tool."""
    result = runner.invoke(
        cli_app,
        ["tool", "schema-diff", "person"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    data = json.loads(result.output)
    assert data["entity_type"] == "person"
    assert data["schema_found"] is True
    assert len(data["new_fields"]) == 1
    mock_mcp.assert_called_once()
    assert mock_mcp.call_args.kwargs["output_format"] == "json"


# --- list-projects ---

LIST_PROJECTS_RESULT = {
    "projects": [
        {
            "name": "main",
            "path": "/home/user/notes",
            "is_default": True,
            "status": "active",
        },
        {
            "name": "research",
            "path": "/home/user/research",
            "is_default": False,
            "status": "active",
        },
    ],
    "count": 2,
}


@patch(
    "basic_memory.cli.commands.tool.mcp_list_projects",
    new_callable=AsyncMock,
    return_value=LIST_PROJECTS_RESULT,
)
def test_list_projects_json_output(mock_mcp):
    """list-projects outputs valid JSON from MCP tool."""
    result = runner.invoke(
        cli_app,
        ["tool", "list-projects"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    data = json.loads(result.output)
    assert data["count"] == 2
    assert len(data["projects"]) == 2
    assert data["projects"][0]["name"] == "main"
    mock_mcp.assert_called_once()
    assert mock_mcp.call_args.kwargs["output_format"] == "json"


# --- list-workspaces ---

LIST_WORKSPACES_RESULT = {
    "workspaces": [
        {
            "tenant_id": "tenant-abc",
            "name": "My Workspace",
            "workspace_type": "personal",
            "role": "owner",
            "organization_id": None,
            "has_active_subscription": True,
        },
    ],
    "count": 1,
}


@patch(
    "basic_memory.cli.commands.tool.mcp_list_workspaces",
    new_callable=AsyncMock,
    return_value=LIST_WORKSPACES_RESULT,
)
def test_list_workspaces_json_output(mock_mcp):
    """list-workspaces outputs valid JSON from MCP tool."""
    result = runner.invoke(
        cli_app,
        ["tool", "list-workspaces"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    data = json.loads(result.output)
    assert data["count"] == 1
    assert data["workspaces"][0]["tenant_id"] == "tenant-abc"
    assert data["workspaces"][0]["name"] == "My Workspace"
    mock_mcp.assert_called_once()
    assert mock_mcp.call_args.kwargs["output_format"] == "json"


@patch(
    "basic_memory.cli.commands.tool.mcp_list_workspaces",
    new_callable=AsyncMock,
    return_value={"workspaces": [], "count": 0},
)
def test_list_workspaces_empty(mock_mcp):
    """list-workspaces handles empty workspace list."""
    result = runner.invoke(
        cli_app,
        ["tool", "list-workspaces"],
    )

    assert result.exit_code == 0, f"CLI failed: {result.output}"
    data = json.loads(result.output)
    assert data["workspaces"] == []
    assert data["count"] == 0