Coverage for lintro / tools / implementations / ruff / fix.py: 99%
112 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-04-03 18:53 +0000
1"""Ruff fix execution logic.
3Functions for running ruff fix commands and processing results.
4"""
6import subprocess # nosec B404 - subprocess used safely to execute ruff commands with controlled input
7from collections.abc import Generator
8from contextlib import contextmanager
9from typing import TYPE_CHECKING
11from loguru import logger
13from lintro.parsers.ruff.ruff_parser import (
14 parse_ruff_format_check_output,
15 parse_ruff_output,
16)
17from lintro.tools.core.timeout_utils import (
18 create_timeout_result,
19 get_timeout_value,
20)
21from lintro.utils.path_filtering import walk_files_with_excludes
23if TYPE_CHECKING:
24 from lintro.models.core.tool_result import ToolResult
25 from lintro.tools.definitions.ruff import RuffPlugin
27# Constants from tool_ruff.py
28RUFF_DEFAULT_TIMEOUT: int = 30
29DEFAULT_REMAINING_ISSUES_DISPLAY: int = 5
32@contextmanager
33def _temporary_option(
34 tool: "RuffPlugin",
35 option_key: str,
36 option_value: object,
37) -> Generator[None]:
38 """Context manager for temporarily setting a tool option.
40 Safely mutates tool.options for the duration of the context, ensuring
41 the original value is always restored even if an exception occurs.
43 Args:
44 tool: RuffTool instance whose options will be temporarily modified.
45 option_key: Key of the option to temporarily set.
46 option_value: Value to temporarily set for the option.
48 Yields:
49 None: The context manager yields control after setting the option.
51 Example:
52 >>> with _temporary_option(tool, "unsafe_fixes", True):
53 ... # tool.options["unsafe_fixes"] is now True
54 ... build_command(tool)
55 >>> # tool.options["unsafe_fixes"] is restored to original value
56 """
57 # Check if key existed before (distinguishes None value from missing key)
58 key_existed = option_key in tool.options
59 original_value = tool.options.get(option_key)
60 try:
61 tool.options[option_key] = option_value
62 yield
63 finally:
64 # Always restore the original state, even if an exception occurs
65 if key_existed:
66 tool.options[option_key] = original_value
67 elif option_key in tool.options:
68 # Remove the key if it wasn't originally present
69 del tool.options[option_key]
72def execute_ruff_fix(
73 tool: "RuffPlugin",
74 paths: list[str],
75) -> "ToolResult":
76 """Execute ruff fix command and process results.
78 Args:
79 tool: RuffTool instance
80 paths: list[str]: List of file or directory paths to fix.
82 Returns:
83 ToolResult: ToolResult instance.
84 """
85 from lintro.models.core.tool_result import ToolResult
86 from lintro.tools.implementations.ruff.commands import (
87 build_ruff_check_command,
88 build_ruff_format_command,
89 )
91 # Check version requirements
92 version_result = tool._verify_tool_version()
93 if version_result is not None:
94 return version_result
96 tool._validate_paths(paths=paths)
97 if not paths:
98 return ToolResult(
99 name=tool.definition.name,
100 success=True,
101 output="No files to fix.",
102 issues_count=0,
103 )
105 # Use shared utility for file discovery
106 python_files: list[str] = walk_files_with_excludes(
107 paths=paths,
108 file_patterns=tool.definition.file_patterns,
109 exclude_patterns=tool.exclude_patterns,
110 include_venv=tool.include_venv,
111 incremental=bool(tool.options.get("incremental", False)),
112 tool_name="ruff",
113 )
115 if not python_files:
116 return ToolResult(
117 name=tool.definition.name,
118 success=True,
119 output="No Python files found to fix.",
120 issues_count=0,
121 )
123 timeout: int = get_timeout_value(tool, RUFF_DEFAULT_TIMEOUT)
124 overall_success: bool = True
126 # Track unsafe fixes for internal decisioning; do not emit as user-facing noise
127 unsafe_fixes_enabled: bool = bool(tool.options.get("unsafe_fixes", False))
129 # First, count issues before fixing
130 cmd_check: list[str] = build_ruff_check_command(
131 tool=tool,
132 files=python_files,
133 fix=False,
134 )
135 success_check: bool = False
136 output_check: str = ""
137 try:
138 success_check, output_check = tool._run_subprocess(
139 cmd=cmd_check,
140 timeout=timeout,
141 )
142 except subprocess.TimeoutExpired:
143 timeout_result = create_timeout_result(
144 tool=tool,
145 timeout=timeout,
146 cmd=cmd_check,
147 )
148 return ToolResult(
149 name=tool.definition.name,
150 success=timeout_result.success,
151 output=timeout_result.output,
152 issues_count=timeout_result.issues_count,
153 issues=timeout_result.issues,
154 initial_issues_count=None,
155 fixed_issues_count=None,
156 remaining_issues_count=None,
157 )
158 initial_issues = parse_ruff_output(output=output_check)
159 initial_count: int = len(initial_issues)
161 # Also check formatting issues before fixing
162 initial_format_count: int = 0
163 format_files: list[str] = []
164 if tool.options.get("format", False):
165 format_cmd_check: list[str] = build_ruff_format_command(
166 tool=tool,
167 files=python_files,
168 check_only=True,
169 )
170 success_format_check: bool = False
171 output_format_check: str = ""
172 try:
173 success_format_check, output_format_check = tool._run_subprocess(
174 cmd=format_cmd_check,
175 timeout=timeout,
176 )
177 except subprocess.TimeoutExpired:
178 timeout_msg = (
179 f"Ruff execution timed out ({timeout}s limit exceeded).\n\n"
180 "This may indicate:\n"
181 " - Large codebase taking too long to process\n"
182 " - Need to increase timeout via --tool-options ruff:timeout=N"
183 )
184 return ToolResult(
185 name=tool.definition.name,
186 success=False,
187 output=timeout_msg,
188 issues_count=1, # Count timeout as execution failure
189 # Include any lint issues found before timeout
190 issues=initial_issues,
191 initial_issues_count=initial_count,
192 fixed_issues_count=0,
193 remaining_issues_count=1,
194 )
195 format_files = parse_ruff_format_check_output(output=output_format_check)
196 initial_format_count = len(format_files)
198 # Track initial totals separately for accurate fixed/remaining math
199 total_initial_count: int = initial_count + initial_format_count
201 # Optionally run ruff check --fix (lint fixes)
202 remaining_issues = []
203 remaining_count = 0
204 success: bool = True # Default to True when lint_fix is disabled
205 if tool.options.get("lint_fix", True):
206 cmd: list[str] = build_ruff_check_command(
207 tool=tool,
208 files=python_files,
209 fix=True,
210 )
211 output: str = ""
212 try:
213 success, output = tool._run_subprocess(cmd=cmd, timeout=timeout)
214 except subprocess.TimeoutExpired:
215 timeout_msg = (
216 f"Ruff execution timed out ({timeout}s limit exceeded).\n\n"
217 "This may indicate:\n"
218 " - Large codebase taking too long to process\n"
219 " - Need to increase timeout via --tool-options ruff:timeout=N"
220 )
221 return ToolResult(
222 name=tool.definition.name,
223 success=False,
224 output=timeout_msg,
225 issues_count=1, # Count timeout as execution failure
226 issues=initial_issues, # Include initial issues found
227 initial_issues_count=total_initial_count,
228 fixed_issues_count=0,
229 remaining_issues_count=1,
230 )
231 remaining_issues = parse_ruff_output(output=output)
232 remaining_count = len(remaining_issues)
234 # Compute fixed lint issues by diffing initial vs remaining (internal only)
235 # Not used for display; summary counts reflect totals.
237 # Calculate how many lint issues were actually fixed
238 fixed_lint_count: int = max(0, initial_count - remaining_count)
239 fixed_count: int = fixed_lint_count
241 # If there are remaining issues, check if any are fixable with unsafe fixes
242 # If unsafe fixes are disabled, check if any remaining issues are
243 # fixable with unsafe fixes
244 if remaining_count > 0 and not unsafe_fixes_enabled:
245 # Try running ruff with unsafe fixes in dry-run mode to see if it
246 # would fix more
247 # Use context manager to safely temporarily enable unsafe_fixes
248 remaining_unsafe = remaining_issues
249 success_unsafe: bool = False
250 output_unsafe: str = ""
251 with _temporary_option(tool, "unsafe_fixes", True):
252 # Build command with unsafe_fixes temporarily enabled
253 cmd_unsafe: list[str] = build_ruff_check_command(
254 tool=tool,
255 files=python_files,
256 fix=True,
257 )
258 # Command is built, option will be restored by context manager
259 # Run the command (option already restored)
260 try:
261 success_unsafe, output_unsafe = tool._run_subprocess(
262 cmd=cmd_unsafe,
263 timeout=timeout,
264 )
265 except subprocess.TimeoutExpired:
266 # If unsafe check times out, just continue with current results
267 # Don't fail the entire operation for this optional check
268 pass
269 else:
270 remaining_unsafe = parse_ruff_output(output=output_unsafe)
271 if len(remaining_unsafe) < remaining_count:
272 logger.warning(
273 "Some remaining issues could be fixed by enabling unsafe "
274 "fixes (use --tool-options ruff:unsafe_fixes=True)",
275 )
276 # Log remaining issues for debugging (if verbose)
277 # Note: Issue details are already included in the ToolResult.issues list
279 if not (success and remaining_count == 0):
280 overall_success = False
282 # Run ruff format if enabled (default: True)
283 if tool.options.get("format", False):
284 format_cmd: list[str] = build_ruff_format_command(
285 tool=tool,
286 files=python_files,
287 check_only=False,
288 )
289 format_success: bool = False
290 format_output: str = ""
291 try:
292 format_success, format_output = tool._run_subprocess(
293 cmd=format_cmd,
294 timeout=timeout,
295 )
296 # For ruff format, exit code 1 means files were formatted (success)
297 # Only consider it a failure if there were no initial format issues
298 if not format_success and initial_format_count > 0:
299 format_success = True
300 except subprocess.TimeoutExpired:
301 timeout_msg = (
302 f"Ruff execution timed out ({timeout}s limit exceeded).\n\n"
303 "This may indicate:\n"
304 " - Large codebase taking too long to process\n"
305 " - Need to increase timeout via --tool-options ruff:timeout=N"
306 )
307 return ToolResult(
308 name=tool.definition.name,
309 success=False,
310 output=timeout_msg,
311 issues_count=1, # Count timeout as execution failure
312 issues=remaining_issues, # Include any issues found before timeout
313 initial_issues_count=total_initial_count,
314 fixed_issues_count=fixed_lint_count,
315 remaining_issues_count=1,
316 )
317 # Formatting fixes are counted separately from lint fixes
318 if initial_format_count > 0:
319 fixed_count = fixed_lint_count + initial_format_count
320 # Only consider formatting failure if there are actual formatting
321 # issues. Don't fail the overall operation just because formatting
322 # failed when there are no issues
323 if not format_success and total_initial_count > 0:
324 overall_success = False
326 # Build concise, unified summary output for fmt runs
327 summary_lines: list[str] = []
328 if fixed_count > 0:
329 summary_lines.append(f"Fixed {fixed_count} issue(s)")
330 if remaining_count > 0:
331 summary_lines.append(
332 f"Found {remaining_count} issue(s) that cannot be auto-fixed",
333 )
334 final_output: str = (
335 "\n".join(summary_lines) if summary_lines else "No fixes applied."
336 )
338 return ToolResult(
339 name=tool.definition.name,
340 success=overall_success,
341 output=final_output,
342 # For fix operations, issues_count represents remaining issues
343 # that couldn't be fixed
344 issues_count=remaining_count,
345 # Only include remaining issues that couldn't be fixed
346 # (not successful format operations)
347 issues=remaining_issues,
348 initial_issues_count=total_initial_count,
349 fixed_issues_count=fixed_count,
350 remaining_issues_count=remaining_count,
351 # Store pre-fix issues so the display layer can show what was fixed
352 initial_issues=initial_issues if initial_issues else None,
353 )