Skip to main content

Testing Examples and How Tests Work

This document provides detailed examples and explanations of how the hlquery test suite works.

Table of Contents​

  1. Test Architecture
  2. Test Execution Flow
  3. Detailed Test Examples
  4. Common Test Patterns
  5. Understanding Test Output
  6. Debugging Failed Tests

Test Architecture​

How Tests Are Organized​

api/
├── python/tests/          # 13 tests (core functionality)
├── perl/tests/            # 31 tests (comprehensive)
├── php/tests/             # 31 tests (comprehensive)
├── node/tests/            # 31 tests (comprehensive)
├── cpp/tests/             # 31 tests (comprehensive)
└── rust/tests/            # 31 tests (comprehensive)

Test File Structure​

Every test file follows this pattern:

#!/usr/bin/env python3
"""
Test Description - What this test does
"""

import sys
import os
import time  # If needed for timestamps

# Change to script directory (working directory)
script_dir = os.path.dirname(os.path.abspath(__file__))
os.chdir(script_dir)

# Path setup - allows importing the client library
python_api_path = os.path.join(
    os.path.dirname(os.path.dirname(os.path.dirname(script_dir))), 
    'python'
)
sys.path.insert(0, python_api_path)

from lib import Client

BASE_URL = 'http://localhost:9200'

Important: The os.chdir(script_dir) line changes the working directory to the test script's directory. This means:

  • Tests can be run from anywhere
  • Working directory is always the test directory
  • Relative file paths work correctly

For a complete explanation of working directory management and test execution, see How Tests Work.

def test_feature_name():
    """Main test function"""
    print("=" * 70)
    print("TEST: Feature Name (API Name)")
    print("-" * 70)
    
    # Initialize
    client = Client(BASE_URL)
    tests_passed = 0
    tests_failed = 0
    
    # Test scenarios
    # ...
    
    # Summary
    print("\n" + "=" * 70)
    print(f"Test Summary: {tests_passed} passed, {tests_failed} failed")
    print("=" * 70 + "\n")

if __name__ == '__main__':
    test_feature_name()

Test Execution Flow​

Step-by-Step: What Happens When You Run a Test​

  1. Script Execution Starts

    $ python api/perl/tests/test_search.py

  2. Path Setup

    • Script adds Python API directory to Python path
    • Allows importing lib.Client

  3. Client Initialization

    client = Client('http://localhost:9200')

  4. Test Data Preparation

    # Get existing collection or create new one
    collections = client.list_collections(0, 1)
    if collections.is_success():
        # Use existing collection
    else:
        # Create test collection
        collection_name = 'test_search_' + str(int(time.time()))
        client.collections_api().create(collection_name, schema)

  5. Test Scenarios Execution

    # Test 1: Basic search
    result = client.search(collection_name, {'q': 'test', ...})
    if result.is_success():
        tests_passed += 1
    else:
        tests_failed += 1

  6. Cleanup

    if created_collection:
        client.collections_api().delete(collection_name)

  7. Summary Report

    print(f"Test Summary: {tests_passed} passed, {tests_failed} failed")

Detailed Test Examples​

Example 1: Complete test_search.py Walkthrough​

