Coverage for lintro / tools / core / tool_manager.py: 86%

72 statements  

« prev     ^ index     » next       coverage.py v7.13.0, created at 2026-04-03 18:53 +0000

1"""Tool manager for Lintro. 

2 

3This module provides the ToolManager class for managing tool registration, 

4conflict resolution, and execution ordering using the plugin registry system. 

5""" 

6 

7from __future__ import annotations 

8 

9from dataclasses import dataclass, field 

10from typing import TYPE_CHECKING, Any 

11 

12from loguru import logger 

13 

14from lintro.plugins.discovery import discover_all_tools 

15from lintro.plugins.registry import ToolRegistry 

16from lintro.utils.unified_config import get_ordered_tools 

17 

18if TYPE_CHECKING: 

19 from lintro.plugins.base import BaseToolPlugin 

20 

21 

22@dataclass 

23class ToolManager: 

24 """Manager for tool registration and execution. 

25 

26 This class is responsible for: 

27 - Tool discovery and registration via plugin registry 

28 - Tool conflict resolution 

29 - Tool execution order (priority-based, alphabetical, or custom) 

30 - Tool configuration management 

31 

32 Tool ordering is controlled by [tool.lintro].tool_order in pyproject.toml: 

33 - "priority" (default): Formatters run before linters based on priority values 

34 - "alphabetical": Tools run in alphabetical order by name 

35 - "custom": Tools run in order specified by [tool.lintro].tool_order_custom 

36 """ 

37 

38 _initialized: bool = field(default=False, init=False) 

39 

40 def _ensure_initialized(self) -> None: 

41 """Ensure tools are discovered and registered.""" 

42 if not self._initialized: 

43 discover_all_tools() 

44 self._initialized = True 

45 

46 def get_tool(self, name: str) -> BaseToolPlugin: 

47 """Get a tool instance by name. 

48 

49 Args: 

50 name: The name of the tool (case-insensitive). 

51 

52 Returns: 

53 The tool/plugin instance. 

54 """ 

55 self._ensure_initialized() 

56 return ToolRegistry.get(name) 

57 

58 def get_tool_execution_order( 

59 self, 

60 tool_names: list[str], 

61 ignore_conflicts: bool = False, 

62 ) -> list[str]: 

63 """Get the order in which tools should be executed. 

64 

65 Tool ordering is controlled by [tool.lintro].tool_order in pyproject.toml: 

66 - "priority" (default): Formatters run before linters based on priority 

67 - "alphabetical": Tools run in alphabetical order by name 

68 - "custom": Tools run in order specified by [tool.lintro].tool_order_custom 

69 

70 This method also handles: 

71 - Tool conflicts (unless ignore_conflicts is True) 

72 

73 Args: 

74 tool_names: List of tool names to order. 

75 ignore_conflicts: If True, skip conflict checking. 

76 

77 Returns: 

78 List of tool names in execution order based on configured strategy. 

79 

80 Raises: 

81 ValueError: If duplicate tools are found in tool_names. 

82 """ 

83 if not tool_names: 

84 return [] 

85 

86 # Normalize names to lowercase 

87 normalized_names = [name.lower() for name in tool_names] 

88 

89 # Get tool instances 

90 tools: dict[str, BaseToolPlugin] = { 

91 name: self.get_tool(name) for name in normalized_names 

92 } 

93 

94 # Validate for duplicate tools 

95 seen_names: set[str] = set() 

96 duplicates: list[str] = [] 

97 for name in normalized_names: 

98 if name in seen_names: 

99 duplicates.append(name) 

100 else: 

101 seen_names.add(name) 

102 if duplicates: 

103 raise ValueError( 

104 f"Duplicate tools found in tool_names: {', '.join(duplicates)}", 

105 ) 

106 

107 # Get ordered tool names from unified config 

108 ordered_names = get_ordered_tools(normalized_names) 

109 

110 # Validate that all requested tools are preserved 

111 original_set = set(normalized_names) 

112 sorted_set = set(ordered_names) 

113 missing_tools = original_set - sorted_set 

114 if missing_tools: 

115 # Append missing tools in their original order 

116 missing_list = [n for n in normalized_names if n in missing_tools] 

117 ordered_names.extend(missing_list) 

118 logger.warning( 

119 f"Some tools were not found in ordered list and appended: " 

120 f"{missing_list}", 

121 ) 

122 

123 if ignore_conflicts: 

124 return ordered_names 

125 

126 # Build conflict graph 

127 conflict_graph: dict[str, set[str]] = {name: set() for name in normalized_names} 

128 for tool_name in normalized_names: 

129 tool_instance = tools[tool_name] 

130 for conflict in tool_instance.definition.conflicts_with: 

131 conflict_lower = conflict.lower() 

132 # Only add to conflict graph if conflict is in our tool list 

133 if conflict_lower in normalized_names: 

134 conflict_graph[tool_name].add(conflict_lower) 

135 conflict_graph[conflict_lower].add(tool_name) 

136 

137 # Resolve conflicts by keeping the first tool in ordered sequence 

138 result: list[str] = [] 

139 for tool_name in ordered_names: 

140 # Check if this tool conflicts with any already selected tools 

141 conflicts = conflict_graph[tool_name] & set(result) 

142 if not conflicts: 

143 result.append(tool_name) 

144 

145 return result 

146 

147 def set_tool_options( 

148 self, 

149 name: str, 

150 **options: Any, 

151 ) -> None: 

152 """Set options for a tool. 

153 

154 Args: 

155 name: The name of the tool. 

156 **options: The options to set. 

157 """ 

158 tool = self.get_tool(name) 

159 tool.set_options(**options) 

160 

161 def get_all_tools(self) -> dict[str, BaseToolPlugin]: 

162 """Get all registered tools. 

163 

164 Returns: 

165 Dictionary mapping tool names to plugin instances. 

166 """ 

167 self._ensure_initialized() 

168 return ToolRegistry.get_all() 

169 

170 def get_check_tools(self) -> dict[str, BaseToolPlugin]: 

171 """Get all tools that can check files. 

172 

173 Returns: 

174 Dictionary mapping tool names to plugin instances. 

175 """ 

176 self._ensure_initialized() 

177 return ToolRegistry.get_check_tools() 

178 

179 def get_fix_tools(self) -> dict[str, BaseToolPlugin]: 

180 """Get all tools that can fix files. 

181 

182 Returns: 

183 Dictionary mapping tool names to plugin instances. 

184 """ 

185 self._ensure_initialized() 

186 return ToolRegistry.get_fix_tools() 

187 

188 def get_tool_names(self) -> list[str]: 

189 """Get all registered tool names. 

190 

191 Returns: 

192 Sorted list of tool names. 

193 """ 

194 self._ensure_initialized() 

195 return ToolRegistry.get_names() 

196 

197 def is_tool_registered(self, name: str) -> bool: 

198 """Check if a tool is registered. 

199 

200 Args: 

201 name: Tool name (case-insensitive). 

202 

203 Returns: 

204 True if the tool is registered. 

205 """ 

206 self._ensure_initialized() 

207 return ToolRegistry.is_registered(name)