Katamari MCP Server

""" Tests for ACP Adaptive Learning Components (Phase 2) Tests the adaptive learning engine, feedback collection, and performance tracking. """ import pytest import asyncio import tempfile import shutil from datetime import datetime, timedelta from pathlib import Path from unittest.mock import Mock, patch, AsyncMock from katamari_mcp.acp.adaptive_learning import ( AdaptiveLearningEngine, ExecutionFeedback, HeuristicAdjustment, PerformanceMetrics ) from katamari_mcp.acp.feedback import ( FeedbackCollector, FeedbackSubmission, FeedbackType, FeedbackSource, AutomatedMetrics ) from katamari_mcp.acp.performance_tracker import ( PerformanceTracker, ExecutionMetrics, CapabilityPerformance ) from katamari_mcp.acp.data_models import ( FeedbackEvent, LearningRecord, AdaptationProposal, validate_feedback_event, validate_learning_record, validate_adaptation_proposal ) from katamari_mcp.acp.heuristics import HeuristicTags, HeuristicEngine @pytest.fixture def temp_workspace(): """Create temporary workspace for testing.""" temp_dir = tempfile.mkdtemp() yield temp_dir shutil.rmtree(temp_dir) @pytest.fixture def mock_config(temp_workspace): """Mock configuration for testing.""" config = Mock() config.workspace_root = temp_workspace return config @pytest.fixture def learning_engine(mock_config): """Create adaptive learning engine for testing.""" return AdaptiveLearningEngine(mock_config) @pytest.fixture def feedback_collector(mock_config, learning_engine): """Create feedback collector for testing.""" return FeedbackCollector(mock_config, learning_engine) @pytest.fixture def performance_tracker(mock_config): """Create performance tracker for testing.""" return PerformanceTracker(mock_config) class TestAdaptiveLearningEngine: """Test adaptive learning engine functionality.""" @pytest.mark.asyncio async def test_collect_feedback(self, learning_engine): """Test feedback collection.""" feedback = ExecutionFeedback( capability_id="test_capability", execution_id="exec_123", timestamp=datetime.now(), success=True, execution_time=1.5, heuristic_profile=HeuristicProfile() ) # Should not raise exception await learning_engine.collect_feedback(feedback) # Verify feedback was stored metrics = await learning_engine.get_capability_metrics("test_capability") assert metrics is not None assert metrics.total_executions == 1 assert metrics.success_rate == 1.0 @pytest.mark.asyncio async def test_heuristic_accuracy_analysis(self, learning_engine): """Test heuristic accuracy analysis.""" # Create test feedback with known patterns feedback_list = [ ExecutionFeedback( capability_id="test_cap", execution_id=f"exec_{i}", timestamp=datetime.now(), success=i % 2 == 0, # Alternate success/failure execution_time=1.0, heuristic_profile=HeuristicProfile(risk=1 if i % 2 == 0 else 9) ) for i in range(10) ] # Add feedback to engine for feedback in feedback_list: await learning_engine.collect_feedback(feedback) # Analyze accuracy accuracy_scores = await learning_engine.analyze_heuristic_accuracy() # Should have accuracy data for test_cap assert "test_cap" in accuracy_scores assert 0 <= accuracy_scores["test_cap"] <= 1 @pytest.mark.asyncio async def test_recommend_heuristic_adjustments(self, learning_engine): """Test heuristic adjustment recommendations.""" # Create feedback with failure patterns for i in range(10): feedback = ExecutionFeedback( capability_id="failing_cap", execution_id=f"exec_{i}", timestamp=datetime.now(), success=False, # All failures execution_time=5.0, error_type="TimeoutError", heuristic_profile=HeuristicProfile(risk=8, complexity=7) ) await learning_engine.collect_feedback(feedback) # Get recommendations adjustments = await learning_engine.recommend_heuristic_adjustments() # Should recommend adjustments for failing capability assert len(adjustments) > 0 assert any(adj.capability_id == "failing_cap" for adj in adjustments) @pytest.mark.asyncio async def test_apply_heuristic_adjustment(self, learning_engine): """Test applying heuristic adjustments.""" adjustment = HeuristicAdjustment( capability_id="test_cap", timestamp=datetime.now(), tag=HeuristicTag.risk, old_value=5, new_value=3, reason="Test adjustment", confidence=0.8, based_on_executions=10 ) # Should apply successfully result = await learning_engine.apply_heuristic_adjustment(adjustment) assert result is True @pytest.mark.asyncio async def test_learning_summary(self, learning_engine): """Test learning summary generation.""" # Add some test data feedback = ExecutionFeedback( capability_id="test_cap", execution_id="exec_123", timestamp=datetime.now(), success=True, execution_time=1.0 ) await learning_engine.collect_feedback(feedback) # Get summary summary = await learning_engine.get_learning_summary() # Should contain expected fields assert "total_feedback_collected" in summary assert "capabilities_tracked" in summary assert "total_adjustments_made" in summary assert summary["total_feedback_collected"] == 1 class TestFeedbackCollector: """Test feedback collection system.""" @pytest.mark.asyncio async def test_submit_feedback(self, feedback_collector): """Test feedback submission.""" submission = FeedbackSubmission( capability_id="test_cap", execution_id="exec_123", feedback_type=FeedbackType.USER_SATISFACTION, source=FeedbackSource.DIRECT_USER, rating=5, comment="Great work!" ) # Should submit successfully result = await feedback_collector.submit_feedback(submission) assert result is True @pytest.mark.asyncio async def test_collect_execution_feedback(self, feedback_collector): """Test execution feedback collection.""" result = await feedback_collector.collect_execution_feedback( capability_id="test_cap", execution_id="exec_123", success=True, execution_time=2.5, automated_metrics=AutomatedMetrics( execution_time=2.5, memory_usage=100, cpu_usage=50.0 ) ) assert result is True @pytest.mark.asyncio async def test_collect_user_satisfaction(self, feedback_collector): """Test user satisfaction collection.""" result = await feedback_collector.collect_user_satisfaction( capability_id="test_cap", execution_id="exec_123", rating=4, comment="Good but could be better" ) assert result is True @pytest.mark.asyncio async def test_feedback_validation(self, feedback_collector): """Test feedback validation.""" # Invalid rating invalid_submission = FeedbackSubmission( capability_id="test_cap", feedback_type=FeedbackType.USER_SATISFACTION, source=FeedbackSource.DIRECT_USER, rating=6 # Invalid (should be 1-5) ) result = await feedback_collector.submit_feedback(invalid_submission) assert result is False @pytest.mark.asyncio async def test_feedback_summary(self, feedback_collector): """Test feedback summary generation.""" # Add some test feedback await feedback_collector.collect_user_satisfaction("test_cap", "exec_1", 5) await feedback_collector.collect_user_satisfaction("test_cap", "exec_2", 3) summary = await feedback_collector.get_feedback_summary() assert "total_submissions" in summary assert summary["total_submissions"] >= 2 class TestPerformanceTracker: """Test performance tracking system.""" @pytest.mark.asyncio async def test_execution_tracking(self, performance_tracker): """Test execution performance tracking.""" capability_id = "test_cap" execution_id = "exec_123" # Track execution async with performance_tracker.track_execution(capability_id, execution_id) as metrics: # Simulate some work await asyncio.sleep(0.1) metrics.success = True # Verify metrics were recorded performance = await performance_tracker.get_capability_performance(capability_id) assert performance is not None assert performance.total_executions == 1 assert performance.successful_executions == 1 @pytest.mark.asyncio async def test_performance_summary(self, performance_tracker): """Test performance summary generation.""" # Add some test executions async with performance_tracker.track_execution("test_cap", "exec_1") as metrics: metrics.success = True async with performance_tracker.track_execution("test_cap", "exec_2") as metrics: metrics.success = False summary = await performance_tracker.get_performance_summary() assert "total_executions" in summary assert "success_rate" in summary assert summary["total_executions"] == 2 assert summary["success_rate"] == 0.5 @pytest.mark.asyncio async def test_performance_trends(self, performance_tracker): """Test performance trend analysis.""" capability_id = "test_cap" # Add executions over time for i in range(5): async with performance_tracker.track_execution(capability_id, f"exec_{i}") as metrics: metrics.success = i % 2 == 0 # Alternate success/failure trends = await performance_tracker.get_performance_trends(capability_id) assert "trend_data" in trends assert "current_health_score" in trends assert len(trends["trend_data"]) > 0 @pytest.mark.asyncio async def test_custom_metrics(self, performance_tracker): """Test custom metric collection.""" capability_id = "test_cap" execution_id = "exec_123" async with performance_tracker.track_execution(capability_id, execution_id) as metrics: # Add custom metrics await performance_tracker.add_custom_metric(execution_id, "custom_counter", 42) await performance_tracker.add_custom_metric(execution_id, "custom_flag", True) metrics.success = True # Verify custom metrics were stored # (This would require accessing the stored execution metrics) pass class TestDataModels: """Test data model validation and serialization.""" def test_feedback_event_validation(self): """Test feedback event validation.""" # Valid event valid_event = FeedbackEvent( capability_id="test_cap", success=True, execution_time=1.0 ) issues = validate_feedback_event(valid_event) assert len(issues) == 0 # Invalid event (missing capability_id) invalid_event = FeedbackEvent( capability_id="", success=True, execution_time=1.0 ) issues = validate_feedback_event(invalid_event) assert len(issues) > 0 assert any("capability_id" in issue for issue in issues) def test_learning_record_validation(self): """Test learning record validation.""" # Valid record valid_record = LearningRecord( trigger_event_id="event_123", capability_id="test_cap", target_component="heuristic_engine", target_parameter="risk_weight", confidence=0.8, based_on_samples=10 ) issues = validate_learning_record(valid_record) assert len(issues) == 0 # Invalid record (invalid confidence) invalid_record = LearningRecord( trigger_event_id="event_123", capability_id="test_cap", target_component="heuristic_engine", target_parameter="risk_weight", confidence=1.5, # Invalid (> 1.0) based_on_samples=10 ) issues = validate_learning_record(invalid_record) assert len(issues) > 0 assert any("confidence" in issue for issue in issues) def test_adaptation_proposal_validation(self): """Test adaptation proposal validation.""" # Valid proposal valid_proposal = AdaptationProposal( target_capability="test_cap", target_component="heuristic_engine", target_parameter="risk_weight", current_value=5, proposed_value=3, reasoning="Test adjustment", confidence_score=0.8, impact_confidence=0.7 ) issues = validate_adaptation_proposal(valid_proposal) assert len(issues) == 0 # Invalid proposal (missing reasoning) invalid_proposal = AdaptationProposal( target_capability="test_cap", target_component="heuristic_engine", target_parameter="risk_weight", current_value=5, proposed_value=3, reasoning="", # Missing confidence_score=0.8, impact_confidence=0.7 ) issues = validate_adaptation_proposal(invalid_proposal) assert len(issues) > 0 assert any("reasoning" in issue for issue in issues) class TestIntegration: """Integration tests for adaptive learning components.""" @pytest.mark.asyncio async def test_end_to_end_feedback_loop(self, learning_engine, feedback_collector, performance_tracker): """Test complete feedback loop from execution to learning.""" capability_id = "test_cap" execution_id = "exec_123" # 1. Track execution performance async with performance_tracker.track_execution(capability_id, execution_id) as metrics: await asyncio.sleep(0.1) # Simulate work metrics.success = True metrics.execution_time = 0.1 # 2. Collect execution feedback await feedback_collector.collect_execution_feedback( capability_id=capability_id, execution_id=execution_id, success=True, execution_time=0.1 ) # 3. Collect user satisfaction await feedback_collector.collect_user_satisfaction( capability_id=capability_id, execution_id=execution_id, rating=5, comment="Excellent!" ) # 4. Check learning engine processed feedback capability_metrics = await learning_engine.get_capability_metrics(capability_id) assert capability_metrics is not None assert capability_metrics.total_executions >= 1 # 5. Check performance tracker recorded metrics performance = await performance_tracker.get_capability_performance(capability_id) assert performance is not None assert performance.total_executions >= 1 # 6. Check feedback collector has data summary = await feedback_collector.get_feedback_summary(capability_id=capability_id) assert summary["total_submissions"] >= 2 # execution + satisfaction @pytest.mark.asyncio async def test_adaptive_learning_cycle(self, learning_engine): """Test adaptive learning cycle with multiple executions.""" capability_id = "adaptive_cap" # Phase 1: Initial executions with failures for i in range(5): feedback = ExecutionFeedback( capability_id=capability_id, execution_id=f"exec_{i}", timestamp=datetime.now(), success=False, # All failures execution_time=10.0, error_type="TimeoutError", heuristic_profile=HeuristicProfile(risk=8, complexity=9) ) await learning_engine.collect_feedback(feedback) # Phase 2: Get adjustment recommendations adjustments = await learning_engine.recommend_heuristic_adjustments() assert len(adjustments) > 0 # Phase 3: Apply adjustment if adjustments: await learning_engine.apply_heuristic_adjustment(adjustments[0]) # Phase 4: Subsequent executions with better performance for i in range(5, 10): feedback = ExecutionFeedback( capability_id=capability_id, execution_id=f"exec_{i}", timestamp=datetime.now(), success=True, # Now successful execution_time=2.0, heuristic_profile=HeuristicProfile(risk=3, complexity=4) # Adjusted ) await learning_engine.collect_feedback(feedback) # Phase 5: Check learning progress summary = await learning_engine.get_learning_summary() assert summary["total_feedback_collected"] == 10 assert summary["capabilities_tracked"] == 1 if __name__ == "__main__": pytest.main([__file__])