Coverage for tests / unit / utils / test_fix_retry.py: 100%

1"""Tests for fix convergence retry logic.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from typing import Any 

7 

8from assertpy import assert_that 

9 

10from lintro.models.core.tool_result import ToolResult 

11from lintro.utils.tool_executor import _run_fix_with_retry 

12 

13 

14@dataclass 

15class _MockToolDefinition: 

16 """Minimal mock tool definition. 

17 

18 Attributes: 

19 name: Name of the tool. 

20 """ 

21 

22 name: str = "mock_tool" 

23 

24 

25@dataclass 

26class _ConvergingMockTool: 

27 """Mock tool that converges after a given number of fix passes. 

28 

29 Attributes: 

30 definition: Tool definition mock. 

31 converge_on_pass: Pass number (1-based) on which remaining goes to 0. 

32 """ 

33 

34 definition: _MockToolDefinition = field(default_factory=_MockToolDefinition) 

35 converge_on_pass: int = 2 

36 _call_count: int = field(default=0, init=False) 

37 

38 def fix( 

39 self, 

40 paths: list[str], 

41 options: dict[str, Any] | None = None, 

42 ) -> ToolResult: 

43 """Mock fix that converges after converge_on_pass calls. 

44 

45 Args: 

46 paths: Paths to fix. 

47 options: Fix options. 

48 

49 Returns: 

50 ToolResult with remaining issues depending on call count. 

51 """ 

52 self._call_count += 1 

53 if self._call_count >= self.converge_on_pass: 

54 return ToolResult( 

55 name=self.definition.name, 

56 success=True, 

57 output="all fixed", 

58 issues_count=0, 

59 initial_issues_count=3, 

60 fixed_issues_count=3, 

61 remaining_issues_count=0, 

62 ) 

63 return ToolResult( 

64 name=self.definition.name, 

65 success=False, 

66 output="still has issues", 

67 issues_count=1, 

68 initial_issues_count=3, 

69 fixed_issues_count=2, 

70 remaining_issues_count=1, 

71 ) 

72 

73 

74@dataclass 

75class _NeverConvergingMockTool: 

76 """Mock tool that never converges. 

77 

78 Attributes: 

79 definition: Tool definition mock. 

80 """ 

81 

82 definition: _MockToolDefinition = field(default_factory=_MockToolDefinition) 

83 _call_count: int = field(default=0, init=False) 

84 

85 def fix( 

86 self, 

87 paths: list[str], 

88 options: dict[str, Any] | None = None, 

89 ) -> ToolResult: 

90 """Mock fix that always reports remaining issues. 

91 

92 Args: 

93 paths: Paths to fix. 

94 options: Fix options. 

95 

96 Returns: 

97 ToolResult with remaining issues. 

98 """ 

99 self._call_count += 1 

100 return ToolResult( 

101 name=self.definition.name, 

102 success=False, 

103 output="unfixable", 

104 issues_count=2, 

105 initial_issues_count=5, 

106 fixed_issues_count=3, 

107 remaining_issues_count=2, 

108 ) 

109 

110 

111def test_fix_converges_on_second_attempt() -> None: 

112 """Should succeed when tool converges on the second pass.""" 

113 tool = _ConvergingMockTool(converge_on_pass=2) 

114 

115 result = _run_fix_with_retry( 

116 tool=tool, # type: ignore[arg-type] 

117 paths=["."], 

118 options={}, 

119 max_retries=3, 

120 ) 

121 

122 assert_that(result.success).is_true() 

123 assert_that(result.remaining_issues_count).is_equal_to(0) 

124 assert_that(result.initial_issues_count).is_equal_to(3) 

125 assert_that(result.fixed_issues_count).is_equal_to(3) 

126 assert_that(tool._call_count).is_equal_to(2) 

127 

128 

129def test_fix_reports_unfixable_after_max_retries() -> None: 

130 """Should report remaining issues when max retries are exhausted.""" 

131 tool = _NeverConvergingMockTool() 

132 

133 result = _run_fix_with_retry( 

134 tool=tool, # type: ignore[arg-type] 

135 paths=["."], 

136 options={}, 

137 max_retries=3, 

138 ) 

139 

140 assert_that(result.success).is_false() 

141 assert_that(result.remaining_issues_count).is_equal_to(2) 

142 assert_that(tool._call_count).is_equal_to(3) 

143 

144 

145def test_fix_no_retry_when_first_pass_succeeds() -> None: 

146 """Should not retry when the first pass succeeds.""" 

147 tool = _ConvergingMockTool(converge_on_pass=1) 

148 

149 result = _run_fix_with_retry( 

150 tool=tool, # type: ignore[arg-type] 

151 paths=["."], 

152 options={}, 

153 max_retries=3, 

154 ) 

155 

156 assert_that(result.success).is_true() 

157 assert_that(result.remaining_issues_count).is_equal_to(0) 

158 assert_that(tool._call_count).is_equal_to(1) 

159 

160 

161def test_fix_retry_merges_results_correctly() -> None: 

162 """Should keep initial count from first pass and remaining from last.""" 

163 tool = _ConvergingMockTool(converge_on_pass=3) 

164 

165 result = _run_fix_with_retry( 

166 tool=tool, # type: ignore[arg-type] 

167 paths=["."], 

168 options={}, 

169 max_retries=5, 

170 ) 

171 

172 # initial_issues_count should be from the first pass (3) 

173 assert_that(result.initial_issues_count).is_equal_to(3) 

174 # remaining from last pass (0 since it converged on pass 3) 

175 assert_that(result.remaining_issues_count).is_equal_to(0) 

176 # fixed = initial - remaining = 3 - 0 = 3 

177 assert_that(result.fixed_issues_count).is_equal_to(3) 

178 assert_that(tool._call_count).is_equal_to(3)