def test_search():
    """Test search functionality with multiple scenarios"""
    
    # === SETUP PHASE ===
    client = Client(BASE_URL)
    tests_passed = 0
    tests_failed = 0
    
    # Try to get existing collection
    collections = client.list_collections(0, 1)
    collection_name = None
    created_collection = False
    
    if collections.is_success():
        body = collections.get_body()
        if body.get('collections'):
            collection_name = body['collections'][0]['name']
    
    # Create test collection if none exists
    if not collection_name:
        collection_name = 'test_search_' + str(int(time.time()))
        schema = {
            'fields': [
                {'name': 'title', 'type': 'string'},
                {'name': 'content', 'type': 'string'},
                {'name': 'category', 'type': 'string'},
                {'name': 'price', 'type': 'float'}
            ]
        }
        create_result = client.collections_api().create(collection_name, schema)
        if create_result.is_success():
            created_collection = True
            
            # Add test documents
            test_docs = [
                {'id': 'doc1', 'title': 'Test Document One', 
                 'content': 'This is test content', 'category': 'test', 'price': 10.0},
                {'id': 'doc2', 'title': 'Test Document Two', 
                 'content': 'Another test document', 'category': 'test', 'price': 20.0},
                {'id': 'doc3', 'title': 'Sample Document', 
                 'content': 'Sample content here', 'category': 'sample', 'price': 15.0}
            ]
            for doc in test_docs:
                client.documents_api().add(collection_name, doc)
    
    # === TEST EXECUTION PHASE ===
    
    # Test 1: Basic search
    print("Test 1: Basic search")
    search_params = {
        'q': 'test',
        'query_by': 'title,content',
        'limit': 10
    }
    results = client.search(collection_name, search_params)
    if results.is_success():
        body = results.get_body()
        hits = body.get('hits', [])
        print(f"  ✓ Found {len(hits)} results")
        tests_passed += 1
    else:
        print(f"  ✗ Search failed: {results.get_status_code()}")
        tests_failed += 1
    
    # Test 2: Search with filters
    print("\nTest 2: Search with filters")
    search_params = {
        'q': '*',
        'query_by': 'title',
        'filter_by': 'category:test',
        'limit': 5
    }
    results = client.search(collection_name, search_params)
    if results.is_success():
        body = results.get_body()
        hits = body.get('hits', [])
        print(f"  ✓ Filtered search found {len(hits)} results")
        tests_passed += 1
    else:
        print(f"  ✗ Filtered search failed: {results.get_status_code()}")
        tests_failed += 1
    
    # ... more tests ...
    
    # === CLEANUP PHASE ===
    if created_collection:
        delete_result = client.collections_api().delete(collection_name)
        if delete_result.is_success():
            print(f"\n✓ Cleaned up test collection: {collection_name}")
    
    # === SUMMARY PHASE ===
    print("\n" + "=" * 70)
    print(f"Search Test Summary: {tests_passed} passed, {tests_failed} failed")
    print("=" * 70 + "\n")

What This Test Does:

  1. Setup: Creates a test collection with 3 documents
  2. Test 1: Searches for "test" in title and content fields
  3. Test 2: Searches all documents but filters by category
  4. Test 3-5: More search scenarios (sorting, pagination, wildcard)
  5. Cleanup: Deletes the test collection
  6. Report: Shows pass/fail summary

Example 2: test_bulk_operations.py Deep Dive​

def test_bulk_operations():
    """Test bulk document operations"""
    
    # Create test collection
    collection_name = 'test_bulk_' + str(int(time.time()))
    schema = {
        'fields': [
            {'name': 'title', 'type': 'string'},
            {'name': 'content', 'type': 'string'},
            {'name': 'category', 'type': 'string'}
        ]
    }
    
    create_result = client.collections_api().create(collection_name, schema)
    print(f"✓ Created test collection: {collection_name}\n")
    
    # Prepare bulk documents
    bulk_docs = [
        {'id': 'bulk1', 'title': 'Bulk Document 1', 
         'content': 'Content 1', 'category': 'bulk'},
        {'id': 'bulk2', 'title': 'Bulk Document 2', 
         'content': 'Content 2', 'category': 'bulk'},
        # ... 3 more documents
    ]
    
    # Test 1: Bulk import
    print("Test 1: Bulk import documents")
    import_result = client.documents_api().import_documents(collection_name, bulk_docs)
    if import_result.is_success():
        print(f"  ✓ Bulk imported {len(bulk_docs)} documents")
        tests_passed += 1
    else:
        print(f"  ✗ Bulk import failed: {import_result.get_status_code()}")
        tests_failed += 1
    
    # Test 2: Verify import
    print("\nTest 2: Verify bulk import")
    documents = client.list_documents(collection_name, {'offset': 0, 'limit': 10})
    if documents.is_success():
        body = documents.get_body()
        docs = body.get('documents', [])
        print(f"  ✓ Found {len(docs)} documents in collection")
        if len(docs) >= len(bulk_docs):
            tests_passed += 1
        else:
            tests_failed += 1
    
    # Test 3: Bulk delete by filter
    print("\nTest 3: Bulk delete by filter")
    delete_result = client.documents_api().delete_by_filter(collection_name, 'category:bulk')
    if delete_result.is_success():
        print(f"  ✓ Bulk delete by filter successful")
        tests_passed += 1
    else:
        print(f"  ✗ Bulk delete failed: {delete_result.get_status_code()}")
        tests_failed += 1
    
    # Cleanup
    client.collections_api().delete(collection_name)

