Skip to main content

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

  1. Test Architecture Deep Dive
  2. Test Execution Flow
  3. Complete Test Walkthroughs
  4. Test Patterns Explained
  5. Understanding Test Output
  6. 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?

  1. Shebang Line (#!/usr/bin/env python3): Makes script executable
  2. Path Setup: Allows importing Python client from any API directory
  3. Client Import: Uses the Python client library (works for all APIs)
  4. 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:

  1. Finds all test_*.py files
  2. Runs each as separate subprocess
  3. Captures output and errors
  4. Tracks execution time
  5. 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

  1. Always Clean Up: Delete test data after tests
  2. Use Unique Names: Timestamps prevent conflicts
  3. Track Results: Count pass/fail for each scenario
  4. Handle Errors: Test both success and failure paths
  5. Isolate Tests: Each test should be independent
  6. Provide Context: Print what's being tested
  7. Show Values: Display actual vs expected
  8. Use Real Data: Tests should simulate real usage

Next Steps

  1. Run Tests: Execute individual tests to see them work
  2. Modify Tests: Adapt tests to your use cases
  3. Add Tests: Create tests for your specific scenarios
  4. Integrate: Add tests to your CI/CD pipeline
  5. Document: Keep test documentation updated

For more information: