Coverage for lintro / plugins / execution_preparation.py: 96%
89 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-04-03 18:53 +0000
1"""Execution preparation utilities for tool plugins.
3This module provides execution preparation, version checking, and config injection.
4"""
6from __future__ import annotations
8import os
9from typing import Any
11from loguru import logger
13from lintro.config.lintro_config import LintroConfig
14from lintro.models.core.tool_result import ToolResult
15from lintro.plugins.file_discovery import discover_files, get_cwd, validate_paths
16from lintro.plugins.protocol import ToolDefinition
18# Constants for default values
19DEFAULT_TIMEOUT: int = 30
22def get_effective_timeout(
23 timeout: int | float | None,
24 options: dict[str, object],
25 default_timeout: int,
26) -> float:
27 """Get the effective timeout value.
29 Args:
30 timeout: Override timeout value, or None to use default.
31 options: Options dict that may contain timeout.
32 default_timeout: Default timeout from definition.
34 Returns:
35 Timeout value in seconds.
36 """
37 if timeout is not None:
38 return float(timeout)
40 raw_timeout = options.get("timeout", default_timeout)
41 if isinstance(raw_timeout, (int, float)):
42 return float(raw_timeout)
44 # Warn about invalid timeout value
45 if raw_timeout is not None:
46 type_name = type(raw_timeout).__name__
47 logger.warning(
48 f"Invalid timeout value {raw_timeout!r} (type {type_name}), "
49 f"using default {default_timeout}s",
50 )
51 return float(default_timeout)
54def get_executable_command(tool_name: str) -> list[str]:
55 """Get the command prefix to execute a tool.
57 Delegates to CommandBuilderRegistry for language-specific logic.
59 Args:
60 tool_name: Name of the tool executable.
62 Returns:
63 Command prefix list.
64 """
65 from lintro.enums.tool_name import normalize_tool_name
66 from lintro.tools.core.command_builders import CommandBuilderRegistry
68 try:
69 tool_name_enum = normalize_tool_name(tool_name)
70 except ValueError:
71 tool_name_enum = None
73 result: list[str] = CommandBuilderRegistry.get_command(tool_name, tool_name_enum)
74 return result
77def verify_tool_version(definition: ToolDefinition) -> ToolResult | None:
78 """Verify that the tool meets minimum version requirements.
80 Args:
81 definition: Tool definition with name.
83 Returns:
84 None if version check passes, or a skip result if it fails.
85 """
86 from lintro.tools.core.version_requirements import check_tool_version
88 command = get_executable_command(definition.name)
89 version_info = check_tool_version(definition.name, command)
91 if version_info.version_check_passed:
92 return None
94 skip_message = (
95 f"Skipping {definition.name}: {version_info.error_message}. "
96 f"Minimum required: {version_info.min_version}. "
97 f"{version_info.install_hint}"
98 )
100 return ToolResult(
101 name=definition.name,
102 success=True,
103 output=skip_message,
104 issues_count=0,
105 skipped=True,
106 skip_reason=version_info.error_message,
107 )
110def prepare_execution(
111 paths: list[str],
112 options: dict[str, object],
113 definition: ToolDefinition,
114 exclude_patterns: list[str],
115 include_venv: bool,
116 current_options: dict[str, object],
117 no_files_message: str = "No files to check.",
118) -> dict[str, Any]:
119 """Prepare execution context with common boilerplate steps.
121 This function consolidates repeated patterns:
122 - Merge options with defaults
123 - Validate input paths
124 - Discover files matching patterns (returns early if none found)
125 - Verify tool version requirements (skipped when no files match)
126 - Compute working directory and relative paths
127 - Compute timeout for execution
129 Args:
130 paths: Input paths to process.
131 options: Runtime options to merge with defaults.
132 definition: Tool definition.
133 exclude_patterns: Patterns to exclude.
134 include_venv: Whether to include venv files.
135 current_options: Current plugin options.
136 no_files_message: Message when no files are found.
138 Returns:
139 Dictionary with files, rel_files, cwd, timeout, and optional early_result.
140 """
141 # Merge runtime options with defaults
142 merged_options = dict(current_options)
143 merged_options.update(options)
145 # Validate paths
146 validate_paths(paths)
147 if not paths:
148 return {
149 "early_result": ToolResult(
150 name=definition.name,
151 success=True,
152 output=no_files_message,
153 issues_count=0,
154 ),
155 }
157 # Discover files matching tool patterns
158 files = discover_files(
159 paths=paths,
160 definition=definition,
161 exclude_patterns=exclude_patterns,
162 include_venv=include_venv,
163 )
165 if not files:
166 file_type = "files"
167 patterns = definition.file_patterns
168 if patterns:
169 extensions = [p.replace("*", "") for p in patterns if p.startswith("*.")]
170 if extensions:
171 file_type = "/".join(extensions) + " files"
173 return {
174 "early_result": ToolResult(
175 name=definition.name,
176 success=True,
177 output=f"No {file_type} found to check.",
178 issues_count=0,
179 ),
180 }
182 # Check version requirements (only when files exist to check)
183 version_result = verify_tool_version(definition)
184 if version_result is not None:
185 return {"early_result": version_result}
187 logger.debug(f"Files to process: {files}")
189 # Compute cwd and relative paths
190 cwd = get_cwd(files)
191 rel_files = [os.path.relpath(f, cwd) if cwd else f for f in files]
193 # Get timeout (keep as float to preserve precision)
194 timeout_value = merged_options.get("timeout")
195 timeout = get_effective_timeout(
196 timeout_value if isinstance(timeout_value, (int, float)) else None,
197 merged_options,
198 definition.default_timeout,
199 )
201 logger.debug(
202 f"Prepared execution: {len(files)} files, cwd={cwd}, timeout={timeout}s",
203 )
204 return {
205 "files": files,
206 "rel_files": rel_files,
207 "cwd": cwd,
208 "timeout": timeout,
209 }
212# -------------------------------------------------------------------------
213# Lintro Config Support Functions
214# -------------------------------------------------------------------------
217def get_lintro_config() -> LintroConfig:
218 """Get the current Lintro configuration.
220 Returns:
221 The current LintroConfig instance.
222 """
223 from lintro.tools.core.config_injection import _get_lintro_config
225 result: LintroConfig = _get_lintro_config()
226 return result
229def get_enforced_settings(
230 lintro_config: LintroConfig | None = None,
231) -> dict[str, object]:
232 """Get enforced settings as a dictionary.
234 Args:
235 lintro_config: Optional config to use, or None to get current.
237 Returns:
238 Dictionary of enforced settings.
239 """
240 from lintro.tools.core.config_injection import _get_enforced_settings
242 config = lintro_config or get_lintro_config()
243 result: dict[str, object] = _get_enforced_settings(lintro_config=config)
244 return result
247def get_enforce_cli_args(
248 tool_name: str,
249 lintro_config: LintroConfig | None = None,
250) -> list[str]:
251 """Get CLI arguments for enforced settings.
253 Args:
254 tool_name: Name of the tool.
255 lintro_config: Optional config to use, or None to get current.
257 Returns:
258 List of CLI arguments for enforced settings.
259 """
260 from lintro.tools.core.config_injection import _get_enforce_cli_args
262 config = lintro_config or get_lintro_config()
263 result: list[str] = _get_enforce_cli_args(tool_name=tool_name, lintro_config=config)
264 return result
267def get_defaults_config_args(
268 tool_name: str,
269 lintro_config: LintroConfig | None = None,
270) -> list[str]:
271 """Get CLI arguments for defaults config injection.
273 Args:
274 tool_name: Name of the tool.
275 lintro_config: Optional config to use, or None to get current.
277 Returns:
278 List of CLI arguments for defaults config.
279 """
280 from lintro.tools.core.config_injection import _get_defaults_config_args
282 config = lintro_config or get_lintro_config()
283 result: list[str] = _get_defaults_config_args(
284 tool_name=tool_name,
285 lintro_config=config,
286 )
287 return result
290def should_use_lintro_config(tool_name: str) -> bool:
291 """Check if Lintro config should be used for this tool.
293 Args:
294 tool_name: Name of the tool.
296 Returns:
297 True if Lintro config should be used.
298 """
299 from lintro.tools.core.config_injection import _should_use_lintro_config
301 result: bool = _should_use_lintro_config(tool_name=tool_name)
302 return result
305def build_config_args(
306 tool_name: str,
307 lintro_config: LintroConfig | None = None,
308) -> list[str]:
309 """Build combined CLI arguments for config injection.
311 Args:
312 tool_name: Name of the tool.
313 lintro_config: Optional config to use, or None to get current.
315 Returns:
316 List of combined CLI arguments for config.
317 """
318 from lintro.tools.core.config_injection import _build_config_args
320 config = lintro_config or get_lintro_config()
321 result: list[str] = _build_config_args(tool_name=tool_name, lintro_config=config)
322 return result