#!/usr/bin/env python3 """ Phase 2 Integration Tests Comprehensive test suite for ChromaDB Phase 2 components: - HybridSearchService - ConversationIntelligence - QueryStrategyRouter """ import unittest import asyncio import json import time from pathlib import Path import sys # Add parent directory to path sys.path.insert(0, str(Path(__file__).parent.parent)) from core.search.hybrid_search_service import HybridSearchService from core.search.query_strategy_router import QueryStrategyRouter, SearchStrategy from core.intelligence.conversation_intelligence import ConversationIntelligence from core.storage.chroma_client import get_client as get_chroma_client class TestQueryStrategyRouter(unittest.TestCase): """Test cases for QueryStrategyRouter.""" def setUp(self): """Set up test router.""" self.router = QueryStrategyRouter() def test_simple_keyword_routing(self): """Test that simple keywords route to FAISS.""" simple_queries = [ "neural", "memory system", "API", "CONSTANT_NAME" ] for query in simple_queries: result = self.router.route_query(query) self.assertEqual( result['strategy'], 'faiss', f"Query '{query}' should route to FAISS" ) self.assertGreater(result['confidence'], 0.7) def test_complex_semantic_routing(self): """Test that semantic queries route to ChromaDB.""" semantic_queries = [ "how does neural network work", "what is memory optimization", "explain the search system", "why use ChromaDB" ] for query in semantic_queries: result = self.router.route_query(query) self.assertEqual( result['strategy'], 'chromadb', f"Query '{query}' should route to ChromaDB" ) self.assertIn('semantic', result['reasoning'].lower()) def test_hybrid_query_routing(self): """Test that complex patterns route to hybrid.""" hybrid_queries = [ "find patterns similar to neural networks", "search optimization but not FAISS" ] for query in hybrid_queries: result = self.router.route_query(query) self.assertEqual( result['strategy'], 'hybrid', f"Query '{query}' should route to hybrid" ) def test_metadata_filter_detection(self): """Test metadata filter detection.""" metadata_queries = [ "project:mira neural networks", "type:conversation after:2024-01-01", "tag:important complexity:high" ] for query in metadata_queries: result = self.router.route_query(query) self.assertEqual( result['strategy'], 'chromadb', "Metadata queries should route to ChromaDB" ) self.assertTrue(result['analysis']['has_metadata']) def test_complexity_scoring(self): """Test query complexity scoring.""" # Simple query simple_result = self.router.route_query("test") self.assertLess(simple_result['complexity'], 0.3) # Complex query complex_query = "how to implement a neural network with pattern recognition and memory optimization" complex_result = self.router.route_query(complex_query) self.assertGreater(complex_result['complexity'], 0.5) def test_context_hints(self): """Test context hint application.""" query = "search query" context = { 'force_strategy': 'chromadb', 'require_semantic': True } result = self.router.route_query(query, context) self.assertIn('Forced strategy', result['reasoning']) class TestConversationIntelligence(unittest.TestCase): """Test cases for ConversationIntelligence.""" def setUp(self): """Set up test intelligence.""" self.conv_intel = ConversationIntelligence() def test_conversation_analysis(self): """Test conversation analysis and metadata extraction.""" test_conversation = { "id": "test_conv_1", "timestamp": "2024-01-15T10:00:00Z", "participants": ["user", "assistant"], "messages": [ { "role": "user", "content": "How do I implement a neural network for pattern recognition?" }, { "role": "assistant", "content": "To implement a neural network, you'll need to design the architecture..." } ], "project": "neural_project" } metadata = self.conv_intel.analyze_conversation(test_conversation) # Verify metadata extraction self.assertEqual(metadata['conversation_id'], "test_conv_1") self.assertEqual(metadata['participant_count'], 2) self.assertEqual(metadata['message_count'], 2) self.assertIn('neural', metadata['topics']) self.assertEqual(metadata['project_context'], "neural_project") self.assertGreater(metadata['spark_intensity'], 0) def test_sentiment_analysis(self): """Test sentiment analysis.""" positive_conv = { "messages": [ {"content": "This is excellent work! Amazing implementation!"}, {"content": "Thank you! I'm grateful for your help."} ] } negative_conv = { "messages": [ {"content": "This is broken and causing errors."}, {"content": "I'm stuck and frustrated with this problem."} ] } positive_sentiment = self.conv_intel._analyze_sentiment(positive_conv) negative_sentiment = self.conv_intel._analyze_sentiment(negative_conv) self.assertEqual(positive_sentiment, "positive") self.assertEqual(negative_sentiment, "negative") def test_complexity_calculation(self): """Test conversation complexity calculation.""" simple_conv = { "messages": [ {"content": "Hello"}, {"content": "Hi there"} ] } complex_conv = { "messages": [ {"content": "```python\ndef neural_network():\n pass\n```"}, {"content": "Let's discuss the implementation of this neural network architecture..."} ] } simple_complexity = self.conv_intel._calculate_complexity(simple_conv) complex_complexity = self.conv_intel._calculate_complexity(complex_conv) self.assertEqual(simple_complexity, "low") self.assertIn(complex_complexity, ["medium", "high"]) def test_spark_intensity_measurement(self): """Test The Spark intensity measurement.""" engaged_conv = { "messages": [ {"role": "user", "content": "I have an idea for a new feature"}, {"role": "assistant", "content": "That's interesting! Let's explore it"}, {"role": "user", "content": "We could create something innovative"}, {"role": "assistant", "content": "I like where this is going"}, ] * 3 # Multiple back-and-forth } spark_intensity = self.conv_intel._measure_spark_intensity(engaged_conv) self.assertGreater(spark_intensity, 0.3) # Should show engagement class TestHybridSearchService(unittest.TestCase): """Test cases for HybridSearchService.""" def setUp(self): """Set up test service.""" self.search_service = HybridSearchService() def test_query_analysis(self): """Test query analysis logic.""" # Simple query simple_analysis = self.search_service._analyze_query("test") self.assertEqual(simple_analysis['strategy'], 'faiss') # Complex query complex_analysis = self.search_service._analyze_query("how to implement neural networks") self.assertEqual(complex_analysis['strategy'], 'chromadb') # Metadata query metadata_analysis = self.search_service._analyze_query("project:mira type:code") self.assertEqual(metadata_analysis['strategy'], 'chromadb') def test_collection_determination(self): """Test collection determination logic.""" # Conversation query conv_collection = self.search_service._determine_collection( "find conversation about neural networks", None ) self.assertEqual(conv_collection, 'mira_conversations') # Code query code_collection = self.search_service._determine_collection( "search for function implementation", None ) self.assertEqual(code_collection, 'mira_codebase') # Context-based context_collection = self.search_service._determine_collection( "search query", {'type': 'insight'} ) self.assertEqual(context_collection, 'mira_insights') def test_result_merging(self): """Test result merging logic.""" faiss_results = [ { 'content': 'Result 1', 'metadata': {'score': 0.9, 'source': 'faiss'} }, { 'content': 'Result 2', 'metadata': {'score': 0.7, 'source': 'faiss'} } ] chroma_results = [ { 'content': 'Result 1', # Duplicate 'metadata': {'score': 0.8, 'source': 'chromadb', 'extra': 'data'} }, { 'content': 'Result 3', 'metadata': {'score': 0.85, 'source': 'chromadb'} } ] merged = self.search_service._merge_results(faiss_results, chroma_results, 5) # Should have 3 unique results self.assertEqual(len(merged), 3) # First result should be merged with both engines first_result = merged[0] self.assertIn('faiss', first_result['metadata']['search_engines']) self.assertIn('chromadb', first_result['metadata']['search_engines']) self.assertIn('extra', first_result['metadata']) # ChromaDB metadata preserved def test_performance_tracking(self): """Test performance metric tracking.""" # Track some queries self.search_service._track_performance('faiss', 0.05) self.search_service._track_performance('chromadb', 1.5) self.search_service._track_performance('hybrid', 0.8) stats = self.search_service.get_performance_stats() self.assertIn('faiss', stats) self.assertIn('chromadb', stats) self.assertIn('hybrid', stats) self.assertEqual(stats['faiss']['count'], 1) self.assertAlmostEqual(stats['faiss']['avg_time'], 0.05) class TestIntegration(unittest.TestCase): """Integration tests for all Phase 2 components.""" @classmethod def setUpClass(cls): """Set up integration test environment.""" # Initialize ChromaDB chroma_client = get_chroma_client() chroma_client.initialize_collections() def test_end_to_end_search_flow(self): """Test complete search flow from router to results.""" # Create components router = QueryStrategyRouter() search_service = HybridSearchService() # Test query query = "how does the neural network handle pattern recognition" # Route query routing_decision = router.route_query(query) # Verify routing self.assertEqual(routing_decision['strategy'], 'chromadb') self.assertGreater(routing_decision['confidence'], 0.7) # Would execute search here if we had test data # results = asyncio.run(search_service.intelligent_search(query)) def test_conversation_storage_and_retrieval(self): """Test conversation storage and retrieval flow.""" conv_intel = ConversationIntelligence() # Store test conversation test_conv = { "id": "integration_test_1", "messages": [ {"content": "Testing ChromaDB integration"}, {"content": "Integration test successful"} ], "project": "test_project" } metadata = conv_intel.analyze_conversation(test_conv) # Verify storage self.assertIsNotNone(metadata) self.assertEqual(metadata['conversation_id'], "integration_test_1") # Test retrieval results = conv_intel.find_similar_conversations( "ChromaDB integration test", filters={'project': 'test_project'} ) # Would verify results if collection had data self.assertIsInstance(results, list) def run_tests(): """Run all tests.""" # Create test suite loader = unittest.TestLoader() suite = unittest.TestSuite() # Add test cases suite.addTests(loader.loadTestsFromTestCase(TestQueryStrategyRouter)) suite.addTests(loader.loadTestsFromTestCase(TestConversationIntelligence)) suite.addTests(loader.loadTestsFromTestCase(TestHybridSearchService)) suite.addTests(loader.loadTestsFromTestCase(TestIntegration)) # Run tests runner = unittest.TextTestRunner(verbosity=2) result = runner.run(suite) return result.wasSuccessful() if __name__ == "__main__": success = run_tests() exit(0 if success else 1)