Key Points:

  • Creates collection once
  • Imports multiple documents in single API call
  • Verifies all documents were imported
  • Tests bulk delete functionality
  • Cleans up everything

Example 3: test_error_handling.py Explained​

def test_error_handling():
    """Test error handling scenarios"""
    
    # Test 1: Get non-existent collection
    print("Test 1: Get non-existent collection")
    fake_name = '__nonexistent_collection__' + str(int(time.time()))
    result = client.get_collection(fake_name)
    
    # Should fail with 404
    if not result.is_success() and result.get_status_code() in [404, 400]:
        print(f"  ✓ Correctly returned error: {result.get_status_code()}")
        tests_passed += 1
    else:
        print(f"  ✗ Unexpected response: {result.get_status_code()}")
        tests_failed += 1
    
    # Test 2: Get non-existent document
    print("\nTest 2: Get non-existent document")
    # First, get a real collection
    collections = client.list_collections(0, 1)
    if collections.is_success():
        body = collections.get_body()
        if body.get('collections'):
            collection_name = body['collections'][0]['name']
            # Try to get document that doesn't exist
            result = client.get_document(collection_name, '__nonexistent_doc__')
            if not result.is_success() and result.get_status_code() in [404, 400]:
                print(f"  ✓ Correctly returned error: {result.get_status_code()}")
                tests_passed += 1
    
    # Test 3: Invalid schema
    print("\nTest 3: Create collection with invalid schema")
    invalid_schema = {'invalid': 'data'}  # Missing required 'fields'
    result = client.collections_api().create('test_invalid', invalid_schema)
    if not result.is_success() and result.get_status_code() in [400, 422]:
        print(f"  ✓ Correctly rejected invalid schema: {result.get_status_code()}")
        tests_passed += 1

What This Tests:

  • API returns correct error codes (404 for not found, 400 for bad request)
  • API handles invalid input gracefully
  • Error messages are appropriate

Common Test Patterns​

Pattern 1: Create-Use-Cleanup​

# CREATE
collection_name = 'test_' + str(int(time.time()))
client.collections_api().create(collection_name, schema)

# USE
# ... perform operations ...

# CLEANUP
client.collections_api().delete(collection_name)

Pattern 2: Use Existing or Create​

# Try to use existing
collections = client.list_collections(0, 1)
if collections.is_success():
    body = collections.get_body()
    if body.get('collections'):
        collection_name = body['collections'][0]['name']
        use_existing = True

# Create if needed
if not use_existing:
    collection_name = 'test_' + str(int(time.time()))
    client.collections_api().create(collection_name, schema)
    created_collection = True

# ... use collection ...

# Cleanup only if we created it
if created_collection:
    client.collections_api().delete(collection_name)

Pattern 3: Pass/Fail Tracking​

tests_passed = 0
tests_failed = 0

# Test scenario
result = client.some_operation()
if result.is_success():
    print("  ✓ Operation successful")
    tests_passed += 1
else:
    print(f"  ✗ Operation failed: {result.get_status_code()}")
    tests_failed += 1

# Summary
print(f"Summary: {tests_passed} passed, {tests_failed} failed")

Pattern 4: Multiple Test Scenarios​

# Test 1: Basic operation
print("Test 1: Basic operation")
result = client.operation1()
validate_result(result, "Test 1")

