"""Performance tests for the search functionality."""
import random
import string
import time
from datetime import datetime, timedelta
from typing import Any
import pytest
from simplenote_mcp.server.search.engine import SearchEngine
def generate_random_text(min_words=50, max_words=200):
"""Generate random text with a specified number of words."""
words = []
word_count = random.randint(min_words, max_words) # noqa: S311
for _ in range(word_count):
# Generate a random word length between 3 and 10 characters
word_length = random.randint(3, 10) # noqa: S311
word = "".join(
random.choice(string.ascii_lowercase)
for _ in range(word_length) # noqa: S311
)
words.append(word)
# Insert some specific keywords for searching
keywords = [
"project",
"meeting",
"report",
"important",
"todo",
"deadline",
"client",
]
for _ in range(random.randint(3, 8)): # Insert 3-8 keywords # noqa: S311
position = random.randint(0, len(words) - 1) # noqa: S311
keyword = random.choice(keywords) # noqa: S311
words[position] = keyword
return " ".join(words)
@pytest.fixture
def large_note_collection():
"""Generate a large collection of notes for performance testing."""
notes = {}
now = datetime.now()
# Generate 500 notes
for i in range(500):
# Create a random date within the last year
days_ago = random.randint(0, 365) # noqa: S311
note_date = now - timedelta(days=days_ago)
# Select random tags
all_tags = [
"work",
"personal",
"project",
"meeting",
"important",
"todo",
"idea",
]
note_tags = random.sample(all_tags, random.randint(0, 3)) # noqa: S311
# Generate note content
note_content = generate_random_text()
# Create the note
notes[f"note{i}"] = {
"key": f"note{i}",
"content": note_content,
"tags": note_tags,
"modifydate": note_date.isoformat(),
}
return notes
@pytest.mark.perf
class TestSearchPerformance:
"""Test search performance with large datasets."""
def test_simple_term_search_performance(self, large_note_collection):
"""Test performance of simple term search."""
engine = SearchEngine()
# Time a simple search
start_time = time.time()
results = engine.search(large_note_collection, "project")
end_time = time.time()
# Print performance info
elapsed_time = end_time - start_time
print(f"Simple search 'project' over 500 notes took {elapsed_time:.4f} seconds")
print(f"Found {len(results)} matching notes")
# Assert the search completes in a reasonable time
assert elapsed_time < 0.5, "Simple search should complete quickly"
def test_boolean_search_performance(self, large_note_collection):
"""Test performance of boolean search expressions."""
engine = SearchEngine()
# Time a complex boolean search
start_time = time.time()
results = engine.search(
large_note_collection, "project AND meeting AND NOT todo"
)
end_time = time.time()
# Print performance info
elapsed_time = end_time - start_time
print(f"Boolean search over 500 notes took {elapsed_time:.4f} seconds")
print(f"Found {len(results)} matching notes")
# Assert the search completes in a reasonable time
assert elapsed_time < 1.0, "Boolean search should complete in reasonable time"
def test_tag_filter_performance(self, large_note_collection):
"""Test performance of tag filtering."""
engine = SearchEngine()
# Time a tag filter search
start_time = time.time()
results = engine.search(
large_note_collection, "", tag_filters=["work", "important"]
)
end_time = time.time()
# Print performance info
elapsed_time = end_time - start_time
print(f"Tag filter search over 500 notes took {elapsed_time:.4f} seconds")
print(f"Found {len(results)} matching notes")
# Assert the search completes in a reasonable time
assert elapsed_time < 0.5, "Tag filter search should complete quickly"
def test_date_filter_performance(self, large_note_collection):
"""Test performance of date filtering."""
engine = SearchEngine()
# Get dates for a 30-day period
now = datetime.now()
month_ago = now - timedelta(days=30)
# Time a date filter search
start_time = time.time()
results = engine.search(large_note_collection, "", date_range=(month_ago, now))
end_time = time.time()
# Print performance info
elapsed_time = end_time - start_time
print(f"Date filter search over 500 notes took {elapsed_time:.4f} seconds")
print(f"Found {len(results)} matching notes")
# Assert the search completes in a reasonable time
assert elapsed_time < 0.5, "Date filter search should complete quickly"
def test_complex_query_performance(self, large_note_collection):
"""Test performance of a complex query with multiple components."""
engine = SearchEngine()
# Get dates for a 90-day period
now = datetime.now()
three_months_ago = now - timedelta(days=90)
# Time a complex search with multiple filters
start_time = time.time()
results = engine.search(
large_note_collection,
"project AND (report OR meeting) AND NOT deadline",
tag_filters=["work"],
date_range=(three_months_ago, now),
)
end_time = time.time()
# Print performance info
elapsed_time = end_time - start_time
print(f"Complex search over 500 notes took {elapsed_time:.4f} seconds")
print(f"Found {len(results)} matching notes")
# Assert the search completes in a reasonable time
assert elapsed_time < 1.0, "Complex search should complete in reasonable time"
@pytest.mark.perf
class TestSearchStress:
"""Stress test the search functionality with extreme cases."""
@pytest.fixture
def very_large_note(self):
"""Generate a very large note for stress testing."""
# Generate a note with 10,000 words
content = generate_random_text(min_words=10000, max_words=10000)
return {
"large_note": {
"key": "large_note",
"content": content,
"tags": ["test", "large"],
"modifydate": datetime.now().isoformat(),
}
}
@pytest.fixture
def many_small_notes(self):
"""Generate many small notes for stress testing."""
notes = {}
# Generate 1000 small notes
for i in range(1000):
content = generate_random_text(min_words=10, max_words=30)
notes[f"note{i}"] = {
"key": f"note{i}",
"content": content,
"tags": ["test"],
"modifydate": datetime.now().isoformat(),
}
return notes
def test_very_large_note_search(self, very_large_note):
"""Test search performance with a very large note."""
engine = SearchEngine()
# Time search in a very large note
start_time = time.time()
_ = engine.search(
very_large_note, "project"
) # Result ignored; we're testing timing
end_time = time.time()
# Print performance info
elapsed_time = end_time - start_time
print(f"Search in a 10,000-word note took {elapsed_time:.4f} seconds")
# Assert the search completes in a reasonable time
assert elapsed_time < 0.5, "Search in large note should complete quickly"
def test_many_small_notes_search(self, many_small_notes):
"""Test search performance with many small notes."""
engine = SearchEngine()
# Time search across many small notes
start_time = time.time()
_ = engine.search(
many_small_notes, "project"
) # Result ignored; we're testing timing
end_time = time.time()
# Print performance info
elapsed_time = end_time - start_time
print(f"Search across 1000 small notes took {elapsed_time:.4f} seconds")
# Assert the search completes in a reasonable time
assert elapsed_time < 1.0, (
"Search across many notes should complete in reasonable time"
)
def test_complex_query_on_many_notes(self, many_small_notes):
"""Test complex query performance on many notes."""
engine = SearchEngine()
# Time complex search across many notes
start_time = time.time()
_ = engine.search(
many_small_notes,
"(project OR meeting) AND (report OR todo) AND NOT deadline",
) # Result ignored; we're testing timing
end_time = time.time()
# Print performance info
elapsed_time = end_time - start_time
print(f"Complex search across 1000 notes took {elapsed_time:.4f} seconds")
# Assert the search completes in a reasonable time
assert elapsed_time < 1.5, (
"Complex search on many notes should complete in reasonable time"
)
@pytest.mark.asyncio
async def test_concurrent_searches(self, many_small_notes):
"""Test performance with multiple concurrent searches."""
import asyncio
async def run_search(search_term):
"""Run a search asynchronously."""
engine = SearchEngine()
start_time = time.time()
results = engine.search(many_small_notes, search_term)
end_time = time.time()
elapsed = end_time - start_time
return search_term, len(results), elapsed
# Define different search queries to run concurrently
search_terms = [
"project",
"meeting",
"report",
"todo",
"deadline",
"project AND meeting",
"report OR todo",
"project AND NOT deadline",
]
# Run searches concurrently
overall_start = time.time()
tasks = [run_search(term) for term in search_terms]
results = await asyncio.gather(*tasks)
overall_end = time.time()
# Print performance metrics
print("\nConcurrent search performance results:")
print(
f"Total time for {len(search_terms)} concurrent searches: {overall_end - overall_start:.4f} seconds"
)
for term, count, elapsed in results:
print(f" Search '{term}': found {count} results in {elapsed:.4f} seconds")
# Calculate average time and throughput
avg_time = sum(elapsed for _, _, elapsed in results) / len(results)
throughput = len(results) / (overall_end - overall_start)
print(f"Average search time: {avg_time:.4f} seconds")
print(f"Search throughput: {throughput:.2f} searches/second")
# Check that concurrent searches don't take too long
assert (overall_end - overall_start) < 3.0, (
"Concurrent searches should complete in a reasonable time"
)
# Individual searches should still be reasonably fast
for _, _, elapsed in results:
assert elapsed < 1.5, (
"Individual concurrent searches should complete in a reasonable time"
)
@pytest.mark.skip(reason="Only run manually for benchmark purposes")
def test_search_benchmark(self, many_small_notes: dict) -> None:
"""Benchmark search performance with different query types."""
engine = SearchEngine()
# Define a list of benchmark cases with proper typing
benchmark_cases: list[tuple[str, str, list[str] | None]] = [
("Simple term", "project", None),
("Multiple terms", "project meeting report", None),
("Boolean AND", "project AND meeting", None),
("Boolean OR", "project OR meeting", None),
("Boolean NOT", "project NOT meeting", None),
(
"Complex boolean",
"(project OR meeting) AND (report OR todo) AND NOT deadline",
None,
),
("Tag filter only", "", ["work"]),
("Term with tag filter", "project", ["work"]),
]
results: dict[str, dict[str, Any]] = {}
# Run each benchmark case multiple times
runs = 5
for name, query, tag_filters in benchmark_cases:
total_time: float = 0.0
for _ in range(runs):
start_time = time.time()
if tag_filters:
_ = engine.search(
many_small_notes, query, tag_filters=tag_filters
) # Result ignored; just timing
else:
_ = engine.search(
many_small_notes, query
) # Result ignored; just timing
elapsed = time.time() - start_time
total_time += elapsed
avg_time = total_time / runs
results[name] = {
"avg_time": avg_time,
"query": query,
"tag_filters": tag_filters,
}
# Print benchmark results
print("\nSearch Performance Benchmark Results:")
print("-" * 80)
print(f"{'Query Type':<30} | {'Avg Time (sec)':<15} | {'Query':<35}")
print("-" * 80)
for name, data in results.items():
query_display = data["query"] if data["query"] else "(empty query)"
if data["tag_filters"]:
query_display += f" tags:{data['tag_filters']}"
print(f"{name:<30} | {data['avg_time']:<15.4f} | {query_display:<35}")
# No assertions - this is purely informational
