Coverage for lintro / tools / definitions / taplo.py: 96%
154 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-04-03 18:53 +0000
1"""Taplo tool definition.
3Taplo is a TOML toolkit with linting and formatting capabilities.
4It validates TOML syntax and can format TOML files consistently.
5"""
7from __future__ import annotations
9import subprocess # nosec B404 - used safely with shell disabled
10from dataclasses import dataclass
11from typing import Any
13from loguru import logger
15from lintro._tool_versions import get_min_version
16from lintro.enums.doc_url_template import DocUrlTemplate
17from lintro.enums.tool_name import ToolName
18from lintro.enums.tool_type import ToolType
19from lintro.models.core.tool_result import ToolResult
20from lintro.parsers.taplo.taplo_issue import TaploIssue
21from lintro.parsers.taplo.taplo_parser import parse_taplo_output
22from lintro.plugins.base import BaseToolPlugin
23from lintro.plugins.protocol import ToolDefinition
24from lintro.plugins.registry import register_tool
25from lintro.tools.core.option_validators import (
26 filter_none_options,
27 validate_bool,
28 validate_str,
29)
31# Constants for Taplo configuration
32TAPLO_DEFAULT_TIMEOUT: int = 30
33TAPLO_DEFAULT_PRIORITY: int = 50
34TAPLO_FILE_PATTERNS: list[str] = ["*.toml"]
37@register_tool
38@dataclass
39class TaploPlugin(BaseToolPlugin):
40 """Taplo TOML linter and formatter plugin.
42 This plugin integrates Taplo with Lintro for linting and formatting
43 TOML files.
44 """
46 @property
47 def definition(self) -> ToolDefinition:
48 """Return the tool definition.
50 Returns:
51 ToolDefinition containing tool metadata.
52 """
53 return ToolDefinition(
54 name="taplo",
55 description="TOML toolkit with linting and formatting capabilities",
56 can_fix=True,
57 tool_type=ToolType.LINTER | ToolType.FORMATTER,
58 file_patterns=TAPLO_FILE_PATTERNS,
59 priority=TAPLO_DEFAULT_PRIORITY,
60 conflicts_with=[],
61 native_configs=["taplo.toml", ".taplo.toml"],
62 version_command=["taplo", "--version"],
63 min_version=get_min_version(ToolName.TAPLO),
64 default_options={
65 "timeout": TAPLO_DEFAULT_TIMEOUT,
66 "schema": None,
67 "aligned_arrays": None,
68 "aligned_entries": None,
69 "array_trailing_comma": None,
70 "indent_string": None,
71 "reorder_keys": None,
72 },
73 default_timeout=TAPLO_DEFAULT_TIMEOUT,
74 )
76 def set_options(
77 self,
78 schema: str | None = None,
79 aligned_arrays: bool | None = None,
80 aligned_entries: bool | None = None,
81 array_trailing_comma: bool | None = None,
82 indent_string: str | None = None,
83 reorder_keys: bool | None = None,
84 **kwargs: Any,
85 ) -> None:
86 """Set Taplo-specific options with validation.
88 Args:
89 schema: Path or URL to JSON schema for validation.
90 aligned_arrays: Align array entries.
91 aligned_entries: Align table entries.
92 array_trailing_comma: Add trailing comma in arrays.
93 indent_string: Indentation string (default: 2 spaces).
94 reorder_keys: Reorder keys alphabetically.
95 **kwargs: Additional base options.
96 """
97 validate_str(schema, "schema")
98 validate_bool(aligned_arrays, "aligned_arrays")
99 validate_bool(aligned_entries, "aligned_entries")
100 validate_bool(array_trailing_comma, "array_trailing_comma")
101 validate_str(indent_string, "indent_string")
102 validate_bool(reorder_keys, "reorder_keys")
104 options = filter_none_options(
105 schema=schema,
106 aligned_arrays=aligned_arrays,
107 aligned_entries=aligned_entries,
108 array_trailing_comma=array_trailing_comma,
109 indent_string=indent_string,
110 reorder_keys=reorder_keys,
111 )
112 super().set_options(**options, **kwargs)
114 def _build_format_args(self) -> list[str]:
115 """Build formatting CLI arguments for Taplo.
117 Returns:
118 CLI arguments for Taplo formatting options.
119 """
120 args: list[str] = []
122 if self.options.get("aligned_arrays"):
123 args.append("--option=aligned_arrays=true")
124 if self.options.get("aligned_entries"):
125 args.append("--option=aligned_entries=true")
126 if self.options.get("array_trailing_comma"):
127 args.append("--option=array_trailing_comma=true")
128 if self.options.get("indent_string"):
129 args.append(f"--option=indent_string={self.options['indent_string']}")
130 if self.options.get("reorder_keys"):
131 args.append("--option=reorder_keys=true")
133 return args
135 def _build_lint_args(self) -> list[str]:
136 """Build linting CLI arguments for Taplo.
138 Returns:
139 CLI arguments for Taplo linting options.
140 """
141 args: list[str] = []
143 if self.options.get("schema"):
144 args.extend(["--schema", str(self.options["schema"])])
146 return args
148 def _handle_timeout_error(
149 self,
150 timeout_val: int,
151 initial_count: int | None = None,
152 initial_issues: list[TaploIssue] | None = None,
153 ) -> ToolResult:
154 """Handle timeout errors consistently.
156 Args:
157 timeout_val: The timeout value that was exceeded.
158 initial_count: Optional initial issues count for fix operations.
159 initial_issues: Optional list of initial issues found before timeout.
161 Returns:
162 Standardized timeout error result.
163 """
164 timeout_msg = (
165 f"Taplo execution timed out ({timeout_val}s limit exceeded).\n\n"
166 "This may indicate:\n"
167 " - Large codebase taking too long to process\n"
168 " - Need to increase timeout via --tool-options taplo:timeout=N"
169 )
170 timeout_issue = TaploIssue(
171 file="execution",
172 line=0,
173 column=0,
174 level="error",
175 code="TIMEOUT",
176 message=f"Taplo execution timed out ({timeout_val}s limit exceeded)",
177 )
178 if initial_count is not None and initial_count > 0:
179 combined_issues = (initial_issues or []) + [timeout_issue]
180 return ToolResult(
181 name=self.definition.name,
182 success=False,
183 output=timeout_msg,
184 issues_count=len(combined_issues),
185 issues=combined_issues,
186 initial_issues_count=initial_count,
187 fixed_issues_count=0,
188 remaining_issues_count=initial_count,
189 )
190 return ToolResult(
191 name=self.definition.name,
192 success=False,
193 output=timeout_msg,
194 issues_count=1,
195 issues=[timeout_issue],
196 )
198 def doc_url(self, code: str) -> str | None:
199 """Return Taplo documentation URL.
201 Args:
202 code: Taplo rule code (e.g., "invalid_value").
204 Returns:
205 URL to the Taplo documentation, or None if code is empty.
206 """
207 if not code:
208 return None
209 return DocUrlTemplate.TAPLO
211 def check(self, paths: list[str], options: dict[str, object]) -> ToolResult:
212 """Check TOML files using Taplo.
214 Runs both `taplo lint` for syntax errors and `taplo fmt --check`
215 for formatting issues.
217 Args:
218 paths: List of file or directory paths to check.
219 options: Runtime options that override defaults.
221 Returns:
222 ToolResult with check results.
223 """
224 # Use shared preparation for version check, path validation, file discovery
225 ctx = self._prepare_execution(paths, options)
226 if ctx.should_skip:
227 return ctx.early_result # type: ignore[return-value]
229 all_issues: list[TaploIssue] = []
230 all_outputs: list[str] = []
231 all_success: bool = True
233 # Run taplo lint for syntax errors
234 lint_cmd: list[str] = self._get_executable_command(tool_name="taplo") + ["lint"]
235 lint_cmd.extend(self._build_lint_args())
236 lint_cmd.extend(ctx.rel_files)
238 logger.debug(
239 f"[TaploPlugin] Running lint: {' '.join(lint_cmd)} (cwd={ctx.cwd})",
240 )
241 try:
242 lint_success, lint_output = self._run_subprocess(
243 cmd=lint_cmd,
244 timeout=ctx.timeout,
245 cwd=ctx.cwd,
246 )
247 except subprocess.TimeoutExpired:
248 return self._handle_timeout_error(ctx.timeout)
250 if not lint_success:
251 all_success = False
252 if lint_output:
253 all_outputs.append(lint_output)
254 lint_issues = parse_taplo_output(output=lint_output)
255 all_issues.extend(lint_issues)
257 # Run taplo fmt --check for formatting issues
258 fmt_cmd: list[str] = self._get_executable_command(tool_name="taplo") + [
259 "fmt",
260 "--check",
261 ]
262 fmt_cmd.extend(self._build_format_args())
263 fmt_cmd.extend(ctx.rel_files)
265 logger.debug(
266 f"[TaploPlugin] Running format check: {' '.join(fmt_cmd)} (cwd={ctx.cwd})",
267 )
268 try:
269 fmt_success, fmt_output = self._run_subprocess(
270 cmd=fmt_cmd,
271 timeout=ctx.timeout,
272 cwd=ctx.cwd,
273 )
274 except subprocess.TimeoutExpired:
275 return self._handle_timeout_error(ctx.timeout)
277 if not fmt_success:
278 all_success = False
279 if fmt_output:
280 all_outputs.append(fmt_output)
281 # Format check output may contain file paths of files that need formatting
282 fmt_issues = parse_taplo_output(output=fmt_output)
283 all_issues.extend(fmt_issues)
285 count = len(all_issues)
286 output = "\n".join(all_outputs) if all_outputs else None
288 return ToolResult(
289 name=self.definition.name,
290 success=(all_success and count == 0),
291 output=output if count > 0 else None,
292 issues_count=count,
293 issues=all_issues,
294 cwd=ctx.cwd,
295 )
297 def fix(self, paths: list[str], options: dict[str, object]) -> ToolResult:
298 """Format TOML files using Taplo.
300 Args:
301 paths: List of file or directory paths to format.
302 options: Runtime options that override defaults.
304 Returns:
305 ToolResult with fix results.
306 """
307 # Use shared preparation for version check, path validation, file discovery
308 ctx = self._prepare_execution(
309 paths,
310 options,
311 no_files_message="No TOML files to format.",
312 )
313 if ctx.should_skip:
314 return ctx.early_result # type: ignore[return-value]
316 # Build check command for before/after comparison
317 check_cmd: list[str] = self._get_executable_command(tool_name="taplo") + [
318 "fmt",
319 "--check",
320 ]
321 check_cmd.extend(self._build_format_args())
322 check_cmd.extend(ctx.rel_files)
324 # Count initial formatting issues
325 try:
326 _, initial_output = self._run_subprocess(
327 cmd=check_cmd,
328 timeout=ctx.timeout,
329 cwd=ctx.cwd,
330 )
331 except subprocess.TimeoutExpired:
332 return self._handle_timeout_error(timeout_val=ctx.timeout, initial_count=0)
334 initial_issues = parse_taplo_output(output=initial_output)
335 initial_count = len(initial_issues)
337 # Also check for lint errors (syntax issues that formatting won't fix)
338 lint_cmd: list[str] = self._get_executable_command(tool_name="taplo") + ["lint"]
339 lint_cmd.extend(self._build_lint_args())
340 lint_cmd.extend(ctx.rel_files)
342 try:
343 _, lint_output = self._run_subprocess(
344 cmd=lint_cmd,
345 timeout=ctx.timeout,
346 cwd=ctx.cwd,
347 )
348 except subprocess.TimeoutExpired:
349 return self._handle_timeout_error(
350 timeout_val=ctx.timeout,
351 initial_count=initial_count,
352 )
354 lint_issues = parse_taplo_output(output=lint_output)
355 initial_issues.extend(lint_issues)
356 initial_count = len(initial_issues)
358 # Apply formatting with taplo fmt
359 fix_cmd: list[str] = self._get_executable_command(tool_name="taplo") + ["fmt"]
360 fix_cmd.extend(self._build_format_args())
361 fix_cmd.extend(ctx.rel_files)
363 logger.debug(f"[TaploPlugin] Fixing: {' '.join(fix_cmd)} (cwd={ctx.cwd})")
364 try:
365 _, _ = self._run_subprocess(
366 cmd=fix_cmd,
367 timeout=ctx.timeout,
368 cwd=ctx.cwd,
369 )
370 except subprocess.TimeoutExpired:
371 return self._handle_timeout_error(
372 timeout_val=ctx.timeout,
373 initial_count=initial_count,
374 )
376 # Check for remaining formatting issues
377 try:
378 final_success, final_output = self._run_subprocess(
379 cmd=check_cmd,
380 timeout=ctx.timeout,
381 cwd=ctx.cwd,
382 )
383 except subprocess.TimeoutExpired:
384 return self._handle_timeout_error(
385 timeout_val=ctx.timeout,
386 initial_count=initial_count,
387 )
389 remaining_format_issues = parse_taplo_output(output=final_output)
391 # Re-check lint errors (these won't be fixed by formatting)
392 try:
393 _, final_lint_output = self._run_subprocess(
394 cmd=lint_cmd,
395 timeout=ctx.timeout,
396 cwd=ctx.cwd,
397 )
398 except subprocess.TimeoutExpired:
399 return self._handle_timeout_error(
400 timeout_val=ctx.timeout,
401 initial_count=initial_count,
402 )
404 remaining_lint_issues = parse_taplo_output(output=final_lint_output)
406 all_remaining_issues = remaining_format_issues + remaining_lint_issues
407 remaining_count = len(all_remaining_issues)
408 fixed_count = max(0, initial_count - remaining_count)
410 # Build summary
411 summary: list[str] = []
412 if fixed_count > 0:
413 summary.append(f"Fixed {fixed_count} issue(s)")
414 if remaining_count > 0:
415 summary.append(
416 f"Found {remaining_count} issue(s) that cannot be auto-fixed",
417 )
418 elif remaining_count == 0 and fixed_count > 0:
419 summary.append("All issues were successfully auto-fixed")
420 final_summary = "\n".join(summary) if summary else "No fixes applied."
422 return ToolResult(
423 name=self.definition.name,
424 success=(remaining_count == 0),
425 output=final_summary,
426 issues_count=remaining_count,
427 issues=all_remaining_issues,
428 initial_issues=initial_issues if initial_issues else None,
429 initial_issues_count=initial_count,
430 fixed_issues_count=fixed_count,
431 remaining_issues_count=remaining_count,
432 cwd=ctx.cwd,
433 )