Use Cases
- Test search functionality under load
- Validate search relevance and accuracy
- Test search filters and facets
- Check search performance with different query types
Simple Implementation
Copy
Ask AI
from locust import task, HttpUser
import random
import json
class SearchTestUser(HttpUser):
def on_start(self):
# Search terms for testing
self.search_terms = [
"python", "api", "testing", "performance", "database",
"tutorial", "guide", "example", "documentation", "development"
]
# Filter options
self.filters = {
"category": ["technology", "business", "science", "sports"],
"date_range": ["today", "week", "month", "year"],
"type": ["article", "video", "document", "image"]
}
# API endpoints
self.search_endpoints = {
"search": "/api/search",
"suggest": "/api/search/suggest",
"autocomplete": "/api/search/autocomplete"
}
@task(4)
def test_basic_search(self):
"""Test basic text search"""
query = random.choice(self.search_terms)
search_params = {
"q": query,
"limit": random.choice([10, 20, 50]),
"offset": random.choice([0, 10, 20])
}
with self.client.get(
self.search_endpoints["search"],
params=search_params,
name="Basic Search"
) as response:
if response.status_code == 200:
try:
results = response.json()
# Handle different response formats
if isinstance(results, dict):
items = results.get("results", results.get("hits", []))
total = results.get("total", len(items))
took = results.get("took", 0)
else:
items = results if isinstance(results, list) else []
total = len(items)
took = 0
print(f"Search '{query}': {len(items)} results ({took}ms)")
except json.JSONDecodeError:
response.failure("Invalid JSON response")
else:
response.failure(f"Search failed: {response.status_code}")
@task(3)
def test_filtered_search(self):
"""Test search with filters"""
query = random.choice(self.search_terms)
category = random.choice(self.filters["category"])
date_range = random.choice(self.filters["date_range"])
search_params = {
"q": query,
"category": category,
"date": date_range,
"limit": 20
}
with self.client.get(
self.search_endpoints["search"],
params=search_params,
name="Filtered Search"
) as response:
if response.status_code == 200:
try:
results = response.json()
if isinstance(results, dict):
items = results.get("results", results.get("hits", []))
facets = results.get("facets", {})
print(f"Filtered search '{query}' in {category}: {len(items)} results")
# Validate filters were applied
if facets:
print(f"Available facets: {list(facets.keys())}")
else:
items = results if isinstance(results, list) else []
print(f"Filtered search: {len(items)} results")
except json.JSONDecodeError:
response.failure("Invalid JSON response")
else:
response.failure(f"Filtered search failed: {response.status_code}")
@task(2)
def test_search_suggestions(self):
"""Test search suggestions/autocomplete"""
partial_query = random.choice(self.search_terms)[:3] # First 3 characters
suggest_params = {
"q": partial_query,
"limit": 10
}
with self.client.get(
self.search_endpoints["suggest"],
params=suggest_params,
name="Search Suggestions"
) as response:
if response.status_code == 200:
try:
suggestions = response.json()
if isinstance(suggestions, dict):
items = suggestions.get("suggestions", suggestions.get("items", []))
else:
items = suggestions if isinstance(suggestions, list) else []
print(f"Suggestions for '{partial_query}': {len(items)} items")
except json.JSONDecodeError:
response.failure("Invalid JSON response")
else:
response.failure(f"Search suggestions failed: {response.status_code}")
@task(2)
def test_advanced_search(self):
"""Test advanced search with multiple parameters"""
search_data = {
"query": {
"text": random.choice(self.search_terms),
"fields": ["title", "content", "tags"]
},
"filters": {
"type": random.choice(self.filters["type"]),
"category": random.choice(self.filters["category"])
},
"sort": random.choice(["relevance", "date", "popularity"]),
"pagination": {
"page": random.randint(1, 3),
"per_page": 20
}
}
with self.client.post(
self.search_endpoints["search"],
json=search_data,
name="Advanced Search"
) as response:
if response.status_code == 200:
try:
results = response.json()
if isinstance(results, dict):
items = results.get("results", results.get("hits", []))
total = results.get("total", len(items))
page = results.get("page", 1)
print(f"Advanced search: {len(items)} results, page {page}")
else:
items = results if isinstance(results, list) else []
print(f"Advanced search: {len(items)} results")
except json.JSONDecodeError:
response.failure("Invalid JSON response")
else:
response.failure(f"Advanced search failed: {response.status_code}")
@task(1)
def test_empty_search(self):
"""Test search with empty or invalid queries"""
empty_queries = ["", " ", "xyz123nonexistent", "!@#$%^&*()"]
query = random.choice(empty_queries)
search_params = {
"q": query,
"limit": 10
}
with self.client.get(
self.search_endpoints["search"],
params=search_params,
name="Empty/Invalid Search"
) as response:
if response.status_code in [200, 400]:
try:
results = response.json()
if isinstance(results, dict):
items = results.get("results", results.get("hits", []))
error = results.get("error")
if error:
print(f"Search error for '{query}': {error}")
else:
print(f"Empty search '{query}': {len(items)} results")
else:
items = results if isinstance(results, list) else []
print(f"Empty search: {len(items)} results")
except json.JSONDecodeError:
print(f"Non-JSON response for empty search")
else:
print(f"Empty search returned: {response.status_code}")
@task(1)
def test_search_aggregations(self):
"""Test search with aggregations/facets"""
query = random.choice(self.search_terms)
agg_params = {
"q": query,
"facets": "category,type,date",
"limit": 20
}
with self.client.get(
self.search_endpoints["search"],
params=agg_params,
name="Search with Aggregations"
) as response:
if response.status_code == 200:
try:
results = response.json()
if isinstance(results, dict):
items = results.get("results", results.get("hits", []))
aggregations = results.get("aggregations", results.get("facets", {}))
print(f"Aggregated search '{query}': {len(items)} results")
if aggregations:
print(f"Aggregations: {list(aggregations.keys())}")
else:
items = results if isinstance(results, list) else []
print(f"Aggregated search: {len(items)} results")
except json.JSONDecodeError:
response.failure("Invalid JSON response")
else:
response.failure(f"Aggregated search failed: {response.status_code}")
Setup Instructions
- Replace search endpoints with your actual search API URLs
- Update search terms to match your domain/content
- Adjust filter options based on your search facets
- Configure authentication if required by your search API
What This Tests
- Basic Search: Tests simple text search functionality
- Filtered Search: Tests search with category and date filters
- Search Suggestions: Tests autocomplete and suggestion features
- Advanced Search: Tests complex queries with multiple parameters
- Edge Cases: Tests empty queries and invalid input handling
- Aggregations: Tests search facets and result grouping
Best Practices
- Use realistic search terms relevant to your content
- Test various search result sizes and pagination
- Validate search performance with different query complexities
- Test both successful and edge case scenarios
- Monitor search response times and relevance
Common Issues
- Search Indexing: New content may not appear in search immediately
- Query Parsing: Complex queries may require special escaping
- Performance: Large result sets or complex filters may be slow
- Relevance: Search ranking may vary based on content and algorithms