"""Tests for the SearchRepository."""
from datetime import datetime, timezone
import pytest
import pytest_asyncio
from sqlalchemy import text
from basic_memory import db
from basic_memory.models import Entity
from basic_memory.models.project import Project
from basic_memory.repository.search_repository import SearchIndexRow
from basic_memory.repository.postgres_search_repository import PostgresSearchRepository
from basic_memory.schemas.search import SearchItemType
def is_postgres_backend(search_repository):
"""Helper to check if search repository is Postgres-based."""
return isinstance(search_repository, PostgresSearchRepository)
@pytest_asyncio.fixture
async def search_entity(session_maker, test_project: Project):
"""Create a test entity for search testing."""
async with db.scoped_session(session_maker) as session:
entity = Entity(
project_id=test_project.id,
title="Search Test Entity",
entity_type="test",
permalink="test/search-test-entity",
file_path="test/search_test_entity.md",
content_type="text/markdown",
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
session.add(entity)
await session.flush()
return entity
@pytest_asyncio.fixture
async def second_project(project_repository):
"""Create a second project for testing project isolation."""
project_data = {
"name": "Second Test Project",
"description": "Another project for testing",
"path": "/second/project/path",
"is_active": True,
"is_default": None,
}
return await project_repository.create(project_data)
@pytest_asyncio.fixture
async def second_project_repository(session_maker, second_project, search_repository):
"""Create a backend-appropriate repository for the second project.
Uses the same type as search_repository to ensure backend consistency.
"""
# Use the same repository class as the main search_repository
return type(search_repository)(session_maker, project_id=second_project.id)
@pytest_asyncio.fixture
async def second_entity(session_maker, second_project: Project):
"""Create a test entity in the second project."""
async with db.scoped_session(session_maker) as session:
entity = Entity(
project_id=second_project.id,
title="Second Project Entity",
entity_type="test",
permalink="test/second-project-entity",
file_path="test/second_project_entity.md",
content_type="text/markdown",
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
session.add(entity)
await session.flush()
return entity
@pytest.mark.asyncio
async def test_init_search_index(search_repository, app_config):
"""Test that search index can be initialized."""
from basic_memory.config import DatabaseBackend
await search_repository.init_search_index()
# Verify search_index table exists (backend-specific query)
async with db.scoped_session(search_repository.session_maker) as session:
if app_config.database_backend == DatabaseBackend.POSTGRES:
# For Postgres, query information_schema
result = await session.execute(
text(
"SELECT table_name FROM information_schema.tables "
"WHERE table_schema = 'public' AND table_name = 'search_index';"
)
)
else:
# For SQLite, query sqlite_master
result = await session.execute(
text("SELECT name FROM sqlite_master WHERE type='table' AND name='search_index';")
)
table_name = result.scalar()
assert table_name == "search_index"
@pytest.mark.asyncio
async def test_index_item(search_repository, search_entity):
"""Test indexing an item with project_id."""
# Create search index row for the entity
search_row = SearchIndexRow(
id=search_entity.id,
type=SearchItemType.ENTITY.value,
title=search_entity.title,
content_stems="search test entity content",
content_snippet="This is a test entity for search",
permalink=search_entity.permalink,
file_path=search_entity.file_path,
entity_id=search_entity.id,
metadata={"entity_type": search_entity.entity_type},
created_at=search_entity.created_at,
updated_at=search_entity.updated_at,
project_id=search_repository.project_id,
)
# Index the item
await search_repository.index_item(search_row)
# Search for the item
results = await search_repository.search(search_text="search test")
# Verify we found the item
assert len(results) == 1
assert results[0].title == search_entity.title
assert results[0].project_id == search_repository.project_id
@pytest.mark.asyncio
async def test_index_item_upsert_on_duplicate_permalink(search_repository, search_entity):
"""Test that indexing the same permalink twice uses upsert instead of failing.
This tests the fix for the race condition where parallel entity indexing
could cause IntegrityError on the unique permalink constraint.
"""
# First insert
search_row1 = SearchIndexRow(
id=search_entity.id,
type=SearchItemType.ENTITY.value,
title="Original Title",
content_stems="original content",
content_snippet="Original content snippet",
permalink=search_entity.permalink,
file_path=search_entity.file_path,
entity_id=search_entity.id,
metadata={"entity_type": search_entity.entity_type},
created_at=search_entity.created_at,
updated_at=search_entity.updated_at,
project_id=search_repository.project_id,
)
await search_repository.index_item(search_row1)
# Verify first insert worked
results = await search_repository.search(search_text="original")
assert len(results) == 1
assert results[0].title == "Original Title"
# Second insert with same permalink but different content (simulates race condition)
# This should NOT raise IntegrityError - it should upsert (update) instead
search_row2 = SearchIndexRow(
id=search_entity.id,
type=SearchItemType.ENTITY.value,
title="Updated Title",
content_stems="updated content",
content_snippet="Updated content snippet",
permalink=search_entity.permalink, # Same permalink!
file_path=search_entity.file_path,
entity_id=search_entity.id,
metadata={"entity_type": search_entity.entity_type},
created_at=search_entity.created_at,
updated_at=search_entity.updated_at,
project_id=search_repository.project_id,
)
# This should succeed without raising IntegrityError
await search_repository.index_item(search_row2)
# Verify the row was updated, not duplicated
results_after = await search_repository.search(search_text="updated")
assert len(results_after) == 1
assert results_after[0].title == "Updated Title"
# Verify old content is gone (was replaced)
results_old = await search_repository.search(search_text="original")
assert len(results_old) == 0
@pytest.mark.asyncio
async def test_bulk_index_items_upsert_on_duplicate_permalink(search_repository, search_entity):
"""Test that bulk_index_items uses upsert for duplicate permalinks.
This tests the fix for race conditions during bulk entity indexing.
"""
# First bulk insert
search_row1 = SearchIndexRow(
id=search_entity.id,
type=SearchItemType.ENTITY.value,
title="Bulk Original Title",
content_stems="bulk original content",
content_snippet="Bulk original content snippet",
permalink=search_entity.permalink,
file_path=search_entity.file_path,
entity_id=search_entity.id,
metadata={"entity_type": search_entity.entity_type},
created_at=search_entity.created_at,
updated_at=search_entity.updated_at,
project_id=search_repository.project_id,
)
await search_repository.bulk_index_items([search_row1])
# Verify first insert worked
results = await search_repository.search(search_text="bulk original")
assert len(results) == 1
assert results[0].title == "Bulk Original Title"
# Second bulk insert with same permalink (simulates race condition)
search_row2 = SearchIndexRow(
id=search_entity.id,
type=SearchItemType.ENTITY.value,
title="Bulk Updated Title",
content_stems="bulk updated content",
content_snippet="Bulk updated content snippet",
permalink=search_entity.permalink, # Same permalink!
file_path=search_entity.file_path,
entity_id=search_entity.id,
metadata={"entity_type": search_entity.entity_type},
created_at=search_entity.created_at,
updated_at=search_entity.updated_at,
project_id=search_repository.project_id,
)
# This should succeed without raising IntegrityError
await search_repository.bulk_index_items([search_row2])
# Verify the row was updated
results_after = await search_repository.search(search_text="bulk updated")
assert len(results_after) == 1
assert results_after[0].title == "Bulk Updated Title"
@pytest.mark.asyncio
async def test_project_isolation(
search_repository, second_project_repository, search_entity, second_entity
):
"""Test that search is isolated by project."""
# Index entities in both projects
search_row1 = SearchIndexRow(
id=search_entity.id,
type=SearchItemType.ENTITY.value,
title=search_entity.title,
content_stems="unique first project content",
content_snippet="This is a test entity in the first project",
permalink=search_entity.permalink,
file_path=search_entity.file_path,
entity_id=search_entity.id,
metadata={"entity_type": search_entity.entity_type},
created_at=search_entity.created_at,
updated_at=search_entity.updated_at,
project_id=search_repository.project_id,
)
search_row2 = SearchIndexRow(
id=second_entity.id,
type=SearchItemType.ENTITY.value,
title=second_entity.title,
content_stems="unique second project content",
content_snippet="This is a test entity in the second project",
permalink=second_entity.permalink,
file_path=second_entity.file_path,
entity_id=second_entity.id,
metadata={"entity_type": second_entity.entity_type},
created_at=second_entity.created_at,
updated_at=second_entity.updated_at,
project_id=second_project_repository.project_id,
)
# Index items in their respective repositories
await search_repository.index_item(search_row1)
await second_project_repository.index_item(search_row2)
# Search in first project
results1 = await search_repository.search(search_text="unique first")
assert len(results1) == 1
assert results1[0].title == search_entity.title
assert results1[0].project_id == search_repository.project_id
# Search in second project
results2 = await second_project_repository.search(search_text="unique second")
assert len(results2) == 1
assert results2[0].title == second_entity.title
assert results2[0].project_id == second_project_repository.project_id
# Make sure first project can't see second project's content
results_cross1 = await search_repository.search(search_text="unique second")
assert len(results_cross1) == 0
# Make sure second project can't see first project's content
results_cross2 = await second_project_repository.search(search_text="unique first")
assert len(results_cross2) == 0
@pytest.mark.asyncio
async def test_delete_by_permalink(search_repository, search_entity):
"""Test deleting an item by permalink respects project isolation."""
# Index the item
search_row = SearchIndexRow(
id=search_entity.id,
type=SearchItemType.ENTITY.value,
title=search_entity.title,
content_stems="content to delete",
content_snippet="This content should be deleted",
permalink=search_entity.permalink,
file_path=search_entity.file_path,
entity_id=search_entity.id,
metadata={"entity_type": search_entity.entity_type},
created_at=search_entity.created_at,
updated_at=search_entity.updated_at,
project_id=search_repository.project_id,
)
await search_repository.index_item(search_row)
# Verify it exists
results = await search_repository.search(search_text="content to delete")
assert len(results) == 1
# Delete by permalink
await search_repository.delete_by_permalink(search_entity.permalink)
# Verify it's gone
results_after = await search_repository.search(search_text="content to delete")
assert len(results_after) == 0
@pytest.mark.asyncio
async def test_delete_by_entity_id(search_repository, search_entity):
"""Test deleting an item by entity_id respects project isolation."""
# Index the item
search_row = SearchIndexRow(
id=search_entity.id,
type=SearchItemType.ENTITY.value,
title=search_entity.title,
content_stems="entity to delete",
content_snippet="This entity should be deleted",
permalink=search_entity.permalink,
file_path=search_entity.file_path,
entity_id=search_entity.id,
metadata={"entity_type": search_entity.entity_type},
created_at=search_entity.created_at,
updated_at=search_entity.updated_at,
project_id=search_repository.project_id,
)
await search_repository.index_item(search_row)
# Verify it exists
results = await search_repository.search(search_text="entity to delete")
assert len(results) == 1
# Delete by entity_id
await search_repository.delete_by_entity_id(search_entity.id)
# Verify it's gone
results_after = await search_repository.search(search_text="entity to delete")
assert len(results_after) == 0
@pytest.mark.asyncio
async def test_to_insert_includes_project_id(search_repository):
"""Test that the to_insert method includes project_id."""
# Create a search index row with project_id
row = SearchIndexRow(
id=1234,
type=SearchItemType.ENTITY.value,
title="Test Title",
content_stems="test content",
content_snippet="test snippet",
permalink="test/permalink",
file_path="test/file.md",
metadata={"test": "metadata"},
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
project_id=search_repository.project_id,
)
# Get insert data
insert_data = row.to_insert()
# Verify project_id is included
assert "project_id" in insert_data
assert insert_data["project_id"] == search_repository.project_id
def test_directory_property():
"""Test the directory property of SearchIndexRow."""
# Test a file in a nested directory
row1 = SearchIndexRow(
id=1,
type=SearchItemType.ENTITY.value,
file_path="projects/notes/ideas.md",
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
project_id=1,
)
assert row1.directory == "/projects/notes"
# Test a file at the root level
row2 = SearchIndexRow(
id=2,
type=SearchItemType.ENTITY.value,
file_path="README.md",
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
project_id=1,
)
assert row2.directory == "/"
# Test a non-entity type with empty file_path
row3 = SearchIndexRow(
id=3,
type=SearchItemType.OBSERVATION.value,
file_path="",
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
project_id=1,
)
assert row3.directory == ""
class TestSearchTermPreparation:
"""Test cases for search term preparation.
Note: Tests with `[sqlite]` marker test SQLite FTS5-specific implementation details.
Tests with `[asyncio-sqlite]` or `[asyncio-postgres]` test backend-agnostic functionality.
"""
def test_simple_terms_get_prefix_wildcard(self, search_repository):
"""Simple alphanumeric terms should get prefix matching."""
from basic_memory.repository.postgres_search_repository import PostgresSearchRepository
if isinstance(search_repository, PostgresSearchRepository):
# Postgres tsquery uses :* for prefix matching
assert search_repository._prepare_search_term("hello") == "hello:*"
assert search_repository._prepare_search_term("project") == "project:*"
assert search_repository._prepare_search_term("test123") == "test123:*"
else:
# SQLite FTS5 uses * for prefix matching
assert search_repository._prepare_search_term("hello") == "hello*"
assert search_repository._prepare_search_term("project") == "project*"
assert search_repository._prepare_search_term("test123") == "test123*"
def test_terms_with_existing_wildcard_unchanged(self, search_repository):
"""Terms that already contain * should remain unchanged."""
if is_postgres_backend(search_repository):
# Postgres uses different syntax (:* instead of *)
assert search_repository._prepare_search_term("hello*") == "hello:*"
assert search_repository._prepare_search_term("test*world") == "test:*world"
else:
assert search_repository._prepare_search_term("hello*") == "hello*"
assert search_repository._prepare_search_term("test*world") == "test*world"
def test_boolean_operators_preserved(self, search_repository):
"""Boolean operators should be preserved without modification."""
if is_postgres_backend(search_repository):
# Postgres converts AND/OR/NOT to &/|/!
assert search_repository._prepare_search_term("hello AND world") == "hello & world"
assert search_repository._prepare_search_term("cat OR dog") == "cat | dog"
# NOT must be converted to "& !" for proper tsquery syntax
assert (
search_repository._prepare_search_term("project NOT meeting")
== "project & !meeting"
)
assert (
search_repository._prepare_search_term("(hello AND world) OR test")
== "(hello & world) | test"
)
else:
assert search_repository._prepare_search_term("hello AND world") == "hello AND world"
assert search_repository._prepare_search_term("cat OR dog") == "cat OR dog"
assert (
search_repository._prepare_search_term("project NOT meeting")
== "project NOT meeting"
)
assert (
search_repository._prepare_search_term("(hello AND world) OR test")
== "(hello AND world) OR test"
)
def test_hyphenated_terms_with_boolean_operators(self, search_repository):
"""Hyphenated terms with Boolean operators should be properly quoted."""
if is_postgres_backend(search_repository):
pytest.skip("This test is for SQLite FTS5-specific quoting behavior")
# Test the specific case from the GitHub issue
result = search_repository._prepare_search_term("tier1-test AND unicode")
assert result == '"tier1-test" AND unicode'
# Test other hyphenated Boolean combinations
assert (
search_repository._prepare_search_term("multi-word OR single")
== '"multi-word" OR single'
)
assert (
search_repository._prepare_search_term("well-formed NOT badly-formed")
== '"well-formed" NOT "badly-formed"'
)
assert (
search_repository._prepare_search_term("test-case AND (hello OR world)")
== '"test-case" AND (hello OR world)'
)
# Test mixed special characters with Boolean operators
assert (
search_repository._prepare_search_term("config.json AND test-file")
== '"config.json" AND "test-file"'
)
assert (
search_repository._prepare_search_term("C++ OR python-script")
== '"C++" OR "python-script"'
)
def test_programming_terms_should_work(self, search_repository):
"""Programming-related terms with special chars should be searchable."""
if is_postgres_backend(search_repository):
pytest.skip("This test is for SQLite FTS5-specific behavior")
# These should be quoted to handle special characters safely
assert search_repository._prepare_search_term("C++") == '"C++"*'
assert search_repository._prepare_search_term("function()") == '"function()"*'
assert search_repository._prepare_search_term("email@domain.com") == '"email@domain.com"*'
assert search_repository._prepare_search_term("array[index]") == '"array[index]"*'
assert search_repository._prepare_search_term("config.json") == '"config.json"*'
def test_malformed_fts5_syntax_quoted(self, search_repository):
"""Malformed FTS5 syntax should be quoted to prevent errors."""
if is_postgres_backend(search_repository):
pytest.skip("This test is for SQLite FTS5-specific behavior")
# Multiple operators without proper syntax
assert search_repository._prepare_search_term("+++invalid+++") == '"+++invalid+++"*'
assert search_repository._prepare_search_term("!!!error!!!") == '"!!!error!!!"*'
assert search_repository._prepare_search_term("@#$%^&*()") == '"@#$%^&*()"*'
def test_quoted_strings_handled_properly(self, search_repository):
"""Strings with quotes should have quotes escaped."""
if is_postgres_backend(search_repository):
pytest.skip("This test is for SQLite FTS5-specific behavior")
assert search_repository._prepare_search_term('say "hello"') == '"say ""hello"""*'
assert search_repository._prepare_search_term("it's working") == '"it\'s working"*'
def test_file_paths_no_prefix_wildcard(self, search_repository):
"""File paths should not get prefix wildcards."""
if is_postgres_backend(search_repository):
pytest.skip("This test is for SQLite FTS5-specific behavior")
assert (
search_repository._prepare_search_term("config.json", is_prefix=False)
== '"config.json"'
)
assert (
search_repository._prepare_search_term("docs/readme.md", is_prefix=False)
== '"docs/readme.md"'
)
def test_spaces_handled_correctly(self, search_repository):
"""Terms with spaces should use boolean AND for word order independence."""
if is_postgres_backend(search_repository):
pytest.skip("This test is for SQLite FTS5-specific behavior")
assert search_repository._prepare_search_term("hello world") == "hello* AND world*"
assert (
search_repository._prepare_search_term("project planning") == "project* AND planning*"
)
def test_version_strings_with_dots_handled_correctly(self, search_repository):
"""Version strings with dots should be quoted to prevent FTS5 syntax errors."""
if is_postgres_backend(search_repository):
pytest.skip("This test is for SQLite FTS5-specific behavior")
# This reproduces the bug where "Basic Memory v0.13.0b2" becomes "Basic* AND Memory* AND v0.13.0b2*"
# which causes FTS5 syntax errors because v0.13.0b2* is not valid FTS5 syntax
result = search_repository._prepare_search_term("Basic Memory v0.13.0b2")
# Should be quoted because of dots in v0.13.0b2
assert result == '"Basic Memory v0.13.0b2"*'
def test_mixed_special_characters_in_multi_word_queries(self, search_repository):
"""Multi-word queries with special characters in any word should be fully quoted."""
if is_postgres_backend(search_repository):
pytest.skip("This test is for SQLite FTS5-specific behavior")
# Any word containing special characters should cause the entire phrase to be quoted
assert search_repository._prepare_search_term("config.json file") == '"config.json file"*'
assert (
search_repository._prepare_search_term("user@email.com account")
== '"user@email.com account"*'
)
assert search_repository._prepare_search_term("node.js and react") == '"node.js and react"*'
@pytest.mark.asyncio
async def test_search_with_special_characters_returns_results(self, search_repository):
"""Integration test: search with special characters should work gracefully."""
# This test ensures the search doesn't crash with FTS5 syntax errors
# These should all return empty results gracefully, not crash
results1 = await search_repository.search(search_text="C++")
assert isinstance(results1, list) # Should not crash
results2 = await search_repository.search(search_text="function()")
assert isinstance(results2, list) # Should not crash
results3 = await search_repository.search(search_text="+++malformed+++")
assert isinstance(results3, list) # Should not crash, return empty results
results4 = await search_repository.search(search_text="email@domain.com")
assert isinstance(results4, list) # Should not crash
@pytest.mark.asyncio
async def test_boolean_search_still_works(self, search_repository):
"""Boolean search operations should continue to work."""
# These should not crash and should respect boolean logic
results1 = await search_repository.search(search_text="hello AND world")
assert isinstance(results1, list)
results2 = await search_repository.search(search_text="cat OR dog")
assert isinstance(results2, list)
results3 = await search_repository.search(search_text="project NOT meeting")
assert isinstance(results3, list)
@pytest.mark.asyncio
async def test_permalink_match_exact_with_slash(self, search_repository):
"""Test exact permalink matching with slash (line 249 coverage)."""
# This tests the exact match path: if "/" in permalink_text:
results = await search_repository.search(permalink_match="test/path")
assert isinstance(results, list)
# Should use exact equality matching for paths with slashes
@pytest.mark.asyncio
async def test_permalink_match_simple_term(self, search_repository):
"""Test permalink matching with simple term (no slash)."""
# This tests the simple term path that goes through _prepare_search_term
results = await search_repository.search(permalink_match="simpleterm")
assert isinstance(results, list)
# Should use FTS5 MATCH for simple terms
@pytest.mark.asyncio
async def test_fts5_error_handling_database_error(self, search_repository):
"""Test that non-FTS5 database errors are properly re-raised."""
import unittest.mock
# Mock the scoped_session to raise a non-FTS5 error
with unittest.mock.patch("basic_memory.db.scoped_session") as mock_scoped_session:
mock_session = unittest.mock.AsyncMock()
mock_scoped_session.return_value.__aenter__.return_value = mock_session
# Simulate a database error that's NOT an FTS5 syntax error
mock_session.execute.side_effect = Exception("Database connection failed")
# This should re-raise the exception (not return empty list)
with pytest.raises(Exception, match="Database connection failed"):
await search_repository.search(search_text="test")
@pytest.mark.asyncio
async def test_version_string_search_integration(self, search_repository, search_entity):
"""Integration test: searching for version strings should work without FTS5 errors."""
# Index an entity with version information
search_row = SearchIndexRow(
id=search_entity.id,
type=SearchItemType.ENTITY.value,
title="Basic Memory v0.13.0b2 Release",
content_stems="basic memory version 0.13.0b2 beta release notes features",
content_snippet="Basic Memory v0.13.0b2 is a beta release with new features",
permalink=search_entity.permalink,
file_path=search_entity.file_path,
entity_id=search_entity.id,
metadata={"entity_type": search_entity.entity_type},
created_at=search_entity.created_at,
updated_at=search_entity.updated_at,
project_id=search_repository.project_id,
)
await search_repository.index_item(search_row)
# This should not cause FTS5 syntax errors and should find the entity
results = await search_repository.search(search_text="Basic Memory v0.13.0b2")
assert len(results) == 1
assert results[0].title == "Basic Memory v0.13.0b2 Release"
# Test other version-like patterns
results2 = await search_repository.search(search_text="v0.13.0b2")
assert len(results2) == 1 # Should still find it due to content_stems
# Test with other problematic patterns
results3 = await search_repository.search(search_text="node.js version")
assert isinstance(results3, list) # Should not crash
@pytest.mark.asyncio
async def test_wildcard_only_search(self, search_repository, search_entity):
"""Test that wildcard-only search '*' doesn't cause FTS5 errors (line 243 coverage)."""
# Index an entity for testing
search_row = SearchIndexRow(
id=search_entity.id,
type=SearchItemType.ENTITY.value,
title="Test Entity",
content_stems="test entity content",
content_snippet="This is a test entity",
permalink=search_entity.permalink,
file_path=search_entity.file_path,
entity_id=search_entity.id,
metadata={"entity_type": search_entity.entity_type},
created_at=search_entity.created_at,
updated_at=search_entity.updated_at,
project_id=search_repository.project_id,
)
await search_repository.index_item(search_row)
# Test wildcard-only search - should not crash and should return results
results = await search_repository.search(search_text="*")
assert isinstance(results, list) # Should not crash
assert len(results) >= 1 # Should return all results, including our test entity
# Test empty string search - should also not crash
results_empty = await search_repository.search(search_text="")
assert isinstance(results_empty, list) # Should not crash
# Test whitespace-only search
results_whitespace = await search_repository.search(search_text=" ")
assert isinstance(results_whitespace, list) # Should not crash
def test_boolean_query_empty_parts_coverage(self, search_repository):
"""Test Boolean query parsing with empty parts (line 143 coverage)."""
# Create queries that will result in empty parts after splitting
result1 = search_repository._prepare_boolean_query(
"hello AND AND world"
) # Double operator
assert "hello" in result1 and "world" in result1
result2 = search_repository._prepare_boolean_query(" OR test") # Leading operator
assert "test" in result2
result3 = search_repository._prepare_boolean_query("test OR ") # Trailing operator
assert "test" in result3
def test_parenthetical_term_quote_escaping(self, search_repository):
"""Test quote escaping in parenthetical terms (lines 190-191 coverage)."""
if is_postgres_backend(search_repository):
pytest.skip("This test is for SQLite FTS5-specific behavior")
# Test term with quotes that needs escaping
result = search_repository._prepare_parenthetical_term('(say "hello" world)')
# Should escape quotes by doubling them
assert '""hello""' in result
# Test term with single quotes
result2 = search_repository._prepare_parenthetical_term("(it's working)")
assert "it's working" in result2
def test_needs_quoting_empty_input(self, search_repository):
"""Test _needs_quoting with empty inputs (line 207 coverage)."""
if is_postgres_backend(search_repository):
pytest.skip("This test is for SQLite FTS5-specific behavior")
# Test empty string
assert not search_repository._needs_quoting("")
# Test whitespace-only string
assert not search_repository._needs_quoting(" ")
# Test None-like cases
assert not search_repository._needs_quoting("\t")
def test_prepare_single_term_empty_input(self, search_repository):
"""Test _prepare_single_term with empty inputs (line 227 coverage)."""
# Test empty string
result1 = search_repository._prepare_single_term("")
assert result1 == ""
# Test whitespace-only string
result2 = search_repository._prepare_single_term(" ")
assert result2 == " " # Should return as-is
# Test string that becomes empty after strip
result3 = search_repository._prepare_single_term("\t\n")
assert result3 == "\t\n" # Should return original