# Test 2: With parameters
print("\nTest 2: With parameters")
result = client.operation2(params)
validate_result(result, "Test 2")

# Test 3: Edge case
print("\nTest 3: Edge case")
result = client.operation3(edge_case_params)
validate_result(result, "Test 3")

Understanding Test Output​

Successful Test Output​

----------------------------------------------------------------==============================
TEST: Search Functionality (Perl API)
----------------------------------------------------------------------
✓ Created test collection: test_search_perl_1234567890
✓ Added 3 test documents

Using collection: test_search_perl_1234567890

Test 1: Basic search
  ✓ Found 2 results

Test 2: Search with filters
  ✓ Filtered search found 2 results

Test 3: Search with sorting
  ✓ Sorted search successful

Test 4: Search with pagination
  ✓ Paginated search returned 2 results

Test 5: Wildcard search
  ✓ Wildcard search successful

✓ Cleaned up test collection: test_search_perl_1234567890

----------------------------------------------------------------==============================
Search Test Summary: 5 passed, 0 failed
----------------------------------------------------------------==============================

Reading the Output:

  • === lines show test boundaries
  • ✓ indicates success
  • ✗ indicates failure
  • Summary shows pass/fail counts

Failed Test Output​

----------------------------------------------------------------==============================
TEST: Search Functionality (Perl API)
----------------------------------------------------------------------
✓ Created test collection: test_search_perl_1234567890
✓ Added 3 test documents

Using collection: test_search_perl_1234567890

Test 1: Basic search
  ✓ Found 2 results

Test 2: Search with filters
  ✗ Filtered search failed: 400

Test 3: Search with sorting
  ✓ Sorted search successful

...

----------------------------------------------------------------==============================
Search Test Summary: 4 passed, 1 failed
----------------------------------------------------------------==============================

What to Look For:

  • Which test failed (Test 2)
  • Error code (400 = Bad Request)
  • How many passed vs failed

Debugging Failed Tests​

Step 1: Run Individual Test​

python api/perl/tests/test_search.py

Step 2: Check Error Messages​

Look for:

  • HTTP status codes (400, 404, 500, etc.)
  • Error messages in response body
  • Which specific test scenario failed

Step 3: Verify Server is Running​

curl http://localhost:9200/health

Should return:

{"status": "ok"}

Step 4: Check Test Data​

Some tests require existing collections. Check if:

  • Server has collections
  • Collections have documents
  • Schema matches test expectations

Step 5: Enable Verbose Output​

Modify test to print more details:

result = client.search(collection_name, params)
if not result.is_success():
    print(f"  ✗ Failed: {result.get_status_code()}")
    body = result.get_body()
    print(f"  Error details: {body}")  # Add this line

Real-World Usage Examples​

Example: Testing Your Own Collection​

# Modify test to use your collection
collection_name = 'my_collection'  # Your collection name

# Run test scenarios
results = client.search(collection_name, {
    'q': 'your search term',
    'query_by': 'title,content',
    'limit': 10
})

if results.is_success():
    body = results.get_body()
    hits = body.get('hits', [])
    print(f"Found {len(hits)} results")
    for hit in hits:
        print(f"  - {hit.get('document', {}).get('title', 'N/A')}")

Example: Testing with Authentication​

client = Client('http://localhost:9200')
client.set_auth_token('your_token_here', 'bearer')

# Now all requests use authentication
result = client.health()

Example: Testing Against Remote Server​

BASE_URL = 'http://your-server:9200'  # Change this
client = Client(BASE_URL)

Detailed Test Examples by Category​

Example: Complete test_get_document.py Walkthrough​

