Coverage for tests / unit / ai / test_retry.py: 99%

1"""Tests for AI retry decorator.""" 

2 

3from __future__ import annotations 

4 

5from unittest.mock import patch 

6 

7import pytest 

8from assertpy import assert_that 

9 

10from lintro.ai.exceptions import ( 

11 AIAuthenticationError, 

12 AIProviderError, 

13 AIRateLimitError, 

14) 

15from lintro.ai.retry import with_retry 

16 

17 

18def test_retry_succeeds_on_first_attempt(): 

19 """Verify decorated function returns immediately on success without retrying.""" 

20 call_count = 0 

21 

22 @with_retry(max_retries=3) 

23 def fn(): 

24 nonlocal call_count 

25 call_count += 1 

26 return "ok" 

27 

28 result = fn() 

29 assert_that(result).is_equal_to("ok") 

30 assert_that(call_count).is_equal_to(1) 

31 

32 

33@patch("lintro.ai.retry.time.sleep") 

34def test_retry_retries_on_provider_error(mock_sleep): 

35 """Verify AIProviderError triggers retries until success.""" 

36 call_count = 0 

37 

38 @with_retry(max_retries=3, base_delay=1.0) 

39 def fn(): 

40 nonlocal call_count 

41 call_count += 1 

42 if call_count < 3: 

43 raise AIProviderError("server error") 

44 return "ok" 

45 

46 result = fn() 

47 assert_that(result).is_equal_to("ok") 

48 assert_that(call_count).is_equal_to(3) 

49 assert_that(mock_sleep.call_count).is_equal_to(2) 

50 

51 

52@patch("lintro.ai.retry.time.sleep") 

53def test_retry_retries_on_rate_limit_error(mock_sleep): 

54 """Verify AIRateLimitError triggers retries until success.""" 

55 call_count = 0 

56 

57 @with_retry(max_retries=2, base_delay=1.0) 

58 def fn(): 

59 nonlocal call_count 

60 call_count += 1 

61 if call_count < 2: 

62 raise AIRateLimitError("rate limited") 

63 return "ok" 

64 

65 result = fn() 

66 assert_that(result).is_equal_to("ok") 

67 assert_that(call_count).is_equal_to(2) 

68 mock_sleep.assert_called_once() 

69 

70 

71def test_retry_does_not_retry_on_authentication_error(): 

72 """Verify AIAuthenticationError is raised immediately without retrying.""" 

73 call_count = 0 

74 

75 @with_retry(max_retries=3) 

76 def fn(): 

77 nonlocal call_count 

78 call_count += 1 

79 raise AIAuthenticationError("bad key") 

80 

81 with pytest.raises(AIAuthenticationError): 

82 fn() 

83 assert_that(call_count).is_equal_to(1) 

84 

85 

86@patch("lintro.ai.retry.time.sleep") 

87def test_retry_raises_after_max_retries_exhausted(mock_sleep): 

88 """Verify the original error is raised after all retry attempts are exhausted.""" 

89 

90 @with_retry(max_retries=2, base_delay=0.1) 

91 def fn(): 

92 raise AIProviderError("always fails") 

93 

94 with pytest.raises(AIProviderError, match="always fails"): 

95 fn() 

96 assert_that(mock_sleep.call_count).is_equal_to(2) 

97 

98 

99@patch("lintro.ai.retry.random.uniform", return_value=1.0) 

100@patch("lintro.ai.retry.time.sleep") 

101def test_retry_exponential_backoff_delays(mock_sleep, _mock_uniform): 

102 """Verify retry delays follow exponential backoff progression.""" 

103 call_count = 0 

104 

105 @with_retry(max_retries=3, base_delay=1.0, backoff_factor=2.0) 

106 def fn(): 

107 nonlocal call_count 

108 call_count += 1 

109 if call_count <= 3: 

110 raise AIProviderError("fail") 

111 return "ok" 

112 

113 result = fn() 

114 assert_that(result).is_equal_to("ok") 

115 

116 delays = [call.args[0] for call in mock_sleep.call_args_list] 

117 assert_that(delays).is_equal_to([1.0, 2.0, 4.0]) 

118 

119 

120@patch("lintro.ai.retry.random.uniform", return_value=1.0) 

121@patch("lintro.ai.retry.time.sleep") 

122def test_retry_max_delay_cap(mock_sleep, _mock_uniform): 

123 """Verify retry delays are capped at the configured max_delay value.""" 

124 call_count = 0 

125 

126 @with_retry( 

127 max_retries=5, 

128 base_delay=10.0, 

129 backoff_factor=3.0, 

130 max_delay=25.0, 

131 ) 

132 def fn(): 

133 nonlocal call_count 

134 call_count += 1 

135 if call_count <= 5: 

136 raise AIProviderError("fail") 

137 return "ok" 

138 

139 fn() 

140 delays = [call.args[0] for call in mock_sleep.call_args_list] 

141 assert_that(delays).is_length(5) 

142 assert_that(delays).is_equal_to([10.0, 25.0, 25.0, 25.0, 25.0]) 

143 

144 

145def test_retry_does_not_retry_non_ai_exceptions(): 

146 """Verify non-AI exceptions propagate immediately without retrying.""" 

147 call_count = 0 

148 

149 @with_retry(max_retries=3) 

150 def fn(): 

151 nonlocal call_count 

152 call_count += 1 

153 raise ValueError("not an AI error") 

154 

155 with pytest.raises(ValueError, match="not an AI error"): 

156 fn() 

157 assert_that(call_count).is_equal_to(1) 

158 

159 

160def test_retry_preserves_function_metadata(): 

161 """Verify the retry decorator preserves the wrapped function name and docstring.""" 

162 

163 @with_retry(max_retries=1) 

164 def my_function(): 

165 """My docstring.""" 

166 return 42 

167 

168 assert_that(my_function.__name__).is_equal_to("my_function") 

169 assert_that(my_function.__doc__).is_equal_to("My docstring.")