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

fix: three cloud-testing bugs (#640, #641, #642)

Browse Source 
🔧 #640 — LinkResolver selects worst match instead of best
Replace `min(results, key=lambda x: x.score)` with `results[0]`.
Both SQLite and Postgres return results sorted best-first in SQL,
so using `results[0]` is backend-agnostic and correct.

🔧 #641 — search_notes output_format="text" returns raw Pydantic model
Add `_format_search_markdown()` that formats SearchResponse as readable
markdown with title, permalink, score, and matched snippet per result.
Update prompts to use `output_format="json"` since they need structured
data for result counting and branching logic.

🔧 #642 — metadata_filters with `note_type` key returns empty results
Add `_METADATA_KEY_ALIASES` mapping at the tool level that aliases
`note_type` → `type` before passing metadata_filters to the search query.
The frontmatter field is `type`, not `note_type`.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>
This commit is contained in:
phernandez
2026-03-02 12:58:22 -06:00
parent 8b7f39ee9b
commit 7f2d4d2a6f
9 changed files with 431 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic_memory/mcp/prompts/continue_conversation.py
+13 -16
View File
@@ -13,7 +13,6 @@ from pydantic import Field
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.recent_activity import recent_activity
from basic_memory.mcp.tools.search import search_notes
from basic_memory.schemas.search import SearchResponse
@mcp.prompt(
@@ -42,15 +41,12 @@ async def continue_conversation(
logger.info(f"Continuing session, topic: {topic}, timeframe: {timeframe}")
if topic:
# Search for the topic using the search tool directly
result = await search_notes(query=topic, after_date=timeframe)
# Use json format to get structured data for result counting and branching
result = await search_notes(query=topic, after_date=timeframe, output_format="json")
if isinstance(result, SearchResponse):
context_text = _format_continuation_results(result, topic)
result_count = len(result.results)
elif isinstance(result, dict):
if isinstance(result, dict):
results = result.get("results", [])
context_text = str(result)
context_text = _format_continuation_results(results, topic)
result_count = len(results)
else:
# Error string
@@ -111,23 +107,24 @@ async def continue_conversation(
return prompt
def _format_continuation_results(result: SearchResponse, topic: str) -> str:
"""Format search results for conversation continuation context."""
if not result.results:
def _format_continuation_results(results: list[dict], topic: str) -> str:
"""Format search result dicts for conversation continuation context."""
if not results:
return f"No previous context found for '{topic}'."
lines = [f"## Previous Context for '{topic}'\n"]
for item in result.results:
title = item.title or "Untitled"
permalink = item.permalink or ""
for item in results:
title = item.get("title", "Untitled")
permalink = item.get("permalink", "")
lines.append(f"### {title}")
if permalink:
lines.append(f"permalink: {permalink}")
lines.append(f'Read with: `read_note("{permalink}")`')
if item.content:
content = item.content[:300] + "..." if len(item.content) > 300 else item.content
content = item.get("content")
if content:
content = content[:300] + "..." if len(content) > 300 else content
lines.append(f"\n{content}")
lines.append("")
src/basic_memory/mcp/prompts/search.py
+17 -23
View File
@@ -11,7 +11,6 @@ from pydantic import Field
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.search import search_notes
from basic_memory.schemas.search import SearchResponse
@mcp.prompt(
@@ -39,18 +38,14 @@ async def search_prompt(
"""
logger.info(f"Searching knowledge base, query: {query}, timeframe: {timeframe}")
# Call the search tool directly — it returns SearchResponse, dict, or error string
result = await search_notes(query=query, after_date=timeframe)
# Use json format to get structured data for result counting and formatting
result = await search_notes(query=query, after_date=timeframe, output_format="json")
# Format the tool output into a prompt with guidance
if isinstance(result, SearchResponse):
result_count = len(result.results)
result_text = _format_search_results(result, query)
elif isinstance(result, dict):
# json output format
if isinstance(result, dict):
results = result.get("results", [])
result_count = len(results)
result_text = str(result)
result_text = _format_search_results(results, query)
else:
# Error string from search tool
result_count = 0
@@ -76,28 +71,27 @@ async def search_prompt(
""")
def _format_search_results(result: SearchResponse, query: str) -> str:
"""Format SearchResponse into readable markdown."""
if not result.results:
def _format_search_results(results: list[dict], query: str) -> str:
"""Format search result dicts into readable markdown."""
if not results:
return f"No results found for '{query}'."
lines = [f"Found {len(result.results)} results:\n"]
lines = [f"Found {len(results)} results:\n"]
for item in result.results:
title = item.title or "Untitled"
permalink = item.permalink or ""
score = f" (score: {item.score:.2f})" if item.score else ""
for item in results:
title = item.get("title", "Untitled")
permalink = item.get("permalink", "")
score = item.get("score")
score_text = f" (score: {score:.2f})" if score else ""
lines.append(f"- **{title}**{score}")
lines.append(f"- **{title}**{score_text}")
if permalink:
lines.append(f" permalink: {permalink}")
if item.content:
content = item.get("content")
if content:
# Truncate content snippet
content = item.content[:200] + "..." if len(item.content) > 200 else item.content
content = content[:200] + "..." if len(content) > 200 else content
lines.append(f" {content}")
lines.append("")
if result.has_more:
lines.append("*More results available. Use page=2 to see next page.*")
return "\n".join(lines)
src/basic_memory/mcp/tools/search.py
+51 -3
View File
@@ -251,6 +251,46 @@ Error searching for '{query}': {error_message}
- **Patterns**: `tag:example`, `category:observation`"""
def _format_search_markdown(result: SearchResponse, project: str, query: str | None) -> str:
"""Format SearchResponse as compact markdown text.
Produces a human-readable markdown representation suitable for LLM
consumption when structured data isn't needed.
"""
if not result.results:
return f"No results found for '{query or ''}' in project '{project}'."
parts = []
# --- Header ---
if query:
parts.append(f"# Search Results: {query}")
else:
parts.append("# Search Results")
parts.append(f"*project: {project}*")
parts.append("")
# --- Result blocks ---
for r in result.results:
parts.append(f"### {r.title}")
parts.append(f"- permalink: {r.permalink}")
parts.append(f"- score: {r.score:.4f}")
if r.matched_chunk:
parts.append(f"- match: {r.matched_chunk[:200]}")
parts.append("")
# --- Footer with pagination ---
parts.append("---")
count = len(result.results)
parts.append(
f"*{count} result{'s' if count != 1 else ''}"
f" | page {result.current_page}, page_size {result.page_size}"
f"{' | more available' if result.has_more else ''}*"
)
return "\n".join(parts)
@mcp.tool(
description="Search across all content in the knowledge base with advanced syntax support.",
# TODO: re-enable once MCP client rendering is working
@@ -273,7 +313,7 @@ async def search_notes(
status: Optional[str] = None,
min_similarity: Optional[float] = None,
context: Context | None = None,
) -> SearchResponse | dict | str:
) -> dict | str:
"""Search across all content in the knowledge base with comprehensive syntax support.
This tool searches the knowledge base using full-text search, pattern matching,
@@ -373,7 +413,8 @@ async def search_notes(
context: Optional FastMCP context for performance caching.
Returns:
SearchResponse with results and pagination info, or helpful error guidance if search fails
Formatted markdown text (output_format="text"), dict (output_format="json"),
or helpful error guidance string if search fails
Examples:
# Basic text search
@@ -519,6 +560,13 @@ async def search_notes(
if after_date:
search_query.after_date = after_date
if metadata_filters:
# Alias common column/model names to their frontmatter key equivalents.
# Users often pass "note_type" (the entity model column) when the
# frontmatter field is actually "type".
_METADATA_KEY_ALIASES = {"note_type": "type"}
metadata_filters = {
_METADATA_KEY_ALIASES.get(k, k): v for k, v in metadata_filters.items()
}
search_query.metadata_filters = metadata_filters
if tags:
search_query.tags = tags
@@ -565,7 +613,7 @@ async def search_notes(
if output_format == "json":
return result.model_dump(mode="json", exclude_none=True)
return result
return _format_search_markdown(result, active_project.name, query)
except Exception as e:
logger.error(
src/basic_memory/services/link_resolver.py
+4 -2
View File
@@ -301,8 +301,10 @@ class LinkResolver:
)
if results:
# Look for best match
best_match = min(results, key=lambda x: x.score) # pyright: ignore
# Both SQLite and Postgres return results sorted best-first in SQL
# (SQLite: ORDER BY score ASC for negative BM25, Postgres: ORDER BY score DESC
# for positive ts_rank). Using results[0] is backend-agnostic and correct.
best_match = results[0]
logger.trace(
f"Selected best match from {len(results)} results: {best_match.permalink}"
)
test-int/mcp/test_delete_note_integration.py
+4 -10
View File
@@ -105,11 +105,8 @@ async def test_delete_note_by_permalink(mcp_server, app, test_project):
},
)
# Should have no results
assert (
'"results": []' in search_result.content[0].text
or '"results":[]' in search_result.content[0].text
)
# Default text format returns "No results found" when empty
assert "No results found" in search_result.content[0].text
@pytest.mark.asyncio
@@ -387,11 +384,8 @@ async def test_delete_multiple_notes_sequentially(mcp_server, app, test_project)
},
)
# Should have no results
assert (
'"results": []' in search_result.content[0].text
or '"results":[]' in search_result.content[0].text
)
# Default text format returns "No results found" when empty
assert "No results found" in search_result.content[0].text
@pytest.mark.asyncio
test-int/mcp/test_search_integration.py
+6 -5
View File
@@ -362,9 +362,9 @@ async def test_search_pagination(mcp_server, app, test_project):
)
result_text = search_result.content[0].text
# Should contain 5 results and pagination info
assert '"current_page":1' in result_text
assert '"page_size":5' in result_text
# Text format includes pagination info in footer
assert "page 1" in result_text
assert "page_size 5" in result_text
# Search page 2
search_result = await client.call_tool(
@@ -378,7 +378,7 @@ async def test_search_pagination(mcp_server, app, test_project):
)
result_text = search_result.content[0].text
assert '"current_page":2' in result_text
assert "page 2" in result_text
@pytest.mark.asyncio
@@ -407,8 +407,9 @@ async def test_search_no_results(mcp_server, app, test_project):
},
)
# Default text format returns "No results found" when empty
result_text = search_result.content[0].text
assert '"results": []' in result_text or '"results":[]' in result_text
assert "No results found" in result_text
@pytest.mark.asyncio
tests/mcp/test_prompt_tool_delegation.py
+30 -27
View File
@@ -9,7 +9,6 @@ import pytest
from basic_memory.mcp.prompts.search import search_prompt
from basic_memory.mcp.prompts.continue_conversation import continue_conversation
from basic_memory.schemas.search import SearchResponse, SearchResult
# --- search_prompt ---
@@ -20,19 +19,20 @@ async def test_search_prompt_delegates_to_search_notes(monkeypatch):
"""Search prompt should call search_notes tool and wrap output."""
captured_kwargs = {}
fake_result = SearchResponse(
results=[
SearchResult(
type="entity",
title="Test Note",
permalink="test-note",
file_path="test-note.md",
score=0.95,
)
# Prompts use output_format="json", so mock returns a dict
fake_result = {
"results": [
{
"type": "entity",
"title": "Test Note",
"permalink": "test-note",
"file_path": "test-note.md",
"score": 0.95,
}
],
current_page=1,
page_size=10,
)
"current_page": 1,
"page_size": 10,
}
async def fake_search_notes(**kwargs):
captured_kwargs.update(kwargs)
@@ -45,6 +45,7 @@ async def test_search_prompt_delegates_to_search_notes(monkeypatch):
# Verify delegation
assert captured_kwargs["query"] == "my query"
assert captured_kwargs["after_date"] == "1w"
assert captured_kwargs["output_format"] == "json"
# Verify output wrapping
assert 'Search Results: "my query"' in out
@@ -55,7 +56,7 @@ async def test_search_prompt_delegates_to_search_notes(monkeypatch):
@pytest.mark.asyncio
async def test_search_prompt_handles_no_results(monkeypatch):
"""Search prompt should handle empty results gracefully."""
fake_result = SearchResponse(results=[], current_page=1, page_size=10)
fake_result = {"results": [], "current_page": 1, "page_size": 10}
async def fake_search_notes(**kwargs):
return fake_result
@@ -91,19 +92,20 @@ async def test_continue_conversation_delegates_to_search_notes(monkeypatch):
"""Continue conversation with topic should call search_notes."""
captured_kwargs = {}
fake_result = SearchResponse(
results=[
SearchResult(
type="entity",
title="Previous Discussion",
permalink="discussions/previous",
file_path="discussions/previous.md",
score=0.9,
)
# Prompts use output_format="json", so mock returns a dict
fake_result = {
"results": [
{
"type": "entity",
"title": "Previous Discussion",
"permalink": "discussions/previous",
"file_path": "discussions/previous.md",
"score": 0.9,
}
],
current_page=1,
page_size=10,
)
"current_page": 1,
"page_size": 10,
}
async def fake_search_notes(**kwargs):
captured_kwargs.update(kwargs)
@@ -117,6 +119,7 @@ async def test_continue_conversation_delegates_to_search_notes(monkeypatch):
assert captured_kwargs["query"] == "my topic"
assert captured_kwargs["after_date"] == "3d"
assert captured_kwargs["output_format"] == "json"
assert "'my topic'" in out
assert "Previous Discussion" in out
@@ -164,7 +167,7 @@ async def test_continue_conversation_no_topic_default_timeframe(monkeypatch):
@pytest.mark.asyncio
async def test_continue_conversation_no_results_for_topic(monkeypatch):
"""Continue conversation should show capture opportunity when no results found."""
fake_result = SearchResponse(results=[], current_page=1, page_size=10)
fake_result = {"results": [], "current_page": 1, "page_size": 10}
async def fake_search_notes(**kwargs):
return fake_result
tests/mcp/test_tool_search.py
+288 -56
View File
@@ -5,7 +5,11 @@ from contextlib import asynccontextmanager
from datetime import datetime, timedelta
from basic_memory.mcp.tools import write_note
from basic_memory.mcp.tools.search import search_notes, _format_search_error_response
from basic_memory.mcp.tools.search import (
search_notes,
_format_search_error_response,
_format_search_markdown,
)
from basic_memory.schemas.search import SearchResponse
@@ -22,15 +26,18 @@ async def test_search_text(client, test_project):
)
assert result
# Search for it
response = await search_notes(project=test_project.name, query="searchable")
# Search for it (use json format to inspect structured results)
response = await search_notes(
project=test_project.name, query="searchable", output_format="json"
)
# Verify results - handle both success and error cases
if isinstance(response, SearchResponse):
# Success case - verify SearchResponse
assert len(response.results) > 0
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
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
@@ -50,21 +57,22 @@ async def test_search_title(client, test_project):
)
assert result
# Search for it
# Search for it (use json format to inspect structured results)
response = await search_notes(
project=test_project.name, query="Search Note", search_type="title"
project=test_project.name, query="Search Note", search_type="title", output_format="json"
)
# Verify results - handle both success and error cases
if isinstance(response, str):
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}")
else:
# Success case - verify SearchResponse
assert len(response.results) > 0
assert any(
r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results
)
@pytest.mark.asyncio
@@ -80,19 +88,21 @@ async def test_search_permalink(client, test_project):
)
assert result
# Search for it
# 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, SearchResponse):
# Success case - verify SearchResponse
assert len(response.results) > 0
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
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
@@ -112,19 +122,21 @@ async def test_search_permalink_match(client, test_project):
)
assert result
# Search for it
# 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, SearchResponse):
# Success case - verify SearchResponse
assert len(response.results) > 0
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
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
@@ -142,13 +154,15 @@ async def test_search_memory_url_with_project_prefix(client, test_project):
)
assert result
response = await search_notes(query=f"memory://{test_project.name}/test/memory-url-search-note")
response = await search_notes(
query=f"memory://{test_project.name}/test/memory-url-search-note", output_format="json"
)
if isinstance(response, SearchResponse):
assert len(response.results) > 0
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
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}")
@@ -167,17 +181,18 @@ async def test_search_pagination(client, test_project):
)
assert result
# Search for it
# Search for it (use json format to inspect structured results)
response = await search_notes(
project=test_project.name, query="searchable", page=1, page_size=1
project=test_project.name, query="searchable", page=1, page_size=1, output_format="json"
)
# Verify results - handle both success and error cases
if isinstance(response, SearchResponse):
# Success case - verify SearchResponse
assert len(response.results) == 1
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
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
@@ -195,13 +210,15 @@ async def test_search_with_type_filter(client, test_project):
content="# Test\nFiltered by type",
)
# Search with note type filter
response = await search_notes(project=test_project.name, query="type", note_types=["note"])
# Search with note type filter (use json format to inspect structured results)
response = await search_notes(
project=test_project.name, query="type", note_types=["note"], output_format="json"
)
# Verify results - handle both success and error cases
if isinstance(response, SearchResponse):
if isinstance(response, dict):
# Success case - verify all results are entities
assert all(r.type == "entity" for r in response.results)
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}")
@@ -218,13 +235,15 @@ async def test_search_with_entity_type_filter(client, test_project):
content="# Test\nFiltered by type",
)
# Search with entity_types (SearchItemType) filter
response = await search_notes(project=test_project.name, query="type", entity_types=["entity"])
# Search with entity_types (SearchItemType) filter (use json format)
response = await search_notes(
project=test_project.name, query="type", entity_types=["entity"], output_format="json"
)
# Verify results - handle both success and error cases
if isinstance(response, SearchResponse):
if isinstance(response, dict):
# Success case - verify all results are entities
assert all(r.type == "entity" for r in response.results)
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}")
@@ -241,16 +260,19 @@ async def test_search_with_date_filter(client, test_project):
content="# Test\nRecent content",
)
# Search with date filter
# 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", after_date=one_hour_ago.isoformat()
project=test_project.name,
query="recent",
after_date=one_hour_ago.isoformat(),
output_format="json",
)
# Verify results - handle both success and error cases
if isinstance(response, SearchResponse):
if isinstance(response, dict):
# Success case - verify we get results within timeframe
assert len(response.results) > 0
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}")
@@ -468,7 +490,8 @@ async def test_search_notes_sets_retrieval_mode_for_semantic_types(monkeypatch,
search_type=search_type,
)
assert isinstance(result, SearchResponse)
# 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
@@ -563,7 +586,8 @@ async def test_search_notes_filter_only_metadata(monkeypatch):
metadata_filters={"status": "in-progress"},
)
assert isinstance(result, SearchResponse)
# 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
@@ -605,7 +629,8 @@ async def test_search_notes_filter_only_tags(monkeypatch):
tags=["security", "oauth"],
)
assert isinstance(result, SearchResponse)
# 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
@@ -1158,7 +1183,8 @@ async def test_search_notes_tag_prefix_converts_to_tags_filter(monkeypatch):
query="tag:security",
)
assert isinstance(result, SearchResponse)
# 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
@@ -1199,7 +1225,8 @@ async def test_search_notes_tag_prefix_merges_with_explicit_tags(monkeypatch):
tags=["oauth"],
)
assert isinstance(result, SearchResponse)
# 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
@@ -1238,7 +1265,8 @@ async def test_search_notes_multiple_tag_prefixes(monkeypatch):
query="tag:coffee AND tag:brewing",
)
assert isinstance(result, SearchResponse)
# 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
@@ -1283,7 +1311,211 @@ async def test_search_notes_tag_prefix_with_remaining_text(monkeypatch):
query="authentication tag:security",
)
assert isinstance(result, SearchResponse)
# 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 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"}
tests/services/test_link_resolver.py
+18
View File
@@ -901,3 +901,21 @@ async def test_resolve_link_non_uuid_falls_through(link_resolver, test_entities,
result = await link_resolver.resolve_link("Core Service")
assert result is not None
assert result.permalink == f"{project_prefix}/components/core-service"
# ============================================================================
# Fuzzy search best-match selection tests (#640)
# ============================================================================
@pytest.mark.asyncio
async def test_fuzzy_search_selects_first_result(link_resolver, project_prefix):
"""Test that fuzzy search uses results[0] (best-ranked by the DB) regardless of score sign.
Both SQLite (BM25, negative scores, ASC) and Postgres (ts_rank, positive scores, DESC)
return the best match first. Using results[0] is backend-agnostic and correct.
"""
result = await link_resolver.resolve_link("Auth Serv")
assert result is not None
# The best match for "Auth Serv" should be Auth Service
assert result.permalink == f"{project_prefix}/components/auth-service"