def test_get_document():
    """Test getting individual documents"""
    
    # === SETUP ===
    client = Client(BASE_URL)
    tests_passed = 0
    tests_failed = 0
    
    # Get or create collection
    collections = client.list_collections(0, 1)
    collection_name = None
    created_collection = False
    
    if collections.is_success():
        body = collections.get_body()
        if body.get('collections'):
            collection_name = body['collections'][0]['name']
    
    if not collection_name:
        # Create test collection
        collection_name = 'test_get_doc_' + str(int(time.time()))
        schema = {
            'fields': [
                {'name': 'title', 'type': 'string'},
                {'name': 'content', 'type': 'string'},
                {'name': 'price', 'type': 'float'}
            ]
        }
        client.collections_api().create(collection_name, schema)
        created_collection = True
        
        # Add test document
        test_doc = {
            'id': 'test_doc_1',
            'title': 'Test Document',
            'content': 'This is test content',
            'price': 99.99
        }
        client.documents_api().add(collection_name, test_doc)
    
    # === TEST 1: Get existing document ===
    print("Test 1: Get document by ID")
    documents = client.list_documents(collection_name, {'offset': 0, 'limit': 1})
    if documents.is_success():
        body = documents.get_body()
        docs = body.get('documents', [])
        if docs:
            doc_id = docs[0].get('id')
            
            # Get the document
            result = client.get_document(collection_name, doc_id)
            if result.is_success():
                doc_body = result.get_body()
                print(f"  ✓ Retrieved document: {doc_id}")
                print(f"  Document title: {doc_body.get('title', 'N/A')}")
                tests_passed += 1
            else:
                print(f"  ✗ Failed: {result.get_status_code()}")
                tests_failed += 1
    
    # === TEST 2: Get non-existent document ===
    print("\nTest 2: Get non-existent document")
    result = client.get_document(collection_name, '__nonexistent_doc__')
    if not result.is_success() and result.get_status_code() in [404, 400]:
        print(f"  ✓ Correctly returned error: {result.get_status_code()}")
        tests_passed += 1
    else:
        print(f"  ✗ Unexpected response: {result.get_status_code()}")
        tests_failed += 1
    
    # === CLEANUP ===
    if created_collection:
        client.collections_api().delete(collection_name)
    
    # === SUMMARY ===
    print(f"\nTest Summary: {tests_passed} passed, {tests_failed} failed")

What This Test Does:

  1. Gets or creates a test collection
  2. Adds a test document
  3. Retrieves the document by ID (should succeed)
  4. Tries to get a non-existent document (should return 404)
  5. Cleans up test data
  6. Reports pass/fail summary

Example: Complete test_stats.py Walkthrough​

def test_stats():
    """Test stats and metrics endpoints"""
    
    client = Client(BASE_URL)
    tests_passed = 0
    tests_failed = 0
    
    # Test 1: Get server stats
    print("Test 1: GET /stats")
    stats = client.stats()
    if stats.is_success():
        body = stats.get_body()
        print(f"  ✓ Stats retrieved")
        print(f"  Collections: {body.get('collections', 'N/A')}")
        print(f"  Documents: {body.get('documents', 'N/A')}")
        tests_passed += 1
    
    # Test 3: Get connections
    print("\nTest 3: GET /connections")
    connections = client.execute_request('GET', '/connections')
    if connections.is_success():
        print(f"  ✓ Connections retrieved")
        body = connections.get_body()
        # Shows active connection count
        tests_passed += 1
    
    # Test 4: Get RocksDB stats
    print("\nTest 4: GET /rocksdb")
    rocksdb = client.execute_request('GET', '/rocksdb')
    if rocksdb.is_success():
        print(f"  ✓ RocksDB stats retrieved")
        # RocksDB storage engine statistics
        tests_passed += 1
    
    # Test 5: Get document total
    print("\nTest 5: GET /doctotal")
    doctotal = client.execute_request('GET', '/doctotal')
    if doctotal.is_success():
        print(f"  ✓ Document total retrieved")
        body = doctotal.get_body()
        # Total document count across all collections
        tests_passed += 1
    
    print(f"\nStats Test Summary: {tests_passed} passed, {tests_failed} failed")

What This Test Does:

  • Tests all statistics endpoints
  • Verifies server monitoring data is accessible
  • No cleanup needed (read-only operations)

