"""Tests for search MCP tools.""" import pytest from contextlib import asynccontextmanager from datetime import datetime, timedelta from types import SimpleNamespace from typing import cast 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 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"