Detailed Testing Guide: How Tests Work
This guide provides in-depth explanations of how each test works, with complete code examples and step-by-step walkthroughs.
Table of Contents
- Test Architecture Deep Dive
- Test Execution Flow
- Complete Test Walkthroughs
- Test Patterns Explained
- Understanding Test Output
- Common Scenarios
Test Architecture Deep Dive
File Structure
Every test file has this exact structure:
#!/usr/bin/env python3
"""
Test Description
"""
import sys
import os
import time # For unique names
# Change to script directory (working directory)
script_dir = os.path.dirname(os.path.abspath(__file__))
os.chdir(script_dir)
# Path setup - CRITICAL for imports
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'
def test_feature():
"""Main test function"""
# Test code here
pass
if __name__ == '__main__':
test_feature()
Working Directory: All tests automatically change to their own directory using os.chdir(script_dir). This ensures:
- Tests work when run from any location
- Working directory is predictable
- File operations use correct paths
For a comprehensive guide to working directory management, path resolution, and test execution lifecycle, see How Tests Work.
Why This Structure?
- Shebang Line (
#!/usr/bin/env python3): Makes script executable - Path Setup: Allows importing Python client from any API directory
- Client Import: Uses the Python client library (works for all APIs)
- Main Guard: Only runs when executed directly, not when imported
Test Execution Flow
Phase 1: Initialization
# Step 1: Create client
client = Client(BASE_URL)
# Step 2: Initialize counters
tests_passed = 0
tests_failed = 0
# Step 3: Print header
print("=" * 70)
print("TEST: Feature Name")
print("-" * 70)
What happens:
- Client connects to server
- Counters track test results
- Header shows what's being tested
Phase 2: Setup
# Option A: Use existing 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']
use_existing = True
# Option B: Create test collection
if not use_existing:
collection_name = 'test_feature_' + str(int(time.time()))
schema = {'fields': [...]}
client.collections_api().create(collection_name, schema)
created_collection = True
What happens:
- Tries to use existing data (faster)
- Creates test data if needed (isolated)
- Tracks if we created data (for cleanup)
Phase 3: Test Execution
# Test 1: Basic operation
print("Test 1: Basic operation")
result = client.some_operation(params)
if result.is_success():
print(" ✓ Operation successful")
tests_passed += 1
else:
print(f" ✗ Operation failed: {result.get_status_code()}")
tests_failed += 1
# Test 2: With parameters
print("\nTest 2: With parameters")
result = client.operation_with_params(different_params)
# ... validation ...
What happens:
- Each test scenario runs independently
- Results are checked immediately
- Pass/fail is tracked per scenario
Phase 4: Cleanup
# Only cleanup if we created the data
if created_collection:
delete_result = client.collections_api().delete(collection_name)
if delete_result.is_success():
print(f"\n✓ Cleaned up collection: {collection_name}")
What happens:
- Only deletes data we created
- Leaves existing data untouched
- Ensures no test pollution
Phase 5: Summary
print("\n" + "=" * 70)
print(f"Test Summary: {tests_passed} passed, {tests_failed} failed")
print("=" * 70 + "\n")
What happens:
- Shows final statistics
- Clear pass/fail indication
- Easy to see test results
Complete Test Walkthroughs
Walkthrough 1: test_auth.py - Step by Step
def test_auth():
"""Test authentication"""
# === INITIALIZATION ===
print("=" * 70)
print("TEST: Authentication")
print("-" * 70)
client = Client('http://localhost:9200')
# === TEST 1: Without Authentication ===
print("Test 1: Health check without auth")
health = client.health()
print(f"Status: {health.get_status_code()}")
# Expected: 200 (if server allows unauthenticated access)
# OR: 401 (if server requires authentication)
# === TEST 2: With Bearer Token ===
print("\nTest 2: Health check with bearer token")
test_token = "test_token_12345"
client.set_auth_token(test_token, 'bearer')
print(f"Set bearer token: {test_token[:8]}...")
health = client.health()
print(f"Status: {health.get_status_code()}")
# Expected: 200 (if token is valid)
# OR: 401 (if token is invalid)
# === TEST 3: With API Key ===
print("\nTest 3: Health check with API key")
client.set_auth_token(test_token, 'api-key')
print(f"Set API key: {test_token[:8]}...")
health = client.health()
print(f"Status: {health.get_status_code()}")
# Expected: 200 (if API key is valid)
# === SUMMARY ===
print("\n✓ Authentication test completed\n")
Execution Timeline:
0.0s: Test starts
0.1s: Client created
0.2s: Health check without auth → 200
0.3s: Bearer token set
0.4s: Health check with bearer → 200
0.5s: API key set
0.6s: Health check with API key → 200
0.7s: Test completes
Walkthrough 2: test_search.py - Complete Flow
def test_search():
"""Test search with 5 scenarios"""
client = Client(BASE_URL)
tests_passed = 0
tests_failed = 0
# === SETUP: Create Test Data ===
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)
# Result: Collection created with 4 fields
# 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)
# Result: 3 documents added to collection
# === TEST 1: Basic Search ===
print("Test 1: Basic search")
search_params = {
'q': 'test', # Search query
'query_by': 'title,content', # Search in these fields
'limit': 10 # Return up to 10 results
}
results = client.search(collection_name, search_params)
# API Call: GET /collections/{name}/documents/search?q=test&query_by=title,content&limit=10
if results.is_success():
body = results.get_body()
hits = body.get('hits', [])
print(f" ✓ Found {len(hits)} results")
# Expected: 2 results (doc1 and doc2 contain "test")
tests_passed += 1
else:
print(f" ✗ Search failed: {results.get_status_code()}")
tests_failed += 1
# === TEST 2: Filtered Search ===
print("\nTest 2: Search with filters")
search_params = {
'q': '*', # Search all
'query_by': 'title', # Search in title
'filter_by': 'category:test', # Filter by category
'limit': 5
}
results = client.search(collection_name, search_params)
# API Call: GET /collections/{name}/documents/search?q=*&query_by=title&filter_by=category:test&limit=5
if results.is_success():
body = results.get_body()
hits = body.get('hits', [])
print(f" ✓ Filtered search found {len(hits)} results")
# Expected: 2 results (only doc1 and doc2 have category='test')
tests_passed += 1
# === TEST 3: Sorted Search ===
print("\nTest 3: Search with sorting")
search_params = {
'q': '*',
'query_by': 'title',
'sort_by': 'price:asc', # Sort by price ascending
'limit': 10
}
results = client.search(collection_name, search_params)
# Results should be: doc1 ($10), doc3 ($15), doc2 ($20)
# === TEST 4: Paginated Search ===
print("\nTest 4: Search with pagination")
search_params = {
'q': '*',
'query_by': 'title,content',
'offset': 0, # Start from first result
'limit': 2 # Return only 2 results
}
results = client.search(collection_name, search_params)
# Expected: 2 results (doc1 and doc2)
# === TEST 5: Wildcard Search ===
print("\nTest 5: Wildcard search")
search_params = {
'q': 'test*', # Wildcard: matches "test", "testing", etc.
'query_by': 'title',
'limit': 10
}
results = client.search(collection_name, search_params)
# Expected: 2 results (doc1 and doc2 titles start with "Test")
# === CLEANUP ===
client.collections_api().delete(collection_name)
# Collection and all documents deleted
# === SUMMARY ===
print(f"\nSearch Test Summary: {tests_passed} passed, {tests_failed} failed")
Complete Execution Flow:
1. Create collection (200ms)
2. Add 3 documents (300ms)
3. Test 1: Basic search (150ms) → 2 results ✓
4. Test 2: Filtered search (120ms) → 2 results ✓
5. Test 3: Sorted search (130ms) → 3 results ✓
6. Test 4: Paginated search (110ms) → 2 results ✓
7. Test 5: Wildcard search (125ms) → 2 results ✓
8. Delete collection (100ms)
Total: ~1.2 seconds
Result: 5 passed, 0 failed
Walkthrough 3: test_bulk_operations.py - Bulk Import
def test_bulk_operations():
"""Test bulk document operations"""
# === SETUP ===
collection_name = 'test_bulk_' + str(int(time.time()))
schema = {'fields': [{'name': 'title', 'type': 'string'}]}
client.collections_api().create(collection_name, schema)
# === PREPARE BULK DATA ===
bulk_docs = []
for i in range(100): # Create 100 documents
bulk_docs.append({
'id': f'bulk_{i}',
'title': f'Bulk Document {i}'
})
# Result: Array with 100 document objects
# === BULK IMPORT ===
print("Test 1: Bulk import 100 documents")
import_result = client.documents_api().import_documents(collection_name, bulk_docs)
# API Call: POST /collections/{name}/documents/import
# Body: Array of 100 documents
# This is MUCH faster than adding 100 documents one by one
if import_result.is_success():
print(f" ✓ Bulk imported {len(bulk_docs)} documents")
# Time: ~500ms for 100 documents (vs ~30 seconds one-by-one)
tests_passed += 1
# === VERIFY IMPORT ===
print("\nTest 2: Verify bulk import")
documents = client.list_documents(collection_name, {'offset': 0, 'limit': 1000})
if documents.is_success():
body = documents.get_body()
docs = body.get('documents', [])
print(f" ✓ Found {len(docs)} documents")
if len(docs) == 100:
tests_passed += 1
else:
print(f" Expected 100, found {len(docs)}")
tests_failed += 1
# === BULK DELETE ===
print("\nTest 3: Bulk delete by filter")
delete_result = client.documents_api().delete_by_filter(collection_name, 'title:Bulk*')
# API Call: DELETE /collections/{name}/documents?filter_by=title:Bulk*
# Deletes all documents matching the filter
if delete_result.is_success():
print(f" ✓ Bulk delete successful")
tests_passed += 1
# === CLEANUP ===
client.collections_api().delete(collection_name)
Performance Comparison:
One-by-one import: 100 documents × 300ms = 30 seconds
Bulk import: 100 documents in one call = 500ms
Speed improvement: 60x faster!
Test Patterns Explained
Pattern 1: Create-Use-Cleanup
When to use: Testing operations that modify data
# 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)
Why: Ensures test isolation, no leftover data
Pattern 2: Use Existing or Create
When to use: Tests that can work with existing data
# Try existing first
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)
Why: Faster execution, works with existing data
Pattern 3: Pass/Fail Tracking
When to use: All tests with multiple scenarios
tests_passed = 0
tests_failed = 0
# Test scenario 1
result = client.operation1()
if result.is_success():
tests_passed += 1
else:
tests_failed += 1
# Test scenario 2
result = client.operation2()
if result.is_success():
tests_passed += 1
else:
tests_failed += 1
# Summary
print(f"Summary: {tests_passed} passed, {tests_failed} failed")
Why: Clear indication of what passed/failed
Pattern 4: Error Validation
When to use: Testing error handling
# Test should fail with 404
result = client.get_collection('__nonexistent__')
if not result.is_success() and result.get_status_code() == 404:
print(" ✓ Correctly returned 404")
tests_passed += 1
else:
print(f" ✗ Unexpected: {result.get_status_code()}")
tests_failed += 1
Why: Verifies error handling works correctly
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 this:
- Header shows test name and API
- Each test scenario shows result
- ✓ means passed
- Summary shows final count
- Cleanup confirmation
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
----------------------------------------------------------------==============================
Reading this:
- Test 2 failed with error 400
- Other tests passed
- Overall: 4 passed, 1 failed
- Need to investigate Test 2
Common Scenarios
Scenario 1: Testing a New Feature
def test_new_feature():
"""Test new API feature"""
client = Client(BASE_URL)
tests_passed = 0
tests_failed = 0
# Setup
collection_name = 'test_new_feature_' + str(int(time.time()))
# ... create collection ...
# Test the new feature
print("Test 1: New feature operation")
result = client.new_feature_operation(collection_name, params)
if result.is_success():
print(" ✓ New feature works")
tests_passed += 1
else:
print(f" ✗ New feature failed: {result.get_status_code()}")
tests_failed += 1
# Cleanup
client.collections_api().delete(collection_name)
print(f"Summary: {tests_passed} passed, {tests_failed} failed")
Scenario 2: Testing Error Cases
def test_error_cases():
"""Test error handling"""
client = Client(BASE_URL)
tests_passed = 0
tests_failed = 0
# Test 1: Invalid input
print("Test 1: Invalid input")
result = client.operation(invalid_params)
if not result.is_success() and result.get_status_code() == 400:
print(" ✓ Correctly rejected invalid input")
tests_passed += 1
# Test 2: Missing resource
print("Test 2: Missing resource")
result = client.get_resource('__nonexistent__')
if not result.is_success() and result.get_status_code() == 404:
print(" ✓ Correctly returned 404")
tests_passed += 1
# Test 3: Unauthorized
print("Test 3: Unauthorized access")
# Don't set auth token
result = client.protected_operation()
if not result.is_success() and result.get_status_code() == 401:
print(" ✓ Correctly returned 401")
tests_passed += 1
print(f"Summary: {tests_passed} passed, {tests_failed} failed")
Scenario 3: Performance Testing
def test_performance():
"""Test operation performance"""
import time
client = Client(BASE_URL)
# Test bulk import performance
collection_name = 'test_perf_' + str(int(time.time()))
client.collections_api().create(collection_name, schema)
# Prepare 1000 documents
bulk_docs = [{'id': f'doc_{i}', 'title': f'Doc {i}'} for i in range(1000)]
# Time the import
start_time = time.time()
result = client.documents_api().import_documents(collection_name, bulk_docs)
elapsed = time.time() - start_time
if result.is_success():
print(f" ✓ Imported 1000 documents in {elapsed:.2f}s")
print(f" Rate: {1000/elapsed:.0f} documents/second")
# Cleanup
client.collections_api().delete(collection_name)
Test Runner Deep Dive
How run_all_tests.py Works
def run_test(test_file):
"""Run a single test file"""
test_path = os.path.join(TEST_DIR, test_file)
# Execute test as subprocess
result = subprocess.run(
[sys.executable, test_path],
capture_output=True,
text=True,
timeout=60 # 60 second timeout per test
)
return {
'name': test_file,
'success': result.returncode == 0, # 0 = success
'returncode': result.returncode,
'stdout': result.stdout, # Test output
'stderr': result.stderr, # Error messages
'elapsed': elapsed_time
}
What happens:
- Finds all
test_*.pyfiles - Runs each as separate subprocess
- Captures output and errors
- Tracks execution time
- Aggregates results
Test Runner Output
----------------------------------------------------------------==============================
hlquery API Test Suite
----------------------------------------------------------------==============================
Found 31 test files
Running tests...
Running test_auth.py...
✓ PASSED (0.45s)
Running test_search.py...
✓ PASSED (1.23s)
Running test_bulk_operations.py...
✓ PASSED (0.89s)
... (28 more tests)
----------------------------------------------------------------==============================
TEST SUMMARY
----------------------------------------------------------------==============================
Total Tests: 31
Passed: 31
Failed: 0
Total Time: 18.45s
----------------------------------------------------------------==============================
What this tells you:
- All 31 tests ran
- All passed
- Total time: 18.45 seconds
- Average: ~0.6 seconds per test
Debugging Failed Tests
Step 1: Run Test Individually
python api/perl/tests/test_search.py
Step 2: Check Error Messages
Look for:
- HTTP status codes
- Error messages in response
- Which specific test scenario failed
Step 3: Add Debug Output
Modify test to print more:
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
print(f" Request params: {params}") # Add this
Step 4: Verify Server State
# Check server is running
curl http://localhost:9200/health
# Check collections
curl http://localhost:9200/collections
# Check specific collection
curl http://localhost:9200/collections/your_collection
Step 5: Check Test Data
Some tests require:
- Existing collections
- Documents in collections
- Specific schema fields
Verify these exist before running tests.
Real-World Examples
Example: E-Commerce Search Test
def test_ecommerce_search():
"""Test e-commerce product search"""
# Create products collection
collection_name = 'products_' + str(int(time.time()))
schema = {
'fields': [
{'name': 'name', 'type': 'string'},
{'name': 'description', 'type': 'string'},
{'name': 'price', 'type': 'float'},
{'name': 'category', 'type': 'string'},
{'name': 'in_stock', 'type': 'bool'},
{'name': 'rating', 'type': 'float'}
]
}
client.collections_api().create(collection_name, schema)
# Add products
products = [
{'id': 'p1', 'name': 'Laptop', 'description': 'High-performance laptop',
'price': 999.99, 'category': 'electronics', 'in_stock': True, 'rating': 4.5},
{'id': 'p2', 'name': 'Phone', 'description': 'Smartphone',
'price': 699.99, 'category': 'electronics', 'in_stock': True, 'rating': 4.8},
{'id': 'p3', 'name': 'Book', 'description': 'Programming book',
'price': 49.99, 'category': 'books', 'in_stock': False, 'rating': 4.2}
]
for product in products:
client.documents_api().add(collection_name, product)
# Test 1: Search by name
results = client.search(collection_name, {
'q': 'laptop',
'query_by': 'name,description',
'limit': 10
})
# Should find: Laptop
# Test 2: Filter by category and price
results = client.search(collection_name, {
'q': '*',
'query_by': 'name',
'filter_by': 'category:electronics && price:>=500',
'limit': 10
})
# Should find: Laptop, Phone
# Test 3: Sort by rating
results = client.search(collection_name, {
'q': '*',
'query_by': 'name',
'sort_by': 'rating:desc',
'limit': 10
})
# Should be: Phone (4.8), Laptop (4.5), Book (4.2)
# Test 4: Filter in-stock items
results = client.search(collection_name, {
'q': '*',
'query_by': 'name',
'filter_by': 'in_stock:true',
'limit': 10
})
# Should find: Laptop, Phone (Book is out of stock)
# Cleanup
client.collections_api().delete(collection_name)
Example: Content Management System Test
def test_cms_search():
"""Test CMS content search"""
# Create articles collection
collection_name = 'articles_' + str(int(time.time()))
schema = {
'fields': [
{'name': 'title', 'type': 'string'},
{'name': 'body', 'type': 'string'},
{'name': 'author', 'type': 'string'},
{'name': 'published_date', 'type': 'string'},
{'name': 'tags', 'type': 'string[]'}
]
}
client.collections_api().create(collection_name, schema)
# Add articles
articles = [
{'id': 'a1', 'title': 'Python Tutorial', 'body': 'Learn Python...',
'author': 'John', 'published_date': '2024-01-01', 'tags': ['python', 'tutorial']},
{'id': 'a2', 'title': 'JavaScript Guide', 'body': 'JavaScript basics...',
'author': 'Jane', 'published_date': '2024-01-15', 'tags': ['javascript', 'guide']},
{'id': 'a3', 'title': 'Python Advanced', 'body': 'Advanced Python...',
'author': 'John', 'published_date': '2024-02-01', 'tags': ['python', 'advanced']}
]
for article in articles:
client.documents_api().add(collection_name, article)
# Test 1: Search by content
results = client.search(collection_name, {
'q': 'python',
'query_by': 'title,body',
'limit': 10
})
# Should find: Python Tutorial, Python Advanced
# Test 2: Filter by author
results = client.search(collection_name, {
'q': '*',
'query_by': 'title',
'filter_by': 'author:John',
'limit': 10
})
# Should find: Python Tutorial, Python Advanced
# Test 3: Search with tags
results = client.search(collection_name, {
'q': 'tutorial',
'query_by': 'title,body,tags',
'limit': 10
})
# Should find: Python Tutorial
# Cleanup
client.collections_api().delete(collection_name)
Best Practices Summary
- Always Clean Up: Delete test data after tests
- Use Unique Names: Timestamps prevent conflicts
- Track Results: Count pass/fail for each scenario
- Handle Errors: Test both success and failure paths
- Isolate Tests: Each test should be independent
- Provide Context: Print what's being tested
- Show Values: Display actual vs expected
- Use Real Data: Tests should simulate real usage
Next Steps
- Run Tests: Execute individual tests to see them work
- Modify Tests: Adapt tests to your use cases
- Add Tests: Create tests for your specific scenarios
- Integrate: Add tests to your CI/CD pipeline
- Document: Keep test documentation updated
For more information:
- Main Testing Guide - Overview of all tests
- Testing Examples - More examples and patterns