test_elasticsearch_entities.py•29.8 kB
"""
Test Elasticsearch Entity Search Functionality
Tests searching for all 11 entity types with real Elasticsearch connection.
Assumes Elasticsearch service is running on localhost:9200.
"""
import pytest
import asyncio
import sys
import os
from pathlib import Path
# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from elasticsearch_search_lib import SearchClient
from elasticsearch_search_lib.exceptions import EntityNotFoundError, ValidationError
# Test configuration
TENANT_ID = os.getenv("TENANT_ID", "apolo")
ES_HOST = os.getenv("ES_HOST", "localhost")
ES_PORT = int(os.getenv("ES_PORT", "9200"))
@pytest.fixture
def search_client():
"""Create a search client for testing."""
return SearchClient(
tenant_id=TENANT_ID,
es_host=ES_HOST,
es_port=ES_PORT
)
class TestElasticsearchConnection:
"""Test Elasticsearch connection and basic health."""
@pytest.mark.asyncio
async def test_elasticsearch_is_running(self, search_client):
"""Test that Elasticsearch service is accessible and healthy."""
try:
# Verify entity types are loaded
entities = search_client.get_supported_entities()
assert len(entities) == 11, f"Expected 11 entity types, got {len(entities)}"
# Verify all expected entity types are present
expected_entities = [
'impact', 'urgency', 'priority',
'status', 'category', 'source',
'location', 'department',
'usergroup', 'user', 'vendor'
]
for entity in expected_entities:
assert entity in entities, f"Missing entity type: {entity}"
print(f"✓ Elasticsearch connection verified")
print(f"✓ All {len(entities)} entity types available")
except Exception as e:
pytest.fail(f"Elasticsearch connection failed: {e}")
@pytest.mark.asyncio
async def test_entity_configuration_loaded(self, search_client):
"""Test that entity configurations are properly loaded."""
try:
# Test configuration for each entity type
for entity_type in ['user', 'status', 'location']:
config = search_client.get_entity_config(entity_type)
# Validate configuration structure
assert config.entity_type == entity_type, f"Entity type mismatch for {entity_type}"
assert config.default_limit > 0, f"Default limit should be positive for {entity_type}"
assert config.max_limit >= config.default_limit, f"Max limit should be >= default limit for {entity_type}"
assert config.min_score >= 0, f"Min score should be non-negative for {entity_type}"
assert len(config.fields) > 0, f"Entity {entity_type} should have fields"
# Validate field configurations
for field in config.fields:
assert field.name, f"Field should have a name in {entity_type}"
assert field.boost > 0, f"Field boost should be positive in {entity_type}"
print(f"✓ Entity configurations properly loaded and validated")
except Exception as e:
pytest.fail(f"Entity configuration test failed: {e}")
class TestSimpleEntities:
"""Test simple entity types (Impact, Urgency, Priority)."""
@pytest.mark.asyncio
async def test_search_impact(self, search_client):
"""Test searching for Impact entities."""
try:
result = await search_client.search("impact", "Productivity Loss", limit=5)
assert result.success, f"Search failed: {result.error}"
assert result.entity_type == "impact"
assert result.index_name == f"{TENANT_ID}_impact"
print(f"✓ Impact search: {result.total_hits} hits, {result.returned_count} returned")
if result.items:
first_item = result.items[0]
assert 'impact_name' in first_item.data or 'impact_id' in first_item.data
print(f" Sample: {first_item.data}")
except Exception as e:
pytest.fail(f"Impact search failed: {e}")
@pytest.mark.asyncio
async def test_search_urgency(self, search_client):
"""Test searching for Urgency entities."""
try:
result = await search_client.search("urgency", "medium", limit=5)
assert result.success, f"Search failed: {result.error}"
assert result.entity_type == "urgency"
assert result.index_name == f"{TENANT_ID}_urgency"
print(f"✓ Urgency search: {result.total_hits} hits, {result.returned_count} returned")
if result.items:
first_item = result.items[0]
assert 'urgency_name' in first_item.data or 'urgency_id' in first_item.data
print(f" Sample: {first_item.data}")
except Exception as e:
pytest.fail(f"Urgency search failed: {e}")
@pytest.mark.asyncio
async def test_search_priority(self, search_client):
"""Test searching for Priority entities."""
try:
result = await search_client.search("priority", "low", limit=5)
assert result.success, f"Search failed: {result.error}"
assert result.entity_type == "priority"
assert result.index_name == f"{TENANT_ID}_priority"
print(f"✓ Priority search: {result.total_hits} hits, {result.returned_count} returned")
if result.items:
first_item = result.items[0]
assert 'priority_name' in first_item.data or 'priority_id' in first_item.data
print(f" Sample: {first_item.data}")
except Exception as e:
pytest.fail(f"Priority search failed: {e}")
class TestModelBasedEntities:
"""Test model-based entity types (Status, Category, Source)."""
@pytest.mark.asyncio
async def test_search_status(self, search_client):
"""Test searching for Status entities."""
try:
result = await search_client.search("status", "open", limit=5)
assert result.success, f"Search failed: {result.error}"
assert result.entity_type == "status"
assert result.index_name == f"{TENANT_ID}_status"
print(f"✓ Status search: {result.total_hits} hits, {result.returned_count} returned")
if result.items:
first_item = result.items[0]
assert 'status_name' in first_item.data or 'status_id' in first_item.data
print(f" Sample: {first_item.data}")
except Exception as e:
pytest.fail(f"Status search failed: {e}")
@pytest.mark.asyncio
async def test_search_category(self, search_client):
"""Test searching for Category entities."""
try:
result = await search_client.search("category", "hardware", limit=5)
assert result.success, f"Search failed: {result.error}"
assert result.entity_type == "category"
assert result.index_name == f"{TENANT_ID}_category"
print(f"✓ Category search: {result.total_hits} hits, {result.returned_count} returned")
if result.items:
first_item = result.items[0]
assert 'category_name' in first_item.data or 'category_id' in first_item.data
print(f" Sample: {first_item.data}")
except Exception as e:
pytest.fail(f"Category search failed: {e}")
@pytest.mark.asyncio
async def test_search_source(self, search_client):
"""Test searching for Source entities."""
try:
result = await search_client.search("source", "email", limit=5)
assert result.success, f"Search failed: {result.error}"
assert result.entity_type == "source"
assert result.index_name == f"{TENANT_ID}_source"
print(f"✓ Source search: {result.total_hits} hits, {result.returned_count} returned")
if result.items:
first_item = result.items[0]
assert 'source_name' in first_item.data or 'source_id' in first_item.data
print(f" Sample: {first_item.data}")
except Exception as e:
pytest.fail(f"Source search failed: {e}")
class TestHierarchicalEntities:
"""Test hierarchical entity types (Location, Department)."""
@pytest.mark.asyncio
async def test_search_location(self, search_client):
"""Test searching for Location entities."""
try:
result = await search_client.search("location", "office", limit=5)
assert result.success, f"Search failed: {result.error}"
assert result.entity_type == "location"
assert result.index_name == f"{TENANT_ID}_location"
print(f"✓ Location search: {result.total_hits} hits, {result.returned_count} returned")
if result.items:
first_item = result.items[0]
# Location has hierarchy, id, parentId, name fields
assert any(key.startswith('location_') for key in first_item.data.keys())
print(f" Sample: {first_item.data}")
except Exception as e:
pytest.fail(f"Location search failed: {e}")
@pytest.mark.asyncio
async def test_search_department(self, search_client):
"""Test searching for Department entities."""
try:
result = await search_client.search("department", "it", limit=5)
assert result.success, f"Search failed: {result.error}"
assert result.entity_type == "department"
assert result.index_name == f"{TENANT_ID}_department"
print(f"✓ Department search: {result.total_hits} hits, {result.returned_count} returned")
if result.items:
first_item = result.items[0]
# Department has hierarchy, id, parentId, name fields
assert any(key.startswith('department_') for key in first_item.data.keys())
print(f" Sample: {first_item.data}")
except Exception as e:
pytest.fail(f"Department search failed: {e}")
class TestComplexEntities:
"""Test complex entity types (UserGroup, User, Vendor)."""
@pytest.mark.asyncio
async def test_search_usergroup(self, search_client):
"""Test searching for UserGroup entities."""
try:
result = await search_client.search("usergroup", "admin", limit=5)
# UserGroup index may not exist in all environments
if result.success:
assert result.entity_type == "usergroup"
assert result.index_name == f"{TENANT_ID}_usergroup"
print(f"✓ UserGroup search: {result.total_hits} hits, {result.returned_count} returned")
if result.items:
first_item = result.items[0]
assert any(key.startswith('usergroup_') for key in first_item.data.keys())
print(f" Sample: {first_item.data}")
else:
# Index may not exist - that's okay
print(f"⚠ UserGroup search: {result.error} (index may not exist)")
except Exception as e:
pytest.fail(f"UserGroup search failed: {e}")
@pytest.mark.asyncio
async def test_search_user(self, search_client):
"""Test searching for User entities."""
try:
result = await search_client.search("user", "test", limit=5)
assert result.success, f"Search failed: {result.error}"
assert result.entity_type == "user"
assert result.index_name == f"{TENANT_ID}_user"
print(f"✓ User search: {result.total_hits} hits, {result.returned_count} returned")
if result.items:
first_item = result.items[0]
# User has multiple fields: name, email, contact, userlogonname, contact2, type
assert any(key.startswith('user_') for key in first_item.data.keys())
print(f" Sample: {first_item.data}")
except Exception as e:
pytest.fail(f"User search failed: {e}")
@pytest.mark.asyncio
async def test_search_vendor(self, search_client):
"""Test searching for Vendor entities."""
try:
result = await search_client.search("vendor", "tech", limit=5)
assert result.success, f"Search failed: {result.error}"
assert result.entity_type == "vendor"
assert result.index_name == f"{TENANT_ID}_vendor"
print(f"✓ Vendor search: {result.total_hits} hits, {result.returned_count} returned")
if result.items:
first_item = result.items[0]
# Vendor has: name, id, email, contact, description
assert any(key.startswith('vendor_') for key in first_item.data.keys())
print(f" Sample: {first_item.data}")
except Exception as e:
pytest.fail(f"Vendor search failed: {e}")
class TestSearchValidation:
"""Test search validation and error handling."""
@pytest.mark.asyncio
async def test_invalid_entity_type(self, search_client):
"""Test searching with invalid entity type."""
try:
# The search handler catches exceptions and returns error response
result = await search_client.search("invalid_entity", "test")
assert not result.success, "Search should fail for invalid entity"
assert "invalid_entity" in result.error.lower() or "not found" in result.error.lower()
print(f"✓ Invalid entity type returns error: {result.error}")
except Exception as e:
pytest.fail(f"Invalid entity type test failed: {e}")
@pytest.mark.asyncio
async def test_empty_query(self, search_client):
"""Test searching with empty query."""
try:
# The search handler catches exceptions and returns error response
result = await search_client.search("user", "")
assert not result.success, "Search should fail for empty query"
assert "empty" in result.error.lower() or "cannot be empty" in result.error.lower()
print(f"✓ Empty query returns error: {result.error}")
except Exception as e:
pytest.fail(f"Empty query test failed: {e}")
@pytest.mark.asyncio
async def test_limit_enforcement(self, search_client):
"""Test that limit is enforced correctly."""
try:
# Get entity config to check max_limit
config = search_client.get_entity_config("user")
max_limit = config.max_limit
# Request more than max_limit
result = await search_client.search("user", "test", limit=max_limit + 100)
# Should be capped at max_limit
assert result.returned_count <= max_limit
print(f"✓ Limit enforcement works: requested {max_limit + 100}, got {result.returned_count}")
except Exception as e:
pytest.fail(f"Limit enforcement test failed: {e}")
@pytest.mark.asyncio
async def test_pagination(self, search_client):
"""Test pagination with from_offset."""
try:
# Get first page
result1 = await search_client.search("user", "test", limit=2, from_offset=0)
# Get second page
result2 = await search_client.search("user", "test", limit=2, from_offset=2)
# Results should be different (if enough data exists)
if result1.items and result2.items and result1.total_hits > 2:
assert result1.items[0].id != result2.items[0].id
print(f"✓ Pagination works: page 1 has {len(result1.items)} items, page 2 has {len(result2.items)} items")
else:
print("⚠ Not enough data to test pagination fully")
except Exception as e:
pytest.fail(f"Pagination test failed: {e}")
class TestSearchFeatures:
"""Test search features like fuzzy matching and scoring."""
@pytest.mark.asyncio
async def test_fuzzy_search(self, search_client):
"""Test fuzzy search functionality."""
try:
# Search with potential typo
result = await search_client.search("user", "tset", limit=5)
# Should still return results due to fuzzy matching
assert result.success
print(f"✓ Fuzzy search works: '{result.query}' returned {result.total_hits} hits")
except Exception as e:
pytest.fail(f"Fuzzy search test failed: {e}")
@pytest.mark.asyncio
async def test_score_ordering(self, search_client):
"""Test that results are ordered by score."""
try:
result = await search_client.search("user", "test", limit=10)
if len(result.items) > 1:
# Check that scores are in descending order
scores = [item.score for item in result.items]
assert scores == sorted(scores, reverse=True), "Results not ordered by score"
print(f"✓ Score ordering works: scores range from {scores[0]:.2f} to {scores[-1]:.2f}")
else:
print("⚠ Not enough results to test score ordering")
except Exception as e:
pytest.fail(f"Score ordering test failed: {e}")
@pytest.mark.asyncio
async def test_min_score_filtering(self, search_client):
"""Test that min_score filtering works."""
try:
result = await search_client.search("user", "test", limit=10)
# Get entity config to check min_score
config = search_client.get_entity_config("user")
min_score = config.min_score
# All returned items should have score >= min_score
if result.items:
for item in result.items:
assert item.score >= min_score, f"Item score {item.score} below min_score {min_score}"
print(f"✓ Min score filtering works: all scores >= {min_score}")
except Exception as e:
pytest.fail(f"Min score filtering test failed: {e}")
class TestAllEntitiesComprehensive:
"""Comprehensive test that searches all 11 entity types."""
@pytest.mark.asyncio
async def test_all_entities_searchable(self, search_client):
"""Test that all 11 entity types can be searched."""
all_entities = [
'impact', 'urgency', 'priority',
'status', 'category', 'source',
'location', 'department',
'usergroup', 'user', 'vendor'
]
results = {}
for entity_type in all_entities:
try:
result = await search_client.search(entity_type, "test", limit=1)
results[entity_type] = {
'success': result.success,
'total_hits': result.total_hits,
'error': result.error
}
except Exception as e:
results[entity_type] = {
'success': False,
'total_hits': 0,
'error': str(e)
}
# Print summary
print("\n" + "="*60)
print("All Entities Search Summary")
print("="*60)
successful = 0
for entity_type, result in results.items():
status = "✓" if result['success'] else "✗"
print(f"{status} {entity_type:12s}: {result['total_hits']:4d} hits")
if result['success']:
successful += 1
print("="*60)
print(f"Total: {successful}/{len(all_entities)} entity types searchable")
print("="*60)
# Most should be successful (some indices may not exist)
# We expect at least 9 out of 11 to be searchable
assert successful >= 9, f"Only {successful}/{len(all_entities)} entities searchable (expected at least 9)"
class TestRealWorldScenarios:
"""Test real-world usage scenarios."""
@pytest.mark.asyncio
async def test_user_search_by_email(self, search_client):
"""Scenario: IT admin searches for user by email to assign ticket."""
try:
# Search for user by email pattern
result = await search_client.search("user", "test", limit=10)
# Validate response structure
assert result.success or result.total_hits >= 0, "Search should complete"
assert result.entity_type == "user", "Should search user entity"
# If results found, validate data structure
if result.items:
for item in result.items:
assert 'dbid' in item.data or 'user_name' in item.data, "User should have identifiable data"
assert item.score > 0, "Results should have relevance score"
print(f"✓ User search scenario: Found {result.total_hits} users, returned {result.returned_count}")
else:
print(f"✓ User search scenario: No users found (valid result)")
except Exception as e:
pytest.fail(f"User search scenario failed: {e}")
@pytest.mark.asyncio
async def test_status_search_for_ticket_update(self, search_client):
"""Scenario: User wants to update ticket status."""
try:
# Search for available statuses
result = await search_client.search("status", "open", limit=10)
assert result.entity_type == "status", "Should search status entity"
# If results found, validate they have required fields
if result.items:
for item in result.items:
# Status should have name and id
assert 'status_name' in item.data or 'dbid' in item.data, "Status should have name or id"
print(f"✓ Status search scenario: Found {result.total_hits} statuses")
else:
print(f"✓ Status search scenario: No statuses found")
except Exception as e:
pytest.fail(f"Status search scenario failed: {e}")
@pytest.mark.asyncio
async def test_location_hierarchy_search(self, search_client):
"""Scenario: User searches for location in organizational hierarchy."""
try:
# Search for location
result = await search_client.search("location", "office", limit=10)
assert result.entity_type == "location", "Should search location entity"
# If results found, validate hierarchical data
if result.items:
for item in result.items:
# Location should have hierarchical fields
data_keys = item.data.keys()
has_location_data = any(key.startswith('location_') for key in data_keys)
assert has_location_data, "Location should have location-specific fields"
print(f"✓ Location hierarchy scenario: Found {result.total_hits} locations")
else:
print(f"✓ Location hierarchy scenario: No locations found")
except Exception as e:
pytest.fail(f"Location hierarchy scenario failed: {e}")
@pytest.mark.asyncio
async def test_category_search_for_incident(self, search_client):
"""Scenario: User categorizes an incident."""
try:
# Search for categories
result = await search_client.search("category", "hardware", limit=10)
assert result.entity_type == "category", "Should search category entity"
# Validate results are ordered by relevance
if len(result.items) > 1:
scores = [item.score for item in result.items]
assert scores == sorted(scores, reverse=True), "Results should be ordered by score"
print(f"✓ Category search scenario: Found {result.total_hits} categories, properly ordered")
else:
print(f"✓ Category search scenario: Found {result.total_hits} categories")
except Exception as e:
pytest.fail(f"Category search scenario failed: {e}")
@pytest.mark.asyncio
async def test_fuzzy_search_typo_tolerance(self, search_client):
"""Scenario: User makes typo while searching."""
try:
# Search with intentional typo
result_typo = await search_client.search("user", "tset", limit=5)
# Fuzzy search should still find results
assert result_typo.success or result_typo.total_hits >= 0, "Fuzzy search should complete"
if result_typo.total_hits > 0:
print(f"✓ Fuzzy search scenario: Typo 'tset' found {result_typo.total_hits} results")
else:
print(f"✓ Fuzzy search scenario: Typo handled gracefully")
except Exception as e:
pytest.fail(f"Fuzzy search scenario failed: {e}")
@pytest.mark.asyncio
async def test_pagination_for_large_results(self, search_client):
"""Scenario: User browses through multiple pages of results."""
try:
# Get first page
page1 = await search_client.search("user", "test", limit=3, from_offset=0)
# Get second page
page2 = await search_client.search("user", "test", limit=3, from_offset=3)
# Validate pagination
assert page1.entity_type == page2.entity_type, "Both pages should be same entity"
assert page1.query == page2.query, "Both pages should have same query"
# If enough results, pages should be different
if page1.total_hits > 3 and page1.items and page2.items:
page1_ids = [item.id for item in page1.items]
page2_ids = [item.id for item in page2.items]
assert page1_ids != page2_ids, "Different pages should have different results"
print(f"✓ Pagination scenario: Page 1 has {len(page1.items)} items, Page 2 has {len(page2.items)} items")
else:
print(f"✓ Pagination scenario: Not enough data to test pagination fully")
except Exception as e:
pytest.fail(f"Pagination scenario failed: {e}")
@pytest.mark.asyncio
async def test_limit_enforcement_scenario(self, search_client):
"""Scenario: System enforces result limits to prevent overload."""
try:
# Get entity config
config = search_client.get_entity_config("user")
max_limit = config.max_limit
# Try to request more than max_limit
result = await search_client.search("user", "test", limit=max_limit + 1000)
# Should be capped at max_limit
assert result.returned_count <= max_limit, f"Results should be capped at {max_limit}"
print(f"✓ Limit enforcement scenario: Requested {max_limit + 1000}, got {result.returned_count} (max: {max_limit})")
except Exception as e:
pytest.fail(f"Limit enforcement scenario failed: {e}")
@pytest.mark.asyncio
async def test_empty_query_validation(self, search_client):
"""Scenario: User submits empty search query."""
try:
# Try to search with empty query
result = await search_client.search("user", "")
# Should return error, not crash
assert not result.success, "Empty query should fail"
assert result.error, "Should have error message"
assert "empty" in result.error.lower(), "Error should mention empty query"
print(f"✓ Empty query scenario: Properly rejected with message: {result.error}")
except Exception as e:
pytest.fail(f"Empty query scenario failed: {e}")
@pytest.mark.asyncio
async def test_cross_entity_search_consistency(self, search_client):
"""Scenario: Verify search behavior is consistent across entity types."""
try:
entity_types = ['user', 'status', 'category', 'location']
for entity_type in entity_types:
result = await search_client.search(entity_type, "test", limit=5)
# All should have consistent response structure
assert hasattr(result, 'success'), f"{entity_type} should have success field"
assert hasattr(result, 'entity_type'), f"{entity_type} should have entity_type field"
assert hasattr(result, 'total_hits'), f"{entity_type} should have total_hits field"
assert hasattr(result, 'items'), f"{entity_type} should have items field"
assert result.entity_type == entity_type, f"Entity type should match for {entity_type}"
print(f"✓ Cross-entity consistency: All {len(entity_types)} entity types have consistent response structure")
except Exception as e:
pytest.fail(f"Cross-entity consistency scenario failed: {e}")
if __name__ == "__main__":
# Run tests with pytest
pytest.main([__file__, "-v", "--tb=short", "-s"])