admin/basicmachines-co-basic-memory
1
0
Fork 0
mirror of https://github.com/basicmachines-co/basic-memory synced 2026-06-21 13:47:35 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
basicmachines-co-basic-memory/tests/mcp/test_tool_search.py
T
Drew Cain 0811c48252 fix(mcp): normalize note_types/entity_types/categories on direct call path and reject non-string list elements 
Codex review of PR #962 identified two real issues:

1. CLI bypass: the BeforeValidator(parse_str_list) on note_types, entity_types, and
   categories only fires through MCP/Pydantic validation. The CLI path in
   cli/commands/tool.py calls search_notes() directly, so `bm tool search-notes
   --type note,task` arrived as note_types=["note,task"] and matched nothing.
   Fix: add in-body parse_str_list() normalization for all three params (mirroring
   the existing parse_tags() call for tags on the same code path).

2. Silent stringify: parse_str_list used str(raw) in the list branch, so [42] became
   ["42"] before Pydantic saw it, accepting invalid input as a no-result search
   instead of rejecting it. Fix: guard against non-string list elements and return
   the original value unchanged so Pydantic rejects it with a clear error.

Tests added: annotation-level split tests for note_types/entity_types/categories,
non-string-element rejection tests, async direct-call regression for note_types,
and unit-level parse_str_list non-string list tests in test_coerce.py.

Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: Drew Cain <groksrc@gmail.com>
2026-06-11 14:38:24 -05:00

2276 lines
81 KiB
Python
Raw Permalink Blame History

"""Tests for search MCP tools."""
import inspect
import pytest
from contextlib import asynccontextmanager
from datetime import datetime, timedelta
from types import SimpleNamespace
from typing import cast
from pydantic import TypeAdapter
from basic_memory.mcp.tools import write_note
from basic_memory.mcp.tools.search import (
search_notes,
_format_search_error_response,
_format_search_markdown,
)
from basic_memory.schemas.search import SearchResponse
@pytest.mark.asyncio
async def test_search_text(client, test_project):
"""Test basic search functionality."""
# Create a test note
result = await write_note(
project=test_project.name,
title="Test Search Note",
directory="test",
content="# Test\nThis is a searchable test note",
tags=["test", "search"],
)
assert result
# Search for it (use json format to inspect structured results)
response = await search_notes(
project=test_project.name,
query="searchable",
search_type="text",
output_format="json",
)
# Verify results - handle both success and error cases
if isinstance(response, dict):
# Success case - verify dict response
assert len(response["results"]) > 0
assert any(
r["permalink"] == f"{test_project.name}/test/test-search-note"
for r in response["results"]
)
else:
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
@pytest.mark.asyncio
async def test_search_title(client, test_project):
"""Test basic search functionality."""
# Create a test note
result = await write_note(
project=test_project.name,
title="Test Search Note",
directory="test",
content="# Test\nThis is a searchable test note",
tags=["test", "search"],
)
assert result
# Search for it (use json format to inspect structured results)
response = await search_notes(
project=test_project.name, query="Search Note", search_type="title", output_format="json"
)
# Verify results - handle both success and error cases
if isinstance(response, dict):
# Success case - verify dict response
assert len(response["results"]) > 0
assert any(
r["permalink"] == f"{test_project.name}/test/test-search-note"
for r in response["results"]
)
else:
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
@pytest.mark.asyncio
async def test_search_permalink(client, test_project):
"""Test basic search functionality."""
# Create a test note
result = await write_note(
project=test_project.name,
title="Test Search Note",
directory="test",
content="# Test\nThis is a searchable test note",
tags=["test", "search"],
)
assert result
# Search for it (use json format to inspect structured results)
response = await search_notes(
project=test_project.name,
query=f"{test_project.name}/test/test-search-note",
search_type="permalink",
output_format="json",
)
# Verify results - handle both success and error cases
if isinstance(response, dict):
# Success case - verify dict response
assert len(response["results"]) > 0
assert any(
r["permalink"] == f"{test_project.name}/test/test-search-note"
for r in response["results"]
)
else:
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
@pytest.mark.asyncio
async def test_search_permalink_match(client, test_project):
"""Test basic search functionality."""
# Create a test note
result = await write_note(
project=test_project.name,
title="Test Search Note",
directory="test",
content="# Test\nThis is a searchable test note",
tags=["test", "search"],
)
assert result
# Search for it (use json format to inspect structured results)
response = await search_notes(
project=test_project.name,
query=f"{test_project.name}/test/test-search-*",
search_type="permalink",
output_format="json",
)
# Verify results - handle both success and error cases
if isinstance(response, dict):
# Success case - verify dict response
assert len(response["results"]) > 0
assert any(
r["permalink"] == f"{test_project.name}/test/test-search-note"
for r in response["results"]
)
else:
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
@pytest.mark.asyncio
async def test_search_memory_url_with_project_prefix(client, test_project):
"""Test searching with a memory:// URL that includes the project prefix."""
result = await write_note(
project=test_project.name,
title="Memory URL Search Note",
directory="test",
content="# Memory URL Search\nThis note should be found via memory URL search",
)
assert result
response = await search_notes(
query=f"memory://{test_project.name}/test/memory-url-search-note", output_format="json"
)
if isinstance(response, dict):
assert len(response["results"]) > 0
assert any(
r["permalink"] == f"{test_project.name}/test/memory-url-search-note"
for r in response["results"]
)
else:
pytest.fail(f"Search failed with error: {response}")
@pytest.mark.asyncio
async def test_search_workspace_memory_url_routes_with_local_config(monkeypatch, config_manager):
"""Workspace-qualified memory URL searches should self-route in mixed mode."""
import importlib
import basic_memory.mcp.project_context as project_context
from basic_memory.config import ProjectEntry
from basic_memory.mcp.project_context import (
WorkspaceProjectEntry,
_build_workspace_project_index,
)
from basic_memory.schemas.cloud import WorkspaceInfo
from basic_memory.schemas.project_info import ProjectItem
from basic_memory.schemas.search import SearchItemType, SearchResult
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
config = config_manager.load_config()
config.projects["hermes-memory"] = ProjectEntry(
path=str(config_manager.config_dir.parent / "hermes-memory")
)
config.cloud_api_key = "bmc_test123"
config_manager.save_config(config)
personal = WorkspaceInfo(
tenant_id="personal-tenant",
workspace_type="personal",
slug="personal",
name="Personal",
role="owner",
is_default=True,
)
project_item = ProjectItem(
id=1,
external_id="11111111-1111-1111-1111-111111111111",
name="main",
path="/tmp/main",
is_default=False,
)
index = _build_workspace_project_index(
(personal,),
(WorkspaceProjectEntry(workspace=personal, project=project_item),),
)
async def fake_index(context=None):
return index
captured: dict[str, object] = {}
@asynccontextmanager
async def fake_get_project_client(project=None, context=None, project_id=None):
captured["project"] = project
captured["project_id"] = project_id
yield object(), SimpleNamespace(name="main", external_id=project_item.external_id)
async def fake_resolve_project_and_path(client, identifier, project=None, context=None):
assert identifier == "memory://personal/main/tests/search-note"
assert project == "main"
return None, "personal/main/tests/search-note", True
class FakeSearchClient:
def __init__(self, client, project_id):
captured["search_project_id"] = project_id
async def search(self, payload, *, page, page_size):
captured["payload"] = payload
return SearchResponse(
results=[
SearchResult(
title="Search Note",
type=SearchItemType.ENTITY,
score=1.0,
permalink="personal/main/tests/search-note",
file_path="tests/Search Note.md",
)
],
current_page=page,
page_size=page_size,
total=1,
)
monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", FakeSearchClient)
response = await search_notes(
query="memory://personal/main/tests/search-note",
output_format="json",
)
assert captured["project"] == "personal/main"
assert captured["project_id"] is None
assert captured["search_project_id"] == "11111111-1111-1111-1111-111111111111"
payload = cast(dict[str, object], captured["payload"])
assert payload["permalink"] == "personal/main/tests/search-note"
assert isinstance(response, dict)
assert response["results"][0]["permalink"] == "personal/main/tests/search-note"
@pytest.mark.asyncio
async def test_search_pagination(client, test_project):
"""Test basic search functionality."""
# Create a test note
result = await write_note(
project=test_project.name,
title="Test Search Note",
directory="test",
content="# Test\nThis is a searchable test note",
tags=["test", "search"],
)
assert result
# Search for it (use json format to inspect structured results)
response = await search_notes(
project=test_project.name,
query="searchable",
search_type="text",
page=1,
page_size=1,
output_format="json",
)
# Verify results - handle both success and error cases
if isinstance(response, dict):
# Success case - verify dict response
assert len(response["results"]) == 1
assert any(
r["permalink"] == f"{test_project.name}/test/test-search-note"
for r in response["results"]
)
else:
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
@pytest.mark.asyncio
async def test_search_with_type_filter(client, test_project):
"""Test search with note type filter."""
# Create test content
await write_note(
project=test_project.name,
title="Note Type Test",
directory="test",
content="# Test\nFiltered by type",
)
# Search with note type filter (use json format to inspect structured results)
response = await search_notes(
project=test_project.name,
query="type",
search_type="text",
note_types=["note"],
output_format="json",
)
# Verify results - handle both success and error cases
if isinstance(response, dict):
# Success case - verify all results are entities
assert all(r["type"] == "entity" for r in response["results"])
else:
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
@pytest.mark.asyncio
async def test_search_with_entity_type_filter(client, test_project):
"""Test search with entity_types (SearchItemType) filter."""
# Create test content
await write_note(
project=test_project.name,
title="Entity Type Test",
directory="test",
content="# Test\nFiltered by type",
)
# Search with entity_types (SearchItemType) filter (use json format)
response = await search_notes(
project=test_project.name,
query="type",
search_type="text",
entity_types=["entity"],
output_format="json",
)
# Verify results - handle both success and error cases
if isinstance(response, dict):
# Success case - verify all results are entities
assert all(r["type"] == "entity" for r in response["results"])
else:
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
@pytest.mark.asyncio
async def test_search_with_categories_filter(client, test_project):
"""Observation category filter returns only the exact category (#430).
Writes a note whose body has a [requirement] observation and a [decision]
observation that also mentions the word "requirement". The categories filter
must return only the requirement observation.
"""
await write_note(
project=test_project.name,
title="Category Filter Note",
directory="test",
content=(
"# Category Filter Note\n"
"- [requirement] The system must enforce auth on every request\n"
"- [decision] We deferred the auth requirement to next sprint\n"
),
)
response = await search_notes(
project=test_project.name,
query="requirement",
search_type="text",
entity_types=["observation"],
categories=["requirement"],
output_format="json",
)
assert isinstance(response, dict), f"Search failed with error: {response}"
results = response["results"]
assert len(results) > 0
# Every result is a requirement observation; the [decision] row is excluded
# even though its text contains the word "requirement".
assert all(r["type"] == "observation" for r in results)
assert all(r["category"] == "requirement" for r in results)
# A non-matching category yields no results for the same text query.
decision = await search_notes(
project=test_project.name,
query="requirement",
search_type="text",
entity_types=["observation"],
categories=["decision"],
output_format="json",
)
assert isinstance(decision, dict), f"Search failed with error: {decision}"
assert all(r["category"] == "decision" for r in decision["results"])
# The requirement observation must not leak into a decision-scoped search.
assert all("requirement" != r.get("category") for r in decision["results"])
@pytest.mark.asyncio
async def test_search_categories_without_entity_types_returns_observations(client, test_project):
"""categories=[...] WITHOUT entity_types must return the matching observations (#908).
search_notes defaults entity_types to "entity" when unset, but categories only exist on
observations — so a category filter without an explicit entity_types would AND the
category against entity rows (which have NULL category) and return nothing. The implicit
default must scope to observations when categories is supplied.
"""
await write_note(
project=test_project.name,
title="Category Default Note",
directory="test",
content=(
"# Category Default Note\n"
"- [requirement] Auth tokens must rotate every 24 hours\n"
"- [decision] We chose JWT for the auth token format\n"
),
)
# Note: no entity_types passed — exercises the implicit default.
response = await search_notes(
project=test_project.name,
query="auth",
search_type="text",
categories=["requirement"],
output_format="json",
)
assert isinstance(response, dict), f"Search failed with error: {response}"
results = response["results"]
assert len(results) > 0, "category-only search must return matching observations"
assert all(r["type"] == "observation" for r in results)
assert all(r["category"] == "requirement" for r in results)
@pytest.mark.asyncio
async def test_search_with_date_filter(client, test_project):
"""Test search with date filter."""
# Create test content
await write_note(
project=test_project.name,
title="Recent Note",
directory="test",
content="# Test\nRecent content",
)
# Search with date filter (use json format to inspect structured results)
one_hour_ago = datetime.now() - timedelta(hours=1)
response = await search_notes(
project=test_project.name,
query="recent",
search_type="text",
after_date=one_hour_ago.isoformat(),
output_format="json",
)
# Verify results - handle both success and error cases
if isinstance(response, dict):
# Success case - verify we get results within timeframe
assert len(response["results"]) > 0
else:
# If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}")
class TestSearchErrorFormatting:
"""Test search error formatting for better user experience."""
def test_format_search_error_fts5_syntax(self):
"""Test formatting for FTS5 syntax errors."""
result = _format_search_error_response(
"test-project", "syntax error in FTS5", "test query("
)
assert "# Search Failed - Invalid Syntax" in result
assert "The search query 'test query(' contains invalid syntax" in result
assert "Special characters" in result
assert "test query" in result # Clean query without special chars
def test_format_search_error_no_results(self):
"""Test formatting for no results found."""
result = _format_search_error_response(
"test-project", "no results found", "very specific query"
)
assert "# Search Complete - No Results Found" in result
assert "No content found matching 'very specific query'" in result
assert "Broaden your search" in result
assert "very" in result # Simplified query
def test_format_search_error_server_error(self):
"""Test formatting for server errors."""
result = _format_search_error_response(
"test-project", "internal server error", "test query"
)
assert "# Search Failed - Server Error" in result
assert "The search service encountered an error while processing 'test query'" in result
assert "Try again" in result
assert "Check project status" in result
def test_format_search_error_permission_denied(self):
"""Test formatting for permission errors."""
result = _format_search_error_response("test-project", "permission denied", "test query")
assert "# Search Failed - Access Error" in result
assert "You don't have permission to search" in result
assert "Check your project access" in result
def test_format_search_error_project_not_found(self):
"""Test formatting for project not found errors."""
result = _format_search_error_response(
"test-project", "current project not found", "test query"
)
assert "# Search Failed - Project Not Found" in result
assert "The current project is not accessible" in result
assert "Check available projects" in result
def test_format_search_error_semantic_disabled(self):
"""Test formatting for semantic-search-disabled errors."""
result = _format_search_error_response(
"test-project",
"Semantic search is disabled. Set BASIC_MEMORY_SEMANTIC_SEARCH_ENABLED=true.",
"semantic query",
"vector",
)
assert "# Search Failed - Semantic Search Disabled" in result
assert "BASIC_MEMORY_SEMANTIC_SEARCH_ENABLED=true" in result
assert 'search_type="text"' in result
def test_format_search_error_semantic_dependencies_missing(self):
"""Test formatting for missing semantic dependencies."""
result = _format_search_error_response(
"test-project",
"fastembed package is missing. Install/update basic-memory to include semantic dependencies: pip install -U basic-memory",
"semantic query",
"hybrid",
)
assert "# Search Failed - Semantic Dependencies Missing" in result
assert "pip install -U basic-memory" in result
def test_format_search_error_corrupt_embedding_model(self):
"""Test formatting for a corrupt/missing FastEmbed model (ONNX NO_SUCHFILE)."""
from basic_memory.config import ConfigManager
from basic_memory.repository.embedding_provider_factory import _resolve_cache_dir
result = _format_search_error_response(
"test-project",
"[ONNXRuntimeError] : 3 : NO_SUCHFILE : Load model from "
"/home/u/.basic-memory/fastembed_cache/models--qdrant--bge-small-en-v1.5-onnx-q/"
"snapshots/abc/model_optimized.onnx failed. File doesn't exist",
"semantic query",
"hybrid",
)
expected_cache_dir = _resolve_cache_dir(ConfigManager().config)
assert "# Search Failed - Embedding Model Missing or Corrupt" in result
# Names the actual resolved cache dir so the user knows what to delete.
assert expected_cache_dir in result
# Offers full-text search as an immediate workaround.
assert 'search_type="text"' in result
def test_format_search_error_load_model_phrase_does_not_overmatch(self):
"""A generic error mentioning 'load model' (no 'from') must not hit the embedding branch.
The marker was tightened from the broad 'load model' to the exact ONNX phrasing
'load model from' so unrelated failures fall through to the generic handler.
"""
result = _format_search_error_response(
"test-project",
"Failed to load model configuration for this project",
"test query",
)
assert "# Search Failed - Embedding Model Missing or Corrupt" not in result
assert "# Search Failed" in result
def test_format_search_error_generic(self):
"""Test formatting for generic errors."""
result = _format_search_error_response("test-project", "unknown error", "test query")
assert "# Search Failed" in result
assert "Error searching for 'test query': unknown error" in result
assert "## Troubleshooting steps:" in result
class TestSearchToolErrorHandling:
"""Test search tool exception handling."""
@pytest.mark.asyncio
async def test_search_notes_exception_handling(self, monkeypatch):
"""Test exception handling in search_notes."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
# Mock SearchClient to raise an exception
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, *args, **kwargs):
raise Exception("syntax error")
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
# Patch at the clients module level where the import happens
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
result = await search_mod.search_notes(project="test-project", query="test query")
assert isinstance(result, str)
assert "# Search Failed - Invalid Syntax" in result
@pytest.mark.asyncio
async def test_search_notes_permission_error(self, monkeypatch):
"""Test search_notes with permission error."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
# Mock SearchClient to raise a permission error
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, *args, **kwargs):
raise Exception("permission denied")
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
# Patch at the clients module level where the import happens
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
result = await search_mod.search_notes(project="test-project", query="test query")
assert isinstance(result, str)
assert "# Search Failed - Access Error" in result
@pytest.mark.asyncio
@pytest.mark.parametrize("search_type", ["vector", "semantic", "hybrid"])
async def test_search_notes_sets_retrieval_mode_for_semantic_types(monkeypatch, search_type):
"""Vector/hybrid search types should populate retrieval_mode in API payload."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
project_url = "http://test"
name = "test-project"
id = 1
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
result = await search_mod.search_notes(
project="test-project",
query="semantic lookup",
search_type=search_type,
)
# Default text format returns a formatted string for empty results
assert isinstance(result, str)
assert captured_payload["text"] == "semantic lookup"
# "semantic" is an alias for "vector" retrieval mode
expected_mode = "vector" if search_type in ("vector", "semantic") else search_type
assert captured_payload["retrieval_mode"] == expected_mode
# --- Tests for metadata_filters / tags / status params (lines 440-444) ------
@pytest.mark.asyncio
async def test_search_notes_passes_metadata_filters(monkeypatch):
"""metadata_filters param propagates to the search query."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
await search_mod.search_notes(
project="test-project",
query="test",
metadata_filters={"status": "active"},
tags=["important"],
status="published",
)
assert captured_payload["metadata_filters"] == {"status": "active"}
assert captured_payload["tags"] == ["important"]
assert captured_payload["status"] == "published"
# --- Tests for filter-only search (query=None) --------------------------------
@pytest.mark.asyncio
async def test_search_notes_filter_only_metadata(monkeypatch):
"""search_notes with metadata_filters only (no query) sends correct payload."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
result = await search_mod.search_notes(
project="test-project",
metadata_filters={"status": "in-progress"},
)
# Default text format returns a formatted string for empty results
assert isinstance(result, str)
assert captured_payload["metadata_filters"] == {"status": "in-progress"}
# No text/title/permalink should be set
assert captured_payload.get("text") is None
assert captured_payload.get("title") is None
assert captured_payload.get("permalink") is None
@pytest.mark.asyncio
async def test_search_notes_filter_only_tags(monkeypatch):
"""search_notes with tags only (no query) sends correct payload."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
result = await search_mod.search_notes(
project="test-project",
tags=["security", "oauth"],
)
# Default text format returns a formatted string for empty results
assert isinstance(result, str)
assert captured_payload["tags"] == ["security", "oauth"]
assert captured_payload.get("text") is None
@pytest.mark.asyncio
async def test_search_notes_no_criteria_returns_error(monkeypatch):
"""search_notes with no args at all returns a helpful error string."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
result = await search_mod.search_notes(project="test-project")
assert isinstance(result, str)
assert "No Search Criteria" in result
@pytest.mark.asyncio
async def test_search_notes_invalid_search_type_returns_error(monkeypatch):
"""Invalid search_type values should return an error message listing valid options."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, *args, **kwargs):
pytest.fail("SearchClient.search should not be called for invalid search_type")
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
result = await search_mod.search_notes(
project="test-project",
query="test query",
search_type="bogus",
)
# The ValueError is caught by the generic exception handler and formatted
assert isinstance(result, str)
assert "Invalid search_type" in result
assert "bogus" in result
@pytest.mark.asyncio
async def test_search_notes_passes_min_similarity(monkeypatch):
"""min_similarity param propagates to the SearchQuery payload."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
await search_mod.search_notes(
project="test-project",
query="test",
search_type="vector",
min_similarity=0.0,
)
assert captured_payload["min_similarity"] == 0.0
assert captured_payload["retrieval_mode"] == "vector"
@pytest.mark.asyncio
async def test_search_notes_defaults_to_hybrid_when_semantic_enabled(monkeypatch):
"""When search_type is omitted, semantic-enabled configs should default to hybrid."""
import importlib
from dataclasses import dataclass
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
# Stub get_container to return a config with semantic_search_enabled=True
@dataclass
class StubConfig:
semantic_search_enabled: bool = True
default_search_type: str | None = None
@dataclass
class StubContainer:
config: StubConfig | None = None
def __post_init__(self):
if self.config is None:
self.config = StubConfig()
monkeypatch.setattr(search_mod, "get_container", lambda: StubContainer())
await search_mod.search_notes(
project="test-project",
query="test query",
)
# Default mode should be hybrid when semantic search is enabled
assert captured_payload["retrieval_mode"] == "hybrid"
assert captured_payload["text"] == "test query"
@pytest.mark.asyncio
async def test_search_notes_defaults_to_fts_when_semantic_disabled(monkeypatch):
"""When search_type is omitted, semantic-disabled configs should default to FTS."""
import importlib
from dataclasses import dataclass
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
# Stub get_container to return a config with semantic_search_enabled=False
@dataclass
class StubConfig:
semantic_search_enabled: bool = False
default_search_type: str | None = None
@dataclass
class StubContainer:
config: StubConfig | None = None
def __post_init__(self):
if self.config is None:
self.config = StubConfig()
monkeypatch.setattr(search_mod, "get_container", lambda: StubContainer())
await search_mod.search_notes(
project="test-project",
query="test query",
)
# Default mode should be FTS when semantic search is disabled
assert captured_payload["retrieval_mode"] == "fts"
assert captured_payload["text"] == "test query"
@pytest.mark.asyncio
async def test_search_notes_explicit_text_stays_fts_when_semantic_enabled(monkeypatch):
"""Explicit text mode should preserve FTS behavior even when semantic is enabled."""
import importlib
from dataclasses import dataclass
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
@dataclass
class StubConfig:
semantic_search_enabled: bool = True
default_search_type: str | None = None
@dataclass
class StubContainer:
config: StubConfig | None = None
def __post_init__(self):
if self.config is None:
self.config = StubConfig()
monkeypatch.setattr(search_mod, "get_container", lambda: StubContainer())
await search_mod.search_notes(
project="test-project",
query="test query",
search_type="text",
)
assert captured_payload["retrieval_mode"] == "fts"
assert captured_payload["text"] == "test query"
@pytest.mark.asyncio
async def test_search_notes_defaults_to_hybrid_when_container_not_initialized(monkeypatch):
"""CLI fallback config should still default omitted search_type to hybrid."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
# Stub get_container to raise RuntimeError (container not initialized)
def raise_runtime_error():
raise RuntimeError("MCP container not initialized")
monkeypatch.setattr(search_mod, "get_container", raise_runtime_error)
monkeypatch.setattr(
search_mod,
"ConfigManager",
lambda: type(
"StubConfigManager",
(),
{
"config": type(
"Cfg", (), {"semantic_search_enabled": True, "default_search_type": None}
)()
},
)(),
)
await search_mod.search_notes(
project="test-project",
query="test query",
)
# Should upgrade using ConfigManager fallback
assert captured_payload["retrieval_mode"] == "hybrid"
assert captured_payload["text"] == "test query"
@pytest.mark.asyncio
async def test_search_notes_defaults_to_fts_when_container_not_initialized_and_semantic_disabled(
monkeypatch,
):
"""CLI fallback config should default omitted search_type to FTS when semantic is disabled."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
def raise_runtime_error():
raise RuntimeError("MCP container not initialized")
monkeypatch.setattr(search_mod, "get_container", raise_runtime_error)
monkeypatch.setattr(
search_mod,
"ConfigManager",
lambda: type(
"StubConfigManager",
(),
{
"config": type(
"Cfg", (), {"semantic_search_enabled": False, "default_search_type": None}
)()
},
)(),
)
await search_mod.search_notes(
project="test-project",
query="test query",
)
assert captured_payload["retrieval_mode"] == "fts"
assert captured_payload["text"] == "test query"
# --- Tests for default entity_types (issue #31) --------------------------------
@pytest.mark.asyncio
async def test_search_notes_defaults_entity_types_to_entity(monkeypatch):
"""search_notes defaults entity_types to ['entity'] when not explicitly provided.
This prevents individual observations/relations from appearing as separate
search results, since the entity row already indexes full file content.
"""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
await search_mod.search_notes(
project="test-project",
query="test",
)
# entity_types should default to ["entity"]
assert captured_payload["entity_types"] == ["entity"]
@pytest.mark.asyncio
async def test_search_notes_explicit_entity_types_overrides_default(monkeypatch):
"""Explicit entity_types parameter overrides the default ['entity'] filter."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
await search_mod.search_notes(
project="test-project",
query="test",
entity_types=["observation"],
)
# Explicit entity_types should be used, not the default
assert captured_payload["entity_types"] == ["observation"]
# --- Tests for note_types case-insensitivity ------------------------------------
@pytest.mark.asyncio
async def test_search_notes_note_types_lowercased(monkeypatch):
"""note_types values are lowercased so 'Chapter' matches stored 'chapter'."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
await search_mod.search_notes(
project="test-project",
query="test",
note_types=["Chapter", "Person"],
)
# note_types should be lowercased
assert captured_payload["note_types"] == ["chapter", "person"]
# --- Tests for tag: prefix parsing (issue #30) ---------------------------------
@pytest.mark.asyncio
async def test_search_notes_tag_prefix_converts_to_tags_filter(monkeypatch):
"""query='tag:security' should be converted to a tags filter with no text query."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
result = await search_mod.search_notes(
project="test-project",
query="tag:security",
)
# Default text format returns a formatted string for empty results
assert isinstance(result, str)
assert captured_payload["tags"] == ["security"]
# No text query should be set — tag: prefix was consumed
assert captured_payload.get("text") is None
@pytest.mark.asyncio
async def test_search_notes_tag_prefix_merges_with_explicit_tags(monkeypatch):
"""query='tag:security' with tags=['oauth'] should merge both tag values."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
result = await search_mod.search_notes(
project="test-project",
query="tag:security",
tags=["oauth"],
)
# Default text format returns a formatted string for empty results
assert isinstance(result, str)
assert set(captured_payload["tags"]) == {"security", "oauth"}
assert captured_payload.get("text") is None
@pytest.mark.asyncio
async def test_search_notes_multiple_tag_prefixes(monkeypatch):
"""query='tag:coffee AND tag:brewing' should extract both tags."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
result = await search_mod.search_notes(
project="test-project",
query="tag:coffee AND tag:brewing",
)
# Default text format returns a formatted string for empty results
assert isinstance(result, str)
assert set(captured_payload["tags"]) == {"coffee", "brewing"}
# Boolean connector AND should be stripped, leaving no text query
assert captured_payload.get("text") is None
@pytest.mark.asyncio
async def test_search_notes_tag_prefix_with_remaining_text(monkeypatch):
"""query='authentication tag:security' should keep text and extract tag."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
# Remaining text query triggers resolve_project_and_path, so stub it too
async def fake_resolve(client, query, project, context):
return project, query, False
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
result = await search_mod.search_notes(
project="test-project",
query="authentication tag:security",
)
# Default text format returns a formatted string for empty results
assert isinstance(result, str)
assert captured_payload["tags"] == ["security"]
# Remaining text should be preserved as the query
assert captured_payload["text"] == "authentication"
# --- Tests for comma-separated tags parameter (#910) ----------------------------
def test_search_notes_tags_annotation_splits_comma_strings():
"""The tags parameter annotation must parse every documented input form (#910).
Direct function calls bypass the BeforeValidator, so validate through the same
Annotated metadata pydantic applies on the MCP path. coerce_list wrapped a bare
comma string as the single literal tag ["a,b"]; parse_tags splits it like the
tag: query shorthand and write_note's tags convention.
"""
annotation = inspect.signature(search_notes).parameters["tags"].annotation
adapter = TypeAdapter(annotation)
real_list = adapter.validate_python(["a", "b"])
comma_string = adapter.validate_python("a,b")
json_string = adapter.validate_python('["a", "b"]')
single_string = adapter.validate_python("a")
assert real_list == ["a", "b"]
# The comma string and the real list must behave identically (the #910 bug).
assert comma_string == real_list
assert json_string == real_list
assert single_string == ["a"]
@pytest.mark.asyncio
async def test_search_notes_tags_comma_string_filters_via_mcp(mcp, client, test_project):
"""tags="alpha,beta" through the real MCP layer must match like a real list (#910)."""
from fastmcp import Client
async with Client(mcp) as mcp_client:
await mcp_client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "Tag Split Note",
"directory": "test",
"content": "# Tag Split Note\nTagSplitToken body",
"tags": ["alpha", "beta"],
},
)
async def found(tags_value: object) -> bool:
result = await mcp_client.call_tool(
"search_notes",
{
"project": test_project.name,
"query": "TagSplitToken",
"search_type": "text",
"tags": tags_value,
},
)
return "Tag Split Note" in result.content[0].text
as_list = await found(["alpha", "beta"])
as_comma_string = await found("alpha,beta")
as_json_string = await found('["alpha", "beta"]')
as_single_string = await found("alpha")
assert as_list, "real-list tags must match (sanity)"
assert as_comma_string == as_list, "comma string must behave like the real list"
assert as_json_string == as_list
assert as_single_string == as_list
# Negative control: the filter is actually applied, not silently dropped.
assert not await found("gamma")
def test_search_notes_tags_annotation_rejects_non_string_types():
"""Unsupported tag types must fail validation, not be stringified (#932 follow-up).
Bare parse_tags coerces anything to strings (42 -> ["42"], {"a": 1} -> junk tags),
silently turning caller mistakes into no-result searches. The strict_search_tags
wrapper only normalizes str/list/None and lets Pydantic reject everything else.
"""
from pydantic import ValidationError
annotation = inspect.signature(search_notes).parameters["tags"].annotation
adapter = TypeAdapter(annotation)
with pytest.raises(ValidationError):
adapter.validate_python(42)
with pytest.raises(ValidationError):
adapter.validate_python({"a": 1})
# Lists with non-string elements must also fail, not be stringified ([42] -> ["42"]).
with pytest.raises(ValidationError):
adapter.validate_python([42])
with pytest.raises(ValidationError):
adapter.validate_python([{"a": 1}])
with pytest.raises(ValidationError):
adapter.validate_python(["ok", 42])
# JSON-array strings with non-string elements must fail the same way — parse_tags
# would otherwise recursively stringify them before Pydantic validates List[str].
with pytest.raises(ValidationError):
adapter.validate_python("[42]")
with pytest.raises(ValidationError):
adapter.validate_python('[{"a": 1}]')
with pytest.raises(ValidationError):
adapter.validate_python('["ok", 42]')
# All-string lists and all-string JSON-array strings remain valid.
assert adapter.validate_python(["a", "b"]) == ["a", "b"]
assert adapter.validate_python('["a","b"]') == ["a", "b"]
# None stays a valid "no filter" input.
assert adapter.validate_python(None) in (None, [])
@pytest.mark.asyncio
async def test_search_notes_tags_invalid_type_rejected_via_mcp(mcp, client, test_project):
"""tags=42 through the real MCP layer must raise a validation error (#932 follow-up)."""
from fastmcp import Client
from fastmcp.exceptions import ToolError
async with Client(mcp) as mcp_client:
with pytest.raises(ToolError):
await mcp_client.call_tool(
"search_notes",
{
"project": test_project.name,
"query": "anything",
"tags": 42,
},
)
with pytest.raises(ToolError):
await mcp_client.call_tool(
"search_notes",
{
"project": test_project.name,
"query": "anything",
"tags": {"a": 1},
},
)
# Lists with non-string elements must be rejected too, not stringified.
with pytest.raises(ToolError):
await mcp_client.call_tool(
"search_notes",
{
"project": test_project.name,
"query": "anything",
"tags": [42],
},
)
with pytest.raises(ToolError):
await mcp_client.call_tool(
"search_notes",
{
"project": test_project.name,
"query": "anything",
"tags": [{"a": 1}],
},
)
with pytest.raises(ToolError):
await mcp_client.call_tool(
"search_notes",
{
"project": test_project.name,
"query": "anything",
"tags": ["ok", 42],
},
)
# JSON-array strings with non-string elements (clients that serialize arrays as
# strings) must be rejected too, not recursively stringified by parse_tags.
with pytest.raises(ToolError):
await mcp_client.call_tool(
"search_notes",
{
"project": test_project.name,
"query": "anything",
"tags": "[42]",
},
)
with pytest.raises(ToolError):
await mcp_client.call_tool(
"search_notes",
{
"project": test_project.name,
"query": "anything",
"tags": '[{"a": 1}]',
},
)
with pytest.raises(ToolError):
await mcp_client.call_tool(
"search_notes",
{
"project": test_project.name,
"query": "anything",
"tags": '["ok", 42]',
},
)
# Sanity: a valid all-string JSON-array string is still accepted.
await mcp_client.call_tool(
"search_notes",
{
"project": test_project.name,
"query": "anything",
"tags": '["a","b"]',
},
)
@pytest.mark.asyncio
async def test_search_notes_direct_call_splits_comma_tags(client, test_project):
"""Direct callers bypass the BeforeValidator, so the body must normalize tags.
Regression for the CLI path: `bm tool search-notes --tag alpha,beta` calls this
function directly with Typer's collected list ["alpha,beta"], which must split
into ["alpha", "beta"] instead of matching nothing (#910, #932 follow-up).
"""
await write_note(
project=test_project.name,
title="Direct Tag Split Note",
directory="test",
content="# Direct Tag Split Note\nDirectTagToken body",
tags=["alpha", "beta"],
)
async def found(tags_value: list[str] | None) -> bool:
result = await search_notes(
project=test_project.name,
query="DirectTagToken",
search_type="text",
output_format="json",
tags=tags_value,
)
assert isinstance(result, dict), f"search failed: {result}"
return any(r["title"] == "Direct Tag Split Note" for r in result["results"])
assert await found(["alpha"]), "plain tag list must match (sanity)"
# The CLI regression: Typer collects --tag alpha,beta as the single element "alpha,beta".
assert await found(["alpha,beta"])
# Negative control: the filter is still applied.
assert not await found(["gamma"])
# --- Tests for text output format (#641) -----------------------------------
def test_format_search_markdown_with_results():
"""_format_search_markdown returns readable markdown for non-empty results."""
from basic_memory.schemas.search import SearchResult, SearchItemType
result = SearchResponse(
results=[
SearchResult(
title="My Note",
type=SearchItemType.ENTITY,
score=0.85,
permalink="docs/my-note",
file_path="docs/My Note.md",
matched_chunk="This is a matching snippet",
),
SearchResult(
title="Other Note",
type=SearchItemType.ENTITY,
score=0.42,
permalink="docs/other-note",
file_path="docs/Other Note.md",
),
],
current_page=1,
page_size=10,
)
text = _format_search_markdown(result, "test-project", "my query")
assert isinstance(text, str)
assert "# Search Results: my query" in text
assert "test-project" in text
assert "### My Note" in text
assert "permalink: docs/my-note" in text
assert "0.8500" in text
assert "match: This is a matching snippet" in text
assert "### Other Note" in text
assert "2 results" in text
assert "page 1" in text
def test_format_search_markdown_empty_results():
"""_format_search_markdown returns a no-results message when results are empty."""
result = SearchResponse(results=[], current_page=1, page_size=10)
text = _format_search_markdown(result, "test-project", "missing")
assert isinstance(text, str)
assert "No results found" in text
assert "missing" in text
@pytest.mark.asyncio
async def test_search_notes_text_format_returns_string(monkeypatch):
"""search_notes with output_format='text' returns a formatted markdown string."""
import importlib
from basic_memory.schemas.search import SearchResult, SearchItemType
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
return SearchResponse(
results=[
SearchResult(
title="Found Note",
type=SearchItemType.ENTITY,
score=0.9,
permalink="docs/found-note",
file_path="docs/Found Note.md",
matched_chunk="snippet",
),
],
current_page=page,
page_size=page_size,
)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
result = await search_mod.search_notes(
project="test-project",
query="test",
output_format="text",
)
assert isinstance(result, str)
assert "# Search Results: test" in result
assert "### Found Note" in result
assert "permalink: docs/found-note" in result
# --- Tests for metadata_filters key aliasing (#642) ----------------------------
@pytest.mark.asyncio
async def test_search_notes_metadata_filters_aliases_note_type(monkeypatch):
"""metadata_filters={'note_type': 'note'} is aliased to {'type': 'note'}."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
await search_mod.search_notes(
project="test-project",
query="test",
metadata_filters={"note_type": "note"},
)
# "note_type" should be aliased to "type" in the payload
assert captured_payload["metadata_filters"] == {"type": "note"}
@pytest.mark.asyncio
async def test_search_notes_metadata_filters_preserves_non_aliased_keys(monkeypatch):
"""metadata_filters with non-aliased keys pass through unchanged."""
import importlib
search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject:
name = "test-project"
external_id = "test-external-id"
@asynccontextmanager
async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
captured_payload: dict = {}
class MockSearchClient:
def __init__(self, *args, **kwargs):
pass
async def search(self, payload, page, page_size):
captured_payload.update(payload)
return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
await search_mod.search_notes(
project="test-project",
query="test",
metadata_filters={"note_type": "spec", "priority": "high"},
)
# "note_type" aliased to "type", "priority" passes through unchanged
assert captured_payload["metadata_filters"] == {"type": "spec", "priority": "high"}
def test_default_search_type_uses_config_value():
"""_default_search_type should return config.default_search_type when set."""
import sys
from unittest.mock import MagicMock, patch
search_module = sys.modules["basic_memory.mcp.tools.search"]
mock_config = MagicMock()
mock_config.default_search_type = "vector"
mock_config.semantic_search_enabled = True
mock_container = MagicMock()
mock_container.config = mock_config
with patch.object(search_module, "get_container", return_value=mock_container):
assert search_module._default_search_type() == "vector"
def test_default_search_type_falls_back_to_hybrid_when_semantic_enabled():
"""When default_search_type is None and semantic is enabled, default to hybrid."""
import sys
from unittest.mock import MagicMock, patch
search_module = sys.modules["basic_memory.mcp.tools.search"]
mock_config = MagicMock()
mock_config.default_search_type = None
mock_config.semantic_search_enabled = True
mock_container = MagicMock()
mock_container.config = mock_config
with patch.object(search_module, "get_container", return_value=mock_container):
assert search_module._default_search_type() == "hybrid"
def test_default_search_type_falls_back_to_text_when_semantic_disabled():
"""When default_search_type is None and semantic is disabled, default to text."""
import sys
from unittest.mock import MagicMock, patch
search_module = sys.modules["basic_memory.mcp.tools.search"]
mock_config = MagicMock()
mock_config.default_search_type = None
mock_config.semantic_search_enabled = False
mock_container = MagicMock()
mock_container.config = mock_config
with patch.object(search_module, "get_container", return_value=mock_container):
assert search_module._default_search_type() == "text"
# --- Tests for note_types/entity_types/categories comma-split fix (#930, Codex review) ---
def test_search_notes_note_types_annotation_splits_comma_strings():
"""The note_types parameter annotation must parse every documented input form (#930).
Direct function calls bypass the BeforeValidator; validate through the same
Annotated metadata pydantic applies on the MCP path. The old coerce_list wrapped a
bare comma string as the single literal type ["note,task"]; parse_str_list splits it.
"""
annotation = inspect.signature(search_notes).parameters["note_types"].annotation
adapter = TypeAdapter(annotation)
real_list = adapter.validate_python(["note", "task"])
comma_string = adapter.validate_python("note,task")
json_string = adapter.validate_python('["note", "task"]')
single_string = adapter.validate_python("note")
comma_in_list = adapter.validate_python(["note,task"])
assert real_list == ["note", "task"]
assert comma_string == real_list, "comma string must behave like the real list"
assert json_string == real_list
assert single_string == ["note"]
assert comma_in_list == real_list, "list with comma element must be flattened"
def test_search_notes_entity_types_annotation_splits_comma_strings():
"""The entity_types parameter annotation must parse every documented input form (#930)."""
annotation = inspect.signature(search_notes).parameters["entity_types"].annotation
adapter = TypeAdapter(annotation)
real_list = adapter.validate_python(["entity", "observation"])
comma_string = adapter.validate_python("entity,observation")
comma_in_list = adapter.validate_python(["entity,observation"])
assert real_list == ["entity", "observation"]
assert comma_string == real_list
assert comma_in_list == real_list
def test_search_notes_categories_annotation_splits_comma_strings():
"""The categories parameter annotation must parse every documented input form (#930)."""
annotation = inspect.signature(search_notes).parameters["categories"].annotation
adapter = TypeAdapter(annotation)
real_list = adapter.validate_python(["requirement", "decision"])
comma_string = adapter.validate_python("requirement,decision")
comma_in_list = adapter.validate_python(["requirement,decision"])
assert real_list == ["requirement", "decision"]
assert comma_string == real_list
assert comma_in_list == real_list
def test_search_notes_note_types_annotation_rejects_non_string_list_elements():
"""note_types=[42] must fail Pydantic validation, not be stringified to ['42'].
parse_str_list used str(raw) to coerce list elements, silently accepting [42] as
["42"]. The fix guards against non-string list elements and returns the original
value so Pydantic rejects it with a clear error.
"""
from pydantic import ValidationError
annotation = inspect.signature(search_notes).parameters["note_types"].annotation
adapter = TypeAdapter(annotation)
with pytest.raises(ValidationError):
adapter.validate_python([42])
with pytest.raises(ValidationError):
adapter.validate_python(["note", 42])
# All-string lists remain valid.
assert adapter.validate_python(["note", "task"]) == ["note", "task"]
def test_search_notes_entity_types_annotation_rejects_non_string_list_elements():
"""entity_types=[42] must fail Pydantic validation, not be stringified."""
from pydantic import ValidationError
annotation = inspect.signature(search_notes).parameters["entity_types"].annotation
adapter = TypeAdapter(annotation)
with pytest.raises(ValidationError):
adapter.validate_python([42])
with pytest.raises(ValidationError):
adapter.validate_python(["entity", 42])
assert adapter.validate_python(["entity", "observation"]) == ["entity", "observation"]
def test_search_notes_categories_annotation_rejects_non_string_list_elements():
"""categories=[42] must fail Pydantic validation, not be stringified."""
from pydantic import ValidationError
annotation = inspect.signature(search_notes).parameters["categories"].annotation
adapter = TypeAdapter(annotation)
with pytest.raises(ValidationError):
adapter.validate_python([42])
with pytest.raises(ValidationError):
adapter.validate_python(["requirement", 42])
assert adapter.validate_python(["requirement", "decision"]) == ["requirement", "decision"]
@pytest.mark.asyncio
async def test_search_notes_direct_call_splits_comma_note_types(client, test_project):
"""Direct callers bypass the BeforeValidator, so the body must normalize note_types.
Regression for the CLI path: `bm tool search-notes --type note,task` calls this
function directly with Typer's collected list ["note,task"], which must split into
["note", "task"] and match the note correctly (#930, Codex review follow-up).
"""
await write_note(
project=test_project.name,
title="Direct NoteType Split Note",
directory="test",
content="# Direct NoteType Split Note\nNoteTypeSplitToken body",
)
async def found(note_types_value: list[str] | None) -> bool:
result = await search_notes(
project=test_project.name,
query="NoteTypeSplitToken",
search_type="text",
output_format="json",
note_types=note_types_value,
)
assert isinstance(result, dict), f"search failed: {result}"
return any(r["title"] == "Direct NoteType Split Note" for r in result["results"])
assert await found(None), "no filter must match (sanity)"
assert await found(["note"]), "plain single-type list must match (sanity)"
# The CLI regression: Typer collects --type note,task as the single element "note,task".
assert await found(["note,task"]), "comma list element must be flattened and match 'note'"
# Negative control: a specific nonexistent type must not match.
assert not await found(["nonexistent_type"])
def test_search_notes_parse_str_list_rejects_non_string_list_elements_in_place():
"""parse_str_list must return non-str list elements unchanged for Pydantic rejection.
The old implementation used str(raw) which silently coerced [42] -> ['42'],
causing bad caller data to become silent no-result searches instead of a
clear Pydantic validation error.
"""
from basic_memory.utils import parse_str_list
# Non-string list elements pass through unchanged.
assert parse_str_list([42]) == [42] # type: ignore[arg-type]
assert parse_str_list(["ok", 42]) == ["ok", 42] # type: ignore[arg-type]
assert parse_str_list([{"a": 1}]) == [{"a": 1}] # type: ignore[arg-type]
# All-string lists still work correctly.
assert parse_str_list(["note", "task"]) == ["note", "task"]
assert parse_str_list(["note,task"]) == ["note", "task"]
Reference in New Issue View Git Blame Copy Permalink