Example: Complete test_filter_search.py Walkthrough​

def test_filter_search():
    """Test search with filters"""
    
    # Create test collection with products
    collection_name = 'test_filter_' + str(int(time.time()))
    schema = {
        'fields': [
            {'name': 'title', 'type': 'string'},
            {'name': 'category', 'type': 'string'},
            {'name': 'price', 'type': 'float'},
            {'name': 'active', 'type': 'bool'}
        ]
    }
    client.collections_api().create(collection_name, schema)
    
    # Add products
    products = [
        {'id': 'p1', 'title': 'Laptop', 'category': 'electronics', 'price': 999.99, 'active': True},
        {'id': 'p2', 'title': 'Phone', 'category': 'electronics', 'price': 699.99, 'active': True},
        {'id': 'p3', 'title': 'Book', 'category': 'books', 'price': 19.99, 'active': False},
        {'id': 'p4', 'title': 'Tablet', 'category': 'electronics', 'price': 499.99, 'active': True}
    ]
    for product in products:
        client.documents_api().add(collection_name, product)
    
    # Test 1: Filter by category
    print("Test 1: Filter by category")
    result = client.search(collection_name, {
        'q': '*',
        'query_by': 'title',
        'filter_by': 'category:electronics',
        'limit': 10
    })
    if result.is_success():
        body = result.get_body()
        hits = body.get('hits', [])
        print(f"  ✓ Found {len(hits)} electronics products")
        # Should find 3 products (Laptop, Phone, Tablet)
        tests_passed += 1
    
    # Test 2: Filter by price range
    print("\nTest 2: Filter by price range")
    result = client.search(collection_name, {
        'q': '*',
        'query_by': 'title',
        'filter_by': 'price:>=500',
        'limit': 10
    })
    if result.is_success():
        body = result.get_body()
        hits = body.get('hits', [])
        print(f"  ✓ Found {len(hits)} products >= $500")
        # Should find Laptop and Phone
        tests_passed += 1
    
    # Test 3: Multiple filters
    print("\nTest 3: Multiple filters")
    result = client.search(collection_name, {
        'q': '*',
        'query_by': 'title',
        'filter_by': 'category:electronics && price:>=500',
        'limit': 10
    })
    if result.is_success():
        body = result.get_body()
        hits = body.get('hits', [])
        print(f"  ✓ Found {len(hits)} electronics products >= $500")
        # Should find Laptop and Phone
        tests_passed += 1
    
    # Cleanup
    client.collections_api().delete(collection_name)

What This Test Does:

  • Creates a product catalog collection
  • Tests filtering by category
  • Tests filtering by price range
  • Tests combining multiple filters
  • Demonstrates real-world e-commerce search scenarios

Example: Complete test_sort_search.py Walkthrough​

def test_sort_search():
    """Test search with sorting"""
    
    # Create collection with products
    collection_name = 'test_sort_' + str(int(time.time()))
    schema = {
        'fields': [
            {'name': 'title', 'type': 'string'},
            {'name': 'price', 'type': 'float'},
            {'name': 'rating', 'type': 'float'}
        ]
    }
    client.collections_api().create(collection_name, schema)
    
    # Add products with different prices and ratings
    products = [
        {'id': 's1', 'title': 'Product A', 'price': 100.0, 'rating': 4.5},
        {'id': 's2', 'title': 'Product B', 'price': 50.0, 'rating': 4.8},
        {'id': 's3', 'title': 'Product C', 'price': 200.0, 'rating': 4.2},
        {'id': 's4', 'title': 'Product D', 'price': 75.0, 'rating': 4.9}
    ]
    for product in products:
        client.documents_api().add(collection_name, product)
    
    # Test 1: Sort by price ascending (cheapest first)
    print("Test 1: Sort by price ascending")
    result = client.search(collection_name, {
        'q': '*',
        'query_by': 'title',
        'sort_by': 'price:asc',
        'limit': 10
    })
    if result.is_success():
        body = result.get_body()
        hits = body.get('hits', [])
        print(f"  ✓ Sorted {len(hits)} products by price (ascending)")
        # Results should be: Product B ($50), Product D ($75), Product A ($100), Product C ($200)
        if len(hits) > 0:
            first_price = hits[0].get('document', {}).get('price', 0)
            print(f"  First product price: ${first_price}")
        tests_passed += 1
    
    # Test 2: Sort by price descending (most expensive first)
    print("\nTest 2: Sort by price descending")
    result = client.search(collection_name, {
        'q': '*',
        'query_by': 'title',
        'sort_by': 'price:desc',
        'limit': 10
    })
    if result.is_success():
        body = result.get_body()
        hits = body.get('hits', [])
        print(f"  ✓ Sorted {len(hits)} products by price (descending)")
        # Results should be: Product C ($200), Product A ($100), Product D ($75), Product B ($50)
        if len(hits) > 0:
            first_price = hits[0].get('document', {}).get('price', 0)
            print(f"  First product price: ${first_price}")
        tests_passed += 1
    
    # Test 3: Sort by rating (highest rated first)
    print("\nTest 3: Sort by rating descending")
    result = client.search(collection_name, {
        'q': '*',
        'query_by': 'title',
        'sort_by': 'rating:desc',
        'limit': 10
    })
    if result.is_success():
        body = result.get_body()
        hits = body.get('hits', [])
        print(f"  ✓ Sorted {len(hits)} products by rating (descending)")
        # Results should be: Product D (4.9), Product B (4.8), Product A (4.5), Product C (4.2)
        if len(hits) > 0:
            first_rating = hits[0].get('document', {}).get('rating', 0)
            print(f"  First product rating: {first_rating}")
        tests_passed += 1
    
    # Cleanup
    client.collections_api().delete(collection_name)

What This Test Does:

  • Creates products with prices and ratings
  • Tests ascending sort (cheapest first)
  • Tests descending sort (most expensive first)
  • Tests sorting by different fields (rating)
  • Demonstrates e-commerce sorting scenarios

Example: Complete test_import_documents.py Walkthrough​

def test_import_documents():
    """Test document import operations"""
    
    # Create collection
    collection_name = 'test_import_' + str(int(time.time()))
    schema = {
        'fields': [
            {'name': 'title', 'type': 'string'},
            {'name': 'content', 'type': 'string'},
            {'name': 'category', 'type': 'string'},
            {'name': 'price', 'type': 'float'}
        ]
    }
    client.collections_api().create(collection_name, schema)
    
    # Test 1: Import single document
    print("Test 1: Import single document")
    single_doc = [{
        'id': 'import1',
        'title': 'Imported Document 1',
        'content': 'Content 1',
        'category': 'import',
        'price': 10.0
    }]
    result = client.documents_api().import_documents(collection_name, single_doc)
    if result.is_success():
        print(f"  ✓ Imported 1 document")
        tests_passed += 1
    
    # Test 2: Import multiple documents (bulk)
    print("\nTest 2: Import multiple documents")
    multiple_docs = [
        {'id': 'import2', 'title': 'Doc 2', 'content': 'Content 2', 'category': 'import', 'price': 20.0},
        {'id': 'import3', 'title': 'Doc 3', 'content': 'Content 3', 'category': 'import', 'price': 30.0},
        {'id': 'import4', 'title': 'Doc 4', 'content': 'Content 4', 'category': 'import', 'price': 40.0},
        {'id': 'import5', 'title': 'Doc 5', 'content': 'Content 5', 'category': 'import', 'price': 50.0}
    ]
    result = client.documents_api().import_documents(collection_name, multiple_docs)
    if result.is_success():
        print(f"  ✓ Imported {len(multiple_docs)} documents in one operation")
        tests_passed += 1
    
    # Test 3: Verify all documents were imported
    print("\nTest 3: Verify imported documents")
    documents = client.list_documents(collection_name, {'offset': 0, 'limit': 10})
    if documents.is_success():
        body = documents.get_body()
        docs = body.get('documents', [])
        expected_count = len(single_doc) + len(multiple_docs)
        print(f"  ✓ Found {len(docs)} documents (expected {expected_count})")
        if len(docs) >= expected_count:
            tests_passed += 1
        else:
            print(f"  Warning: Expected {expected_count} but found {len(docs)}")
            tests_failed += 1
    
    # Cleanup
    client.collections_api().delete(collection_name)

