Coverage for tests / unit / utils / async_tool_executor / test_run_tools_parallel.py: 100%

1"""Tests for AsyncToolExecutor.run_tools_parallel method.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from typing import Any, cast 

7 

8from assertpy import assert_that 

9 

10from lintro.enums.action import Action 

11from lintro.models.core.tool_result import ToolResult 

12from lintro.plugins.base import BaseToolPlugin 

13from lintro.utils.async_tool_executor import AsyncToolExecutor 

14 

15from .conftest import MockToolDefinition, MockToolPlugin 

16 

17 

18def test_run_tools_parallel_success(executor: AsyncToolExecutor) -> None: 

19 """Test parallel execution of multiple tools. 

20 

21 Args: 

22 executor: AsyncToolExecutor fixture. 

23 """ 

24 # MockToolPlugin implements the same interface as BaseToolPlugin for testing 

25 tools = cast( 

26 list[tuple[str, BaseToolPlugin]], 

27 [ 

28 ("tool1", MockToolPlugin(definition=MockToolDefinition(name="tool1"))), 

29 ("tool2", MockToolPlugin(definition=MockToolDefinition(name="tool2"))), 

30 ("tool3", MockToolPlugin(definition=MockToolDefinition(name="tool3"))), 

31 ], 

32 ) 

33 

34 async def run_test() -> Any: 

35 return await executor.run_tools_parallel( 

36 tools, 

37 paths=["."], 

38 action=Action.CHECK, 

39 ) 

40 

41 results = asyncio.run(run_test()) 

42 

43 assert_that(results).is_length(3) 

44 for _name, result in results: 

45 assert_that(result.success).is_true() 

46 

47 

48def test_run_tools_parallel_with_failures(executor: AsyncToolExecutor) -> None: 

49 """Test parallel execution handles partial failures. 

50 

51 Args: 

52 executor: AsyncToolExecutor fixture. 

53 """ 

54 # MockToolPlugin implements the same interface as BaseToolPlugin for testing 

55 tools = cast( 

56 list[tuple[str, BaseToolPlugin]], 

57 [ 

58 ( 

59 "success_tool", 

60 MockToolPlugin(definition=MockToolDefinition(name="success_tool")), 

61 ), 

62 ( 

63 "failing_tool", 

64 MockToolPlugin( 

65 definition=MockToolDefinition(name="failing_tool"), 

66 check_result=ToolResult( 

67 name="failing_tool", 

68 success=False, 

69 output="Failed", 

70 issues_count=3, 

71 ), 

72 ), 

73 ), 

74 ], 

75 ) 

76 

77 async def run_test() -> Any: 

78 return await executor.run_tools_parallel( 

79 tools, 

80 paths=["."], 

81 action=Action.CHECK, 

82 ) 

83 

84 results = asyncio.run(run_test()) 

85 

86 assert_that(results).is_length(2) 

87 result_dict = dict(results) 

88 assert_that(result_dict["success_tool"].success).is_true() 

89 assert_that(result_dict["failing_tool"].success).is_false() 

90 

91 

92def test_run_tools_parallel_empty_list(executor: AsyncToolExecutor) -> None: 

93 """Test parallel execution with empty tools list. 

94 

95 Args: 

96 executor: AsyncToolExecutor fixture. 

97 """ 

98 

99 async def run_test() -> Any: 

100 return await executor.run_tools_parallel( 

101 tools=[], 

102 paths=["."], 

103 action=Action.CHECK, 

104 ) 

105 

106 results = asyncio.run(run_test()) 

107 

108 assert_that(results).is_empty() 

109 

110 

111def test_run_tools_parallel_single_tool(executor: AsyncToolExecutor) -> None: 

112 """Test parallel execution with single tool. 

113 

114 Args: 

115 executor: AsyncToolExecutor fixture. 

116 """ 

117 # MockToolPlugin implements the same interface as BaseToolPlugin for testing 

118 tools = cast( 

119 list[tuple[str, BaseToolPlugin]], 

120 [ 

121 ("single", MockToolPlugin(definition=MockToolDefinition(name="single"))), 

122 ], 

123 ) 

124 

125 async def run_test() -> Any: 

126 return await executor.run_tools_parallel( 

127 tools, 

128 paths=["."], 

129 action=Action.CHECK, 

130 ) 

131 

132 results = asyncio.run(run_test()) 

133 

134 assert_that(results).is_length(1) 

135 assert_that(results[0][0]).is_equal_to("single") 

136 

137 

138def test_run_tools_parallel_with_options_per_tool( 

139 executor: AsyncToolExecutor, 

140) -> None: 

141 """Test parallel execution passes correct options to each tool. 

142 

143 Args: 

144 executor: AsyncToolExecutor fixture. 

145 """ 

146 tool1_opts: dict[str, Any] = {} 

147 tool2_opts: dict[str, Any] = {} 

148 

149 def check1( 

150 paths: list[str], 

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

152 ) -> ToolResult: 

153 nonlocal tool1_opts 

154 tool1_opts = options or {} 

155 return ToolResult(name="tool1", success=True, output="", issues_count=0) 

156 

157 def check2( 

158 paths: list[str], 

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

160 ) -> ToolResult: 

161 nonlocal tool2_opts 

162 tool2_opts = options or {} 

163 return ToolResult(name="tool2", success=True, output="", issues_count=0) 

164 

165 tool1 = MockToolPlugin(definition=MockToolDefinition(name="tool1")) 

166 # Use object.__setattr__ to bypass dataclass method assignment restriction 

167 object.__setattr__(tool1, "check", check1) 

168 tool2 = MockToolPlugin(definition=MockToolDefinition(name="tool2")) 

169 object.__setattr__(tool2, "check", check2) 

170 

171 # MockToolPlugin implements the same interface as BaseToolPlugin for testing 

172 async def run_test() -> Any: 

173 return await executor.run_tools_parallel( 

174 tools=cast( 

175 list[tuple[str, BaseToolPlugin]], 

176 [("tool1", tool1), ("tool2", tool2)], 

177 ), 

178 paths=["."], 

179 action=Action.CHECK, 

180 options_per_tool={ 

181 "tool1": {"option_a": "value_a"}, 

182 "tool2": {"option_b": "value_b"}, 

183 }, 

184 ) 

185 

186 asyncio.run(run_test()) 

187 

188 assert_that(tool1_opts).contains_key("option_a") 

189 assert_that(tool2_opts).contains_key("option_b") 

190 assert_that(tool1_opts).does_not_contain_key("option_b")