Coverage for tests / integration / test_parallel_execution.py: 100%

1"""Integration tests for parallel tool execution.""" 

2 

3from __future__ import annotations 

4 

5import contextlib 

6import os 

7import tempfile 

8from collections.abc import Iterator 

9 

10import pytest 

11from assertpy import assert_that 

12 

13from lintro.plugins import ToolRegistry 

14from lintro.utils.tool_executor import run_lint_tools_simple 

15 

16 

17@pytest.fixture(autouse=True) 

18def set_lintro_test_mode_env(lintro_test_mode: object) -> Iterator[None]: 

19 """Set test mode for all tests in this module. 

20 

21 Args: 

22 lintro_test_mode: Shared fixture that manages env vars. 

23 

24 Yields: 

25 None: This fixture is used for its side effect only. 

26 """ 

27 yield 

28 

29 

30@pytest.fixture 

31def temp_python_files() -> Iterator[list[str]]: 

32 """Create multiple temporary Python files for parallel testing. 

33 

34 Yields: 

35 list[str]: List of paths to temporary Python files. 

36 """ 

37 files: list[str] = [] 

38 temp_dir = tempfile.mkdtemp() 

39 

40 # Create multiple files with various issues 

41 file_contents = [ 

42 ( 

43 "file1.py", 

44 "import sys\nimport os\n\ndef add(a, b):\n return a + b\n", 

45 ), 

46 ( 

47 "file2.py", 

48 "def greet(name: str) -> str:\n return f'Hello, {name}!'\n", 

49 ), 

50 ( 

51 "file3.py", 

52 "import json\n\ndata = {'key': 'value'}\n", 

53 ), 

54 ] 

55 

56 for filename, content in file_contents: 

57 file_path = os.path.join(temp_dir, filename) 

58 with open(file_path, "w") as f: 

59 f.write(content) 

60 files.append(file_path) 

61 

62 yield files 

63 

64 # Cleanup 

65 for file_path in files: 

66 with contextlib.suppress(FileNotFoundError): 

67 os.unlink(file_path) 

68 with contextlib.suppress(OSError): 

69 os.rmdir(temp_dir) 

70 

71 

72def test_check_multiple_files(temp_python_files: list[str]) -> None: 

73 """Test running check on multiple files. 

74 

75 Args: 

76 temp_python_files: Pytest fixture providing temp files. 

77 """ 

78 exit_code = run_lint_tools_simple( 

79 action="check", 

80 paths=temp_python_files, 

81 tools="ruff", 

82 tool_options=None, 

83 exclude=None, 

84 include_venv=False, 

85 group_by="file", 

86 output_format="grid", 

87 verbose=False, 

88 raw_output=False, 

89 ) 

90 

91 # Should complete without crashing 

92 assert_that(exit_code).is_instance_of(int) 

93 

94 

95def test_consistent_results_across_runs(temp_python_files: list[str]) -> None: 

96 """Test that multiple runs produce consistent results. 

97 

98 Args: 

99 temp_python_files: Pytest fixture providing temp files. 

100 """ 

101 # Run twice 

102 exit_code_1 = run_lint_tools_simple( 

103 action="check", 

104 paths=temp_python_files, 

105 tools="ruff", 

106 tool_options=None, 

107 exclude=None, 

108 include_venv=False, 

109 group_by="file", 

110 output_format="grid", 

111 verbose=False, 

112 ) 

113 

114 exit_code_2 = run_lint_tools_simple( 

115 action="check", 

116 paths=temp_python_files, 

117 tools="ruff", 

118 tool_options=None, 

119 exclude=None, 

120 include_venv=False, 

121 group_by="file", 

122 output_format="grid", 

123 verbose=False, 

124 ) 

125 

126 # Exit codes should match 

127 assert_that(exit_code_1).is_equal_to(exit_code_2) 

128 

129 

130def test_check_with_single_file(temp_python_files: list[str]) -> None: 

131 """Test check with single file. 

132 

133 Args: 

134 temp_python_files: Pytest fixture providing temp files. 

135 """ 

136 exit_code = run_lint_tools_simple( 

137 action="check", 

138 paths=[temp_python_files[0]], 

139 tools="ruff", 

140 tool_options=None, 

141 exclude=None, 

142 include_venv=False, 

143 group_by="file", 

144 output_format="grid", 

145 verbose=False, 

146 ) 

147 

148 assert_that(exit_code).is_instance_of(int) 

149 

150 

151def test_format_action(temp_python_files: list[str]) -> None: 

152 """Test format action. 

153 

154 Args: 

155 temp_python_files: Pytest fixture providing temp files. 

156 """ 

157 exit_code = run_lint_tools_simple( 

158 action="fmt", 

159 paths=temp_python_files, 

160 tools="ruff", 

161 tool_options=None, 

162 exclude=None, 

163 include_venv=False, 

164 group_by="file", 

165 output_format="grid", 

166 verbose=False, 

167 ) 

168 

169 assert_that(exit_code).is_instance_of(int) 

170 

171 

172def test_different_output_formats(temp_python_files: list[str]) -> None: 

173 """Test different output formats. 

174 

175 Args: 

176 temp_python_files: Pytest fixture providing temp files. 

177 """ 

178 for fmt in ["grid", "plain", "json"]: 

179 exit_code = run_lint_tools_simple( 

180 action="check", 

181 paths=temp_python_files, 

182 tools="ruff", 

183 tool_options=None, 

184 exclude=None, 

185 include_venv=False, 

186 group_by="file", 

187 output_format=fmt, 

188 verbose=False, 

189 ) 

190 assert_that(exit_code).is_instance_of(int) 

191 

192 

193def test_tool_definition_exists() -> None: 

194 """Test that ruff tool has proper definition.""" 

195 ruff_tool = ToolRegistry.get("ruff") 

196 

197 assert_that(ruff_tool).is_not_none() 

198 assert_that(ruff_tool.definition).is_not_none() 

199 assert_that(ruff_tool.definition.name).is_equal_to("ruff") 

200 

201 

202def test_tool_respects_execution_order(temp_python_files: list[str]) -> None: 

203 """Test that tool execution order is predictable. 

204 

205 Args: 

206 temp_python_files: Pytest fixture providing temp files. 

207 """ 

208 # Run multiple times to verify consistency 

209 results = [] 

210 for _ in range(3): 

211 exit_code = run_lint_tools_simple( 

212 action="check", 

213 paths=temp_python_files, 

214 tools="ruff", 

215 tool_options=None, 

216 exclude=None, 

217 include_venv=False, 

218 group_by="file", 

219 output_format="grid", 

220 verbose=False, 

221 ) 

222 results.append(exit_code) 

223 

224 # All runs should produce same exit code 

225 assert_that(len(set(results))).is_equal_to(1)