"""Unit tests for profiling utilities.""" import time from unittest.mock import Mock import pytest from abacus_common_logic.utils.profiling import profile class TestProfile: """Tests for profile decorator.""" def test_profile_without_args(self): """Test @profile decorator without arguments.""" mock_logger = Mock() @profile(logger=mock_logger) def dummy_task(): return 'result' result = dummy_task() assert result == 'result' assert mock_logger.info.called log_msg = mock_logger.info.call_args[0][0] assert 'Stats: Time:' in log_msg assert 'Memory:' in log_msg def test_profile_with_logger_arg(self): """Test @profile(logger=...) decorator with custom logger.""" mock_logger = Mock() @profile(logger=mock_logger) def task_with_logger(): return 42 result = task_with_logger() assert result == 42 assert mock_logger.info.call_count == 1 def test_profile_logs_execution_time(self): """Test that profile logs execution time.""" mock_logger = Mock() @profile(logger=mock_logger) def slow_task(): time.sleep(0.01) return 'done' result = slow_task() assert result == 'done' log_msg = mock_logger.info.call_args[0][0] assert 'Time:' in log_msg assert 's' in log_msg # seconds unit def test_profile_logs_memory_delta(self): """Test that profile logs memory delta.""" mock_logger = Mock() @profile(logger=mock_logger) def memory_task(): # Allocate some memory _ = [0] * 10000 return 'done' result = memory_task() assert result == 'done' log_msg = mock_logger.info.call_args[0][0] assert 'Memory:' in log_msg assert 'Delta:' in log_msg def test_profile_preserves_function_name(self): """Test that decorator preserves function metadata.""" @profile def named_function(): """Do something.""" pass assert named_function.__name__ == 'named_function' assert named_function.__doc__ == 'Do something.' def test_profile_with_args_and_kwargs(self): """Test profiling function with arguments.""" mock_logger = Mock() @profile(logger=mock_logger) def task_with_params(x, y, z=10): return x + y + z result = task_with_params(1, 2, z=3) assert result == 6 assert mock_logger.info.called def test_profile_with_exception(self): """Test that profile doesn't suppress exceptions.""" mock_logger = Mock() @profile(logger=mock_logger) def failing_task(): raise ValueError('Task failed') with pytest.raises(ValueError, match='Task failed'): failing_task() # Logger should not be called if exception occurs before completion # (or it might be called depending on when the exception occurs) def test_profile_multiple_calls(self): """Test profiling same function multiple times.""" mock_logger = Mock() @profile(logger=mock_logger) def repeated_task(): return 'done' repeated_task() repeated_task() repeated_task() assert mock_logger.info.call_count == 3 def test_profile_returns_function_result(self): """Test that decorator returns the function's result.""" mock_logger = Mock() @profile(logger=mock_logger) def task_with_return(): return {'status': 'success', 'count': 42} result = task_with_return() assert result == {'status': 'success', 'count': 42} def test_profile_without_logger_uses_default(self): """Test that profile uses default logger when not provided.""" @profile def task_default_logger(): return 'result' # Should not raise an error result = task_default_logger() assert result == 'result' def test_profile_decorator_callable_multiple_times(self): """Test that decorated function can be called multiple times.""" mock_logger = Mock() call_count = 0 @profile(logger=mock_logger) def stateful_task(): nonlocal call_count call_count += 1 return call_count assert stateful_task() == 1 assert stateful_task() == 2 assert stateful_task() == 3 assert mock_logger.info.call_count == 3 def test_profile_with_generator_function(self): """Test profiling a generator function.""" mock_logger = Mock() @profile(logger=mock_logger) def generator_task(): yield 1 yield 2 yield 3 gen = generator_task() # Creating the generator calls the decorator assert mock_logger.info.call_count == 1 # Consuming the generator result = list(gen) assert result == [1, 2, 3]