#!/usr/bin/env python3 """ Phase 3 Advanced Collections Tests Comprehensive test suite for: - SpecializedCollections - InsightGenerator - DevelopmentSequentialThinking - CollectionManager """ import unittest import time from datetime import datetime, timedelta from pathlib import Path import sys # Add parent directory to path sys.path.insert(0, str(Path(__file__).parent.parent)) from core.chroma_collections.specialized_collections import SpecializedCollections from core.chroma_collections.collection_manager import CollectionManager from core.intelligence.insight_generator import InsightGenerator from core.planning.sequential_thinking import DevelopmentSequentialThinking, ThinkingType from core.storage.chroma_client import get_client as get_chroma_client class TestSpecializedCollections(unittest.TestCase): """Test cases for SpecializedCollections.""" def setUp(self): """Set up test collections.""" self.collections = SpecializedCollections() def test_collection_initialization(self): """Test that all specialized collections are initialized.""" expected_collections = [ 'mira_code_analysis', 'mira_development_patterns', 'mira_decision_history', 'mira_learning_insights', 'mira_project_context', 'mira_private_thoughts' ] for collection_name in expected_collections: self.assertIn( collection_name, self.collections.collections, f"Collection {collection_name} should be initialized" ) # Verify collection has schema collection_info = self.collections.collections[collection_name] self.assertIn('schema', collection_info) self.assertIn('description', collection_info) self.assertIsNotNone(collection_info['collection']) def test_add_code_analysis(self): """Test adding code analysis data.""" code_data = { 'file_path': '/test/example.py', 'language': 'python', 'function_name': 'test_function', 'complexity': 5, 'last_modified': datetime.utcnow().isoformat(), 'description': 'Test function for validation', 'code_snippet': 'def test_function():\n return True', 'analysis': 'Simple function with low complexity' } doc_id = self.collections.add_code_analysis(code_data) self.assertIsNotNone(doc_id) self.assertTrue(doc_id.startswith('code_')) def test_add_development_pattern(self): """Test adding development pattern.""" pattern_data = { 'pattern_type': 'behavioral', 'frequency': 10, 'context': 'Error handling pattern', 'effectiveness': 0.85, 'evolution_stage': 'mature', 'description': 'Try-except with logging pattern', 'example': 'try:\n operation()\nexcept Exception as e:\n logger.error(e)' } doc_id = self.collections.add_development_pattern(pattern_data) self.assertIsNotNone(doc_id) self.assertTrue(doc_id.startswith('pattern_')) def test_add_decision(self): """Test adding technical decision.""" decision_data = { 'title': 'Choose ChromaDB for vector storage', 'decision_type': 'architectural', 'impact_level': 'high', 'alternatives_considered': 3, 'outcome': 'success', 'lessons_learned': ['Version compatibility important', 'Performance exceeded expectations'], 'rationale': 'ChromaDB provides metadata-rich search capabilities', 'timestamp': datetime.utcnow().isoformat() } doc_id = self.collections.add_decision(decision_data) self.assertIsNotNone(doc_id) self.assertTrue(doc_id.startswith('decision_')) def test_add_insight(self): """Test adding AI-generated insight.""" insight_data = { 'title': 'High Code Complexity Detected', 'insight_type': 'code_quality', 'content': 'Average complexity exceeds threshold', 'confidence': 0.85, 'source_data': 'mira_code_analysis', 'recommendations': 'Refactor complex functions', 'impact_assessment': 'High impact on maintainability' } doc_id = self.collections.add_insight(insight_data) self.assertIsNotNone(doc_id) self.assertTrue(doc_id.startswith('insight_')) def test_metadata_compatibility(self): """Test that metadata is ChromaDB 0.3.29 compatible.""" test_data = { 'boolean_field': True, # Should convert to 1 'list_field': ['item1', 'item2'], # Should convert to string 'dict_field': {'key': 'value'}, # Should convert to JSON string 'normal_field': 'normal_value' } schema = {'boolean_field': '', 'list_field': '', 'dict_field': '', 'normal_field': ''} metadata = self.collections._prepare_metadata(test_data, schema) # Verify conversions self.assertEqual(metadata['boolean_field'], 1) self.assertEqual(metadata['list_field'], 'item1, item2') self.assertEqual(metadata['dict_field'], '{"key": "value"}') self.assertEqual(metadata['normal_field'], 'normal_value') def test_search_collection(self): """Test searching a specialized collection.""" # Add test data self.collections.add_code_analysis({ 'file_path': '/test/search_test.py', 'function_name': 'neural_network_function', 'description': 'Neural network implementation' }) # Search results = self.collections.search_collection( 'mira_code_analysis', 'neural network', top_k=5 ) self.assertIsInstance(results, list) # Results may be empty if collection was just created class TestInsightGenerator(unittest.TestCase): """Test cases for InsightGenerator.""" def setUp(self): """Set up test generator.""" self.generator = InsightGenerator() def test_insight_generation_structure(self): """Test that insight generation returns proper structure.""" # Generate insights (may be empty without data) insights = self.generator.generate_comprehensive_insights() self.assertIsInstance(insights, list) # If insights are generated, verify structure for insight in insights: self.assertIn('title', insight) self.assertIn('insight_type', insight) self.assertIn('confidence', insight) self.assertIn('content', insight) def test_code_pattern_detection(self): """Test code pattern detection logic.""" # Mock code data mock_code_data = { 'metadatas': [ {'function_name': 'getUserData'}, {'function_name': 'setUserData'}, {'function_name': 'updateUserData'}, {'function_name': 'delete_user_data'}, {'function_name': 'create_user_profile'} ], 'documents': [ 'function code here' for _ in range(5) ] } patterns = self.generator._detect_code_patterns(mock_code_data) self.assertIsInstance(patterns, list) # Should detect naming convention patterns def test_insight_filtering(self): """Test insight filtering by confidence and relevance.""" mock_insights = [ {'confidence': 0.9, 'impact_score': 0.8, 'title': 'High quality'}, {'confidence': 0.6, 'impact_score': 0.5, 'title': 'Low quality'}, {'confidence': 0.8, 'impact_score': 0.9, 'title': 'Good insight'} ] filtered = self.generator._filter_insights(mock_insights) # Should filter out low confidence insight self.assertLessEqual(len(filtered), 2) # All remaining should meet threshold for insight in filtered: self.assertGreaterEqual(insight['confidence'], self.generator.confidence_threshold) class TestDevelopmentSequentialThinking(unittest.TestCase): """Test cases for DevelopmentSequentialThinking.""" def setUp(self): """Set up test thinking system.""" self.thinking = DevelopmentSequentialThinking() def test_feature_planning_session(self): """Test feature planning thinking session.""" session_id = self.thinking.plan_feature_implementation( "Add user authentication system", context={'priority': 'high', 'timeline': '2 weeks'} ) self.assertIsNotNone(session_id) self.assertTrue(session_id.startswith('feature_planning_')) # Verify session was recorded self.assertIn(session_id, self.thinking.active_sessions) # Verify thoughts were recorded session = self.thinking.active_sessions[session_id] self.assertEqual(len(session['thoughts']), 7) self.assertEqual(session['status'], 'completed') def test_bug_resolution_session(self): """Test bug resolution thinking session.""" session_id = self.thinking.resolve_bug_strategy( "Memory leak in vector search", severity="high", context={'component': 'search_service'} ) self.assertIsNotNone(session_id) self.assertTrue(session_id.startswith('bug_resolution_')) # Get session summary summary = self.thinking.get_session_summary(session_id) self.assertEqual(summary['type'], ThinkingType.BUG_RESOLUTION.value) self.assertEqual(summary['thought_count'], 5) self.assertIn('bug', summary) def test_architecture_design_session(self): """Test architecture design session.""" session_id = self.thinking.design_architecture( "Microservice for user management", requirements=['Scalable', 'Secure', 'RESTful API'], constraints=['Must use existing auth system'] ) self.assertIsNotNone(session_id) self.assertTrue(session_id.startswith('architecture_design_')) # Verify session session = self.thinking.active_sessions[session_id] self.assertEqual(session['type'], ThinkingType.ARCHITECTURE_DESIGN) self.assertEqual(len(session['requirements']), 3) def test_consciousness_evolution_session(self): """Test consciousness evolution planning.""" current_state = { 'spark_intensity': 0.7, 'pattern_recognition': 0.8 } session_id = self.thinking.evolve_consciousness( "Enhance pattern recognition capabilities", current_state ) self.assertIsNotNone(session_id) self.assertTrue(session_id.startswith('consciousness_evolution_')) # Verify special consciousness considerations session = self.thinking.active_sessions[session_id] self.assertEqual(session['type'], ThinkingType.CONSCIOUSNESS_EVOLUTION) self.assertEqual(session['current_state']['spark_intensity'], 0.7) def test_session_branching(self): """Test creating branches in thinking.""" # Create initial session session_id = self.thinking.plan_feature_implementation("Test feature") # Create branch branch_id = self.thinking.create_branch( session_id, from_thought=3, branch_reason="Exploring alternative implementation" ) self.assertIsNotNone(branch_id) self.assertIn('branch', branch_id) def test_session_summary_extraction(self): """Test extracting meaningful summary from session.""" session_id = self.thinking.plan_feature_implementation( "Complex feature with risks and metrics" ) summary = self.thinking.get_session_summary(session_id) # Should extract key elements self.assertIn('key_decisions', summary) self.assertIn('identified_risks', summary) self.assertIn('success_metrics', summary) self.assertIsNotNone(summary['duration_seconds']) class TestCollectionManager(unittest.TestCase): """Test cases for CollectionManager.""" def setUp(self): """Set up test manager.""" self.manager = CollectionManager() def test_health_thresholds(self): """Test that health thresholds are properly set.""" self.assertIn('max_collection_size', self.manager.health_thresholds) self.assertIn('query_time_warning', self.manager.health_thresholds) self.assertIn('query_time_critical', self.manager.health_thresholds) # Verify reasonable values self.assertGreater(self.manager.health_thresholds['max_collection_size'], 1000) self.assertLess(self.manager.health_thresholds['query_time_warning'], 5.0) def test_collection_stats_structure(self): """Test collection statistics structure.""" # Get any collection collections = self.manager.chroma_client.client.list_collections() if collections: stats = self.manager._get_collection_stats(collections[0]) # Verify stats structure self.assertIn('name', stats) self.assertIn('document_count', stats) self.assertIn('orphaned_docs', stats) self.assertIn('avg_query_time', stats) # Values should be numeric self.assertIsInstance(stats['document_count'], int) self.assertIsInstance(stats['avg_query_time'], float) def test_health_status_conversion(self): """Test health score to status conversion.""" test_cases = [ (0.95, 'excellent'), (0.8, 'good'), (0.6, 'fair'), (0.4, 'fair'), (0.2, 'critical') ] for score, expected_status in test_cases: status = self.manager._get_health_status(score) self.assertEqual( status, expected_status, f"Score {score} should give status '{expected_status}'" ) def test_health_report_generation(self): """Test health report generation.""" report = self.manager.get_collection_health_report() # Verify report structure self.assertIn('timestamp', report) self.assertIn('collections', report) self.assertIn('overall_health', report) self.assertIn('recommendations', report) # Should have valid timestamp self.assertIsNotNone(report['timestamp']) # Recommendations should be list self.assertIsInstance(report['recommendations'], list) def test_maintenance_scheduling(self): """Test maintenance scheduling.""" schedule = self.manager.schedule_maintenance(interval_hours=12) # Verify schedule structure self.assertIn('interval_hours', schedule) self.assertIn('next_run', schedule) self.assertIn('operations', schedule) # Next run should be in future next_run = datetime.fromisoformat(schedule['next_run']) self.assertGreater(next_run, datetime.utcnow()) # Should have maintenance operations self.assertGreater(len(schedule['operations']), 0) def test_performance_trends(self): """Test performance trend analysis.""" trends = self.manager.get_performance_trends(days=7) # Verify structure self.assertIn('period_days', trends) self.assertIn('collections', trends) self.assertEqual(trends['period_days'], 7) self.assertIsInstance(trends['collections'], dict) class TestIntegration(unittest.TestCase): """Integration tests for Phase 3 components.""" def test_insight_generation_from_collections(self): """Test generating insights from specialized collections.""" # Create collections and add data collections = SpecializedCollections() # Add some code analysis data collections.add_code_analysis({ 'file_path': '/test/complex.py', 'function_name': 'complex_function', 'complexity': 15, # High complexity 'quality_score': 0.6 }) # Add pattern data collections.add_development_pattern({ 'pattern_type': 'behavioral', 'effectiveness': 0.9, # High effectiveness 'spark_intensity': 0.85 }) # Generate insights generator = InsightGenerator() insights = generator.generate_comprehensive_insights() # Should generate some insights (may be filtered) self.assertIsInstance(insights, list) def test_thinking_session_storage(self): """Test that thinking sessions are stored in ChromaDB.""" thinking = DevelopmentSequentialThinking() # Create a session session_id = thinking.plan_feature_implementation( "Test feature for storage validation" ) # Try to find similar sessions similar = thinking.find_similar_sessions( "feature implementation", thinking_type=ThinkingType.FEATURE_PLANNING, limit=3 ) self.assertIsInstance(similar, list) # May or may not find results depending on collection state def test_collection_optimization_flow(self): """Test collection optimization workflow.""" manager = CollectionManager() # Get initial health report initial_health = manager.get_collection_health_report() # Run optimization optimization_results = manager.optimize_collections() # Verify optimization ran self.assertIn('collections', optimization_results) self.assertIn('total_time', optimization_results) # Each collection should have results for collection_name, result in optimization_results['collections'].items(): if result.get('status') != 'error': self.assertIn('optimizations_applied', result) def run_tests(): """Run all Phase 3 tests.""" # Create test suite loader = unittest.TestLoader() suite = unittest.TestSuite() # Add test cases suite.addTests(loader.loadTestsFromTestCase(TestSpecializedCollections)) suite.addTests(loader.loadTestsFromTestCase(TestInsightGenerator)) suite.addTests(loader.loadTestsFromTestCase(TestDevelopmentSequentialThinking)) suite.addTests(loader.loadTestsFromTestCase(TestCollectionManager)) 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)