What This Test Does:

  • Tests importing a single document
  • Tests bulk import (multiple documents at once)
  • Verifies all documents were imported correctly
  • Demonstrates efficient bulk operations

How Each Test Category Works​

Authentication Tests​

Purpose: Verify authentication mechanisms work correctly

How it works:

  1. Test without authentication (should work if server allows)
  2. Set bearer token and test again
  3. Change to API key and test again
  4. Verify all authentication methods work

Example Flow:

Client created → Health check (no auth) → Set token → Health check (with auth) → Change to API key → Health check (API key)

Collection Tests​

Purpose: Test collection lifecycle

How it works:

  1. Create collection with schema
  2. Get collection details
  3. Update collection schema
  4. List collections
  5. Delete collection
  6. Verify deletion

Example Flow:

Create → Get → Update → List → Delete → Verify Deleted

Document Tests​

Purpose: Test document CRUD operations

How it works:

  1. Create collection
  2. Insert document
  3. Get document by ID
  4. Update document
  5. List documents
  6. Delete document
  7. Verify deletion
  8. Clean up collection

Example Flow:

Create Collection → Insert Doc → Get Doc → Update Doc → List Docs → Delete Doc → Verify → Cleanup

Search Tests​

Purpose: Test various search scenarios

How it works:

  1. Create collection with test data
  2. Test basic search
  3. Test filtered search
  4. Test sorted search
  5. Test paginated search
  6. Test wildcard search
  7. Clean up

Example Flow:

Create Collection → Add Test Data → Basic Search → Filtered Search → Sorted Search → Paginated Search → Wildcard Search → Cleanup

Bulk Operations Tests​

Purpose: Test efficient bulk operations

How it works:

  1. Create collection
  2. Prepare array of documents
  3. Import all documents in one call
  4. Verify all documents imported
  5. Bulk delete by filter
  6. Verify deletion
  7. Clean up

Example Flow:

Create Collection → Prepare Docs Array → Bulk Import → Verify Count → Bulk Delete → Verify Deleted → Cleanup

Test Output Interpretation​

Understanding Pass/Fail Indicators​

  • ✓ (Checkmark): Test passed
  • ✗ (X mark): Test failed
  • (Warning): Test passed but with warnings

Reading Test Summaries​

Test Summary: 5 passed, 0 failed

This means:

  • 5 test scenarios passed
  • 0 test scenarios failed
  • Overall test: PASSED
Test Summary: 3 passed, 2 failed

This means:

  • 3 test scenarios passed
  • 2 test scenarios failed
  • Overall test: FAILED (some scenarios failed)

Understanding Error Codes​

  • 200: Success
  • 201: Created (success)
  • 400: Bad Request (invalid input)
  • 401: Unauthorized (authentication required)
  • 404: Not Found (resource doesn't exist)
  • 422: Unprocessable Entity (validation error)
  • 500: Internal Server Error

Best Practices​

  1. Always Clean Up: Tests should delete test data
  2. Use Unique Names: Use timestamps to avoid conflicts
  3. Check Results: Always verify operation success
  4. Handle Errors: Test both success and failure cases
  5. Isolate Tests: Each test should be independent
  6. Track Pass/Fail: Use counters to track test results
  7. Provide Context: Print what test is running
  8. Show Results: Display actual vs expected values

Next Steps​

  • Run individual tests to understand each one
  • Modify tests to match your use cases
  • Add new tests for your specific scenarios
  • Integrate tests into your CI/CD pipeline
  • Review test output to understand API behavior
  • Use tests as examples for your own code

For more information, see the main Testing Guide.