Simple Search API Testing

---

This guide shows how to test search APIs with common operations like text search, filtering, faceted search, and pagination.

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

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

1. Replace search endpoints with your actual search API URLs
2. Update search terms to match your domain/content
3. Adjust filter options based on your search facets
4. 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