MaverickMCP

""" Tests for agent_errors.py - Smart error handling with automatic fixes. This test suite achieves 100% coverage by testing: 1. Error pattern matching for all predefined patterns 2. Sync and async decorator functionality 3. Context manager behavior 4. Edge cases and error scenarios """ import asyncio from unittest.mock import patch import pandas as pd import pytest from maverick_mcp.utils.agent_errors import ( AgentErrorContext, agent_friendly_errors, find_error_fix, get_error_context, ) class TestFindErrorFix: """Test error pattern matching functionality.""" def test_dataframe_column_error_matching(self): """Test DataFrame column case sensitivity error detection.""" error_msg = "KeyError: 'close'" fix_info = find_error_fix(error_msg) assert fix_info is not None assert "Use 'Close' with capital C" in fix_info["fix"] assert "df['Close'] not df['close']" in fix_info["example"] def test_authentication_error_matching(self): """Test authentication error detection.""" error_msg = "401 Unauthorized" fix_info = find_error_fix(error_msg) assert fix_info is not None assert "AUTH_ENABLED=false" in fix_info["fix"] def test_redis_connection_error_matching(self): """Test Redis connection error detection.""" error_msg = "Redis connection refused" fix_info = find_error_fix(error_msg) assert fix_info is not None assert "brew services start redis" in fix_info["fix"] def test_no_match_returns_none(self): """Test that unmatched errors return None.""" error_msg = "Some random error that doesn't match any pattern" fix_info = find_error_fix(error_msg) assert fix_info is None def test_all_error_patterns(self): """Test that all ERROR_FIXES patterns match correctly.""" test_cases = [ ("KeyError: 'close'", "Use 'Close' with capital C"), ("KeyError: 'open'", "Use 'Open' with capital O"), ("KeyError: 'high'", "Use 'High' with capital H"), ("KeyError: 'low'", "Use 'Low' with capital L"), ("KeyError: 'volume'", "Use 'Volume' with capital V"), ("401 Unauthorized", "AUTH_ENABLED=false"), ("Redis connection refused", "brew services start redis"), ("psycopg2 could not connect to server", "Use SQLite for development"), ( "ModuleNotFoundError: No module named 'maverick'", "Install dependencies: uv sync", ), ("ImportError: cannot import name 'ta_lib'", "Install TA-Lib"), ( "TypeError: 'NoneType' object has no attribute 'foo'", "Check if the object exists", ), ("ValueError: not enough values to unpack", "Check the return value"), ("RuntimeError: no running event loop", "Use asyncio.run()"), ("FileNotFoundError", "Check the file path"), ("Address already in use on port 8000", "Stop the existing server"), ] for error_msg, expected_fix_part in test_cases: fix_info = find_error_fix(error_msg) assert fix_info is not None, f"No fix found for: {error_msg}" assert expected_fix_part in fix_info["fix"], ( f"Fix mismatch for: {error_msg}" ) class TestAgentFriendlyErrors: """Test agent_friendly_errors decorator functionality.""" def test_sync_function_with_error(self): """Test decorator on synchronous function that raises an error.""" @agent_friendly_errors def failing_function(): # Use an error message that will be matched raise KeyError("KeyError: 'close'") with pytest.raises(KeyError) as exc_info: failing_function() # Check that error message was enhanced error_msg = ( str(exc_info.value.args[0]) if exc_info.value.args else str(exc_info.value) ) assert "Fix:" in error_msg assert "Use 'Close' with capital C" in error_msg def test_sync_function_success(self): """Test decorator on synchronous function that succeeds.""" @agent_friendly_errors def successful_function(): return "success" result = successful_function() assert result == "success" @pytest.mark.asyncio async def test_async_function_with_error(self): """Test decorator on asynchronous function that raises an error.""" @agent_friendly_errors async def failing_async_function(): raise ConnectionRefusedError("Redis connection refused") with pytest.raises(ConnectionRefusedError) as exc_info: await failing_async_function() error_msg = str(exc_info.value) assert "Fix:" in error_msg assert "brew services start redis" in error_msg @pytest.mark.asyncio async def test_async_function_success(self): """Test decorator on asynchronous function that succeeds.""" @agent_friendly_errors async def successful_async_function(): return "async success" result = await successful_async_function() assert result == "async success" def test_decorator_with_parameters(self): """Test decorator with custom parameters.""" # Test with provide_fix=True but reraise=False to avoid the bug @agent_friendly_errors(provide_fix=True, log_errors=False, reraise=False) def function_with_params(): raise ValueError("Test error") # With reraise=False, should return error info dict instead of raising result = function_with_params() assert isinstance(result, dict) assert result["error_type"] == "ValueError" assert result["error_message"] == "Test error" # Test a different parameter combination @agent_friendly_errors(log_errors=False) def function_with_logging_off(): return "success" result = function_with_logging_off() assert result == "success" def test_decorator_preserves_function_attributes(self): """Test that decorator preserves function metadata.""" @agent_friendly_errors def documented_function(): """This is a documented function.""" return "result" assert documented_function.__name__ == "documented_function" assert documented_function.__doc__ == "This is a documented function." def test_error_with_no_args(self): """Test handling of exceptions with no args.""" @agent_friendly_errors def error_no_args(): # Create error with no args raise ValueError() with pytest.raises(ValueError) as exc_info: error_no_args() # Should handle gracefully - error will have default string representation # When ValueError has no args, str(e) returns empty string assert str(exc_info.value) == "" def test_error_with_multiple_args(self): """Test handling of exceptions with multiple args.""" @agent_friendly_errors def error_multiple_args(): # Need to match the pattern - use the full error string raise KeyError("KeyError: 'close'", "additional", "args") with pytest.raises(KeyError) as exc_info: error_multiple_args() # First arg should be enhanced, others preserved assert "Fix:" in str(exc_info.value.args[0]) assert exc_info.value.args[1] == "additional" assert exc_info.value.args[2] == "args" @patch("maverick_mcp.utils.agent_errors.logger") def test_logging_behavior(self, mock_logger): """Test that errors are logged when log_errors=True.""" @agent_friendly_errors(log_errors=True) def logged_error(): raise ValueError("Test error") with pytest.raises(ValueError): logged_error() mock_logger.error.assert_called() call_args = mock_logger.error.call_args assert "Error in logged_error" in call_args[0][0] assert "ValueError" in call_args[0][0] assert "Test error" in call_args[0][0] class TestAgentErrorContext: """Test AgentErrorContext context manager.""" def test_context_manager_with_error(self): """Test context manager catching and logging errors with fixes.""" with pytest.raises(KeyError): with AgentErrorContext("dataframe operation"): df = pd.DataFrame({"Close": [100, 101, 102]}) _ = df["close"] # Wrong case # Context manager logs but doesn't modify the exception def test_context_manager_success(self): """Test context manager with successful code.""" with AgentErrorContext("test operation"): result = 1 + 1 assert result == 2 # Should complete without error def test_context_manager_with_custom_operation(self): """Test context manager with custom operation name.""" with pytest.raises(ValueError): with AgentErrorContext("custom operation"): raise ValueError("Test error") def test_nested_context_managers(self): """Test nested context managers.""" with pytest.raises(ConnectionRefusedError): with AgentErrorContext("outer operation"): with AgentErrorContext("inner operation"): raise ConnectionRefusedError("Redis connection refused") @patch("maverick_mcp.utils.agent_errors.logger") def test_context_manager_logging(self, mock_logger): """Test context manager logging behavior when fix is found.""" with pytest.raises(KeyError): with AgentErrorContext("test operation"): # Use error message that will match pattern raise KeyError("KeyError: 'close'") # Should log error and fix mock_logger.error.assert_called_once() mock_logger.info.assert_called_once() error_call = mock_logger.error.call_args[0][0] assert "Error during test operation" in error_call info_call = mock_logger.info.call_args[0][0] assert "Fix:" in info_call class TestGetErrorContext: """Test get_error_context utility function.""" def test_basic_error_context(self): """Test extracting context from basic exception.""" try: raise ValueError("Test error") except ValueError as e: context = get_error_context(e) assert context["error_type"] == "ValueError" assert context["error_message"] == "Test error" assert "traceback" in context assert context["traceback"] is not None def test_error_context_with_value_error(self): """Test extracting context from ValueError.""" try: raise ValueError("Test value error", "extra", "args") except ValueError as e: context = get_error_context(e) assert context["error_type"] == "ValueError" assert context["error_message"] == "('Test value error', 'extra', 'args')" assert "value_error_details" in context assert context["value_error_details"] == ("Test value error", "extra", "args") def test_error_context_with_connection_error(self): """Test extracting context from ConnectionError.""" try: raise ConnectionError("Network failure") except ConnectionError as e: context = get_error_context(e) assert context["error_type"] == "ConnectionError" assert context["error_message"] == "Network failure" assert context["connection_type"] == "network" class TestIntegrationScenarios: """Test real-world integration scenarios.""" @pytest.mark.asyncio async def test_async_dataframe_operation(self): """Test async function with DataFrame operations.""" @agent_friendly_errors async def process_dataframe(): df = pd.DataFrame({"Close": [100, 101, 102]}) await asyncio.sleep(0.01) # Simulate async operation # This will raise KeyError: 'close' which will be caught try: return df["close"] # Wrong case except KeyError: # Re-raise with pattern that will match raise KeyError("KeyError: 'close'") with pytest.raises(KeyError) as exc_info: await process_dataframe() assert "Use 'Close' with capital C" in str(exc_info.value.args[0]) def test_multiple_error_types_in_sequence(self): """Test handling different error types in sequence.""" @agent_friendly_errors def multi_error_function(error_type): if error_type == "auth": raise PermissionError("401 Unauthorized") elif error_type == "redis": raise ConnectionRefusedError("Redis connection refused") elif error_type == "port": raise OSError("Address already in use on port 8000") return "success" # Test auth error with pytest.raises(PermissionError) as exc_info: multi_error_function("auth") assert "AUTH_ENABLED=false" in str(exc_info.value) # Test redis error with pytest.raises(ConnectionRefusedError) as exc_info: multi_error_function("redis") assert "brew services start redis" in str(exc_info.value) # Test port error with pytest.raises(OSError) as exc_info: multi_error_function("port") assert "make stop" in str(exc_info.value) def test_decorator_stacking(self): """Test stacking multiple decorators.""" call_order = [] def other_decorator(func): def wrapper(*args, **kwargs): call_order.append("other_before") result = func(*args, **kwargs) call_order.append("other_after") return result return wrapper @agent_friendly_errors @other_decorator def stacked_function(): call_order.append("function") return "result" result = stacked_function() assert result == "result" assert call_order == ["other_before", "function", "other_after"] def test_class_method_decoration(self): """Test decorating class methods.""" class TestClass: @agent_friendly_errors def instance_method(self): raise KeyError("KeyError: 'close'") @classmethod @agent_friendly_errors def class_method(cls): raise ConnectionRefusedError("Redis connection refused") @staticmethod @agent_friendly_errors def static_method(): raise OSError("Address already in use on port 8000") obj = TestClass() # Test instance method with pytest.raises(KeyError) as exc_info: obj.instance_method() assert "Use 'Close' with capital C" in str(exc_info.value.args[0]) # Test class method with pytest.raises(ConnectionRefusedError) as exc_info: TestClass.class_method() assert "brew services start redis" in str(exc_info.value.args[0]) # Test static method with pytest.raises(OSError) as exc_info: TestClass.static_method() assert "make stop" in str(exc_info.value.args[0]) class TestEdgeCases: """Test edge cases and boundary conditions.""" def test_very_long_error_message(self): """Test handling of very long error messages.""" long_message = "A" * 10000 @agent_friendly_errors def long_error(): raise ValueError(long_message) with pytest.raises(ValueError) as exc_info: long_error() # Should handle without truncation issues # The error message is the first argument error_str = ( str(exc_info.value.args[0]) if exc_info.value.args else str(exc_info.value) ) assert len(error_str) >= 10000 def test_unicode_error_messages(self): """Test handling of unicode in error messages.""" @agent_friendly_errors def unicode_error(): raise ValueError("Error with emoji 🐛 and unicode ñ") with pytest.raises(ValueError) as exc_info: unicode_error() # Should preserve unicode characters assert "🐛" in str(exc_info.value) assert "ñ" in str(exc_info.value) def test_circular_reference_in_exception(self): """Test handling of circular references in exception objects.""" @agent_friendly_errors def circular_error(): e1 = ValueError("Error 1") e2 = ValueError("Error 2") e1.__cause__ = e2 e2.__cause__ = e1 # Circular reference raise e1 # Should handle without infinite recursion with pytest.raises(ValueError): circular_error() def test_concurrent_decorator_calls(self): """Test thread safety of decorator.""" import threading results = [] errors = [] @agent_friendly_errors def concurrent_function(should_fail): if should_fail: raise KeyError("KeyError: 'close'") return "success" def thread_function(should_fail): try: result = concurrent_function(should_fail) results.append(result) except Exception as e: # Get the enhanced error message from args error_msg = str(e.args[0]) if e.args else str(e) errors.append(error_msg) threads = [] for i in range(10): t = threading.Thread(target=thread_function, args=(i % 2 == 0,)) threads.append(t) t.start() for t in threads: t.join() assert len(results) == 5 assert len(errors) == 5 assert all("Fix:" in error for error in errors)