Coverage for tests / unit / plugins / test_file_processor.py: 100%

1"""Unit tests for file processor module.""" 

2 

3from __future__ import annotations 

4 

5import pytest 

6from assertpy import assert_that 

7 

8from lintro.parsers.base_issue import BaseIssue 

9from lintro.plugins.file_processor import AggregatedResult, FileProcessingResult 

10 

11# ============================================================================= 

12# Tests for FileProcessingResult dataclass 

13# ============================================================================= 

14 

15 

16def test_file_processing_result_success() -> None: 

17 """FileProcessingResult stores success state correctly.""" 

18 result = FileProcessingResult( 

19 success=True, 

20 output="No issues found", 

21 issues=[], 

22 ) 

23 assert_that(result.success).is_true() 

24 assert_that(result.output).is_equal_to("No issues found") 

25 assert_that(result.issues).is_empty() 

26 assert_that(result.skipped).is_false() 

27 assert_that(result.error).is_none() 

28 

29 

30def test_file_processing_result_with_issues() -> None: 

31 """FileProcessingResult stores issues correctly.""" 

32 issue = BaseIssue(file="test.txt", line=1, message="Test issue") 

33 result = FileProcessingResult( 

34 success=False, 

35 output="Found 1 issue", 

36 issues=[issue], 

37 ) 

38 assert_that(result.success).is_false() 

39 assert_that(result.issues).is_length(1) 

40 assert_that(result.issues[0].message).is_equal_to("Test issue") 

41 

42 

43def test_file_processing_result_skipped() -> None: 

44 """FileProcessingResult marks skipped files correctly.""" 

45 result = FileProcessingResult( 

46 success=False, 

47 output="", 

48 issues=[], 

49 skipped=True, 

50 ) 

51 assert_that(result.skipped).is_true() 

52 

53 

54def test_file_processing_result_with_error() -> None: 

55 """FileProcessingResult stores error messages correctly.""" 

56 result = FileProcessingResult( 

57 success=False, 

58 output="", 

59 issues=[], 

60 error="Connection timeout", 

61 ) 

62 assert_that(result.error).is_equal_to("Connection timeout") 

63 

64 

65# ============================================================================= 

66# Tests for AggregatedResult dataclass 

67# ============================================================================= 

68 

69 

70def test_aggregated_result_defaults() -> None: 

71 """AggregatedResult has correct default values.""" 

72 result = AggregatedResult() 

73 assert_that(result.all_success).is_true() 

74 assert_that(result.all_issues).is_empty() 

75 assert_that(result.all_outputs).is_empty() 

76 assert_that(result.skipped_files).is_empty() 

77 assert_that(result.execution_failures).is_equal_to(0) 

78 assert_that(result.total_issues).is_equal_to(0) 

79 

80 

81# ============================================================================= 

82# Tests for AggregatedResult.add_file_result method 

83# ============================================================================= 

84 

85 

86def test_add_file_result_success() -> None: 

87 """Add successful file result updates aggregated state correctly.""" 

88 aggregated = AggregatedResult() 

89 file_result = FileProcessingResult( 

90 success=True, 

91 output="", 

92 issues=[], 

93 ) 

94 

95 aggregated.add_file_result("/path/to/file.txt", file_result) 

96 

97 assert_that(aggregated.all_success).is_true() 

98 assert_that(aggregated.total_issues).is_equal_to(0) 

99 assert_that(aggregated.all_outputs).is_empty() 

100 

101 

102def test_add_file_result_with_issues() -> None: 

103 """Add file result with issues updates aggregated state correctly.""" 

104 aggregated = AggregatedResult() 

105 issue = BaseIssue(file="test.txt", line=1, message="Test issue") 

106 file_result = FileProcessingResult( 

107 success=True, 

108 output="Found 1 issue", 

109 issues=[issue], 

110 ) 

111 

112 aggregated.add_file_result("/path/to/test.txt", file_result) 

113 

114 assert_that(aggregated.all_success).is_true() 

115 assert_that(aggregated.total_issues).is_equal_to(1) 

116 assert_that(aggregated.all_issues).is_length(1) 

117 assert_that(aggregated.all_outputs).is_length(1) 

118 

119 

120def test_add_file_result_failure() -> None: 

121 """Add failed file result updates all_success correctly.""" 

122 aggregated = AggregatedResult() 

123 file_result = FileProcessingResult( 

124 success=False, 

125 output="Error occurred", 

126 issues=[], 

127 ) 

128 

129 aggregated.add_file_result("/path/to/file.txt", file_result) 

130 

131 assert_that(aggregated.all_success).is_false() 

132 assert_that(aggregated.all_outputs).contains("Error occurred") 

133 

134 

135def test_add_file_result_skipped() -> None: 

136 """Add skipped file result updates skipped_files correctly.""" 

137 aggregated = AggregatedResult() 

138 file_result = FileProcessingResult( 

139 success=False, 

140 output="", 

141 issues=[], 

142 skipped=True, 

143 ) 

144 

145 aggregated.add_file_result("/path/to/file.txt", file_result) 

146 

147 assert_that(aggregated.all_success).is_false() 

148 assert_that(aggregated.skipped_files).contains("/path/to/file.txt") 

149 assert_that(aggregated.execution_failures).is_equal_to(1) 

150 

151 

152def test_add_file_result_with_error() -> None: 

153 """Add file result with error updates execution_failures correctly.""" 

154 aggregated = AggregatedResult() 

155 file_result = FileProcessingResult( 

156 success=False, 

157 output="", 

158 issues=[], 

159 error="Connection refused", 

160 ) 

161 

162 aggregated.add_file_result("/path/to/file.txt", file_result) 

163 

164 assert_that(aggregated.all_success).is_false() 

165 assert_that(aggregated.execution_failures).is_equal_to(1) 

166 assert_that(aggregated.all_outputs[0]).contains("Connection refused") 

167 

168 

169def test_add_multiple_file_results() -> None: 

170 """Add multiple file results accumulates correctly.""" 

171 aggregated = AggregatedResult() 

172 

173 # First file: success with issues 

174 issue1 = BaseIssue(file="file1.txt", line=1, message="Issue 1") 

175 result1 = FileProcessingResult( 

176 success=True, 

177 output="File 1 output", 

178 issues=[issue1], 

179 ) 

180 aggregated.add_file_result("/path/to/file1.txt", result1) 

181 

182 # Second file: failure 

183 result2 = FileProcessingResult( 

184 success=False, 

185 output="File 2 error", 

186 issues=[], 

187 ) 

188 aggregated.add_file_result("/path/to/file2.txt", result2) 

189 

190 # Third file: success with multiple issues 

191 issue2 = BaseIssue(file="file3.txt", line=2, message="Issue 2") 

192 issue3 = BaseIssue(file="file3.txt", line=5, message="Issue 3") 

193 result3 = FileProcessingResult( 

194 success=True, 

195 output="File 3 output", 

196 issues=[issue2, issue3], 

197 ) 

198 aggregated.add_file_result("/path/to/file3.txt", result3) 

199 

200 assert_that(aggregated.all_success).is_false() 

201 assert_that(aggregated.total_issues).is_equal_to(3) 

202 assert_that(aggregated.all_issues).is_length(3) 

203 assert_that(aggregated.all_outputs).is_length(3) 

204 

205 

206# ============================================================================= 

207# Tests for AggregatedResult.build_output method 

208# ============================================================================= 

209 

210 

211def test_build_output_empty() -> None: 

212 """Build output returns None when no output.""" 

213 aggregated = AggregatedResult() 

214 output = aggregated.build_output() 

215 assert_that(output).is_none() 

216 

217 

218def test_build_output_with_outputs() -> None: 

219 """Build output combines all outputs correctly.""" 

220 aggregated = AggregatedResult() 

221 aggregated.all_outputs = ["Output 1", "Output 2"] 

222 

223 output = aggregated.build_output() 

224 

225 assert_that(output).contains("Output 1") 

226 assert_that(output).contains("Output 2") 

227 assert_that(output).contains("\n") 

228 

229 

230def test_build_output_with_skipped_files() -> None: 

231 """Build output includes skipped files information.""" 

232 aggregated = AggregatedResult() 

233 aggregated.skipped_files = ["/path/to/file1.txt", "/path/to/file2.txt"] 

234 aggregated.execution_failures = 2 

235 

236 output = aggregated.build_output() 

237 

238 assert_that(output).is_not_none() 

239 assert_that(output).contains("Skipped/failed 2 file(s)") 

240 assert_that(output).contains("/path/to/file1.txt") 

241 assert_that(output).contains("/path/to/file2.txt") 

242 

243 

244def test_build_output_with_timeout() -> None: 

245 """Build output includes timeout value when provided.""" 

246 aggregated = AggregatedResult() 

247 aggregated.skipped_files = ["/path/to/file.txt"] 

248 aggregated.execution_failures = 1 

249 

250 output = aggregated.build_output(timeout=30) 

251 

252 assert_that(output).is_not_none() 

253 assert_that(output).contains("timeout: 30s") 

254 

255 

256def test_build_output_with_execution_errors() -> None: 

257 """Build output includes execution error count.""" 

258 aggregated = AggregatedResult() 

259 aggregated.execution_failures = 3 

260 

261 output = aggregated.build_output() 

262 

263 assert_that(output).is_not_none() 

264 assert_that(output).contains("Failed to process 3 file(s)") 

265 

266 

267def test_build_output_combines_outputs_and_errors() -> None: 

268 """Build output combines regular outputs and error information.""" 

269 aggregated = AggregatedResult() 

270 aggregated.all_outputs = ["Regular output"] 

271 aggregated.skipped_files = ["/path/to/skipped.txt"] 

272 aggregated.execution_failures = 1 

273 

274 output = aggregated.build_output(timeout=60) 

275 

276 assert_that(output).is_not_none() 

277 assert_that(output).contains("Regular output") 

278 assert_that(output).contains("Skipped/failed 1 file(s)") 

279 assert_that(output).contains("timeout: 60s") 

280 

281 

282# ============================================================================= 

283# Tests for edge cases 

284# ============================================================================= 

285 

286 

287def test_empty_output_not_added() -> None: 

288 """Empty output from successful file is not added to all_outputs.""" 

289 aggregated = AggregatedResult() 

290 file_result = FileProcessingResult( 

291 success=True, 

292 output="", 

293 issues=[], 

294 ) 

295 

296 aggregated.add_file_result("/path/to/file.txt", file_result) 

297 

298 assert_that(aggregated.all_outputs).is_empty() 

299 

300 

301@pytest.mark.parametrize( 

302 ("output", "expected"), 

303 [ 

304 (" ", None), 

305 ("\n\n", None), 

306 ("\t", None), 

307 ("", None), 

308 ], 

309 ids=[ 

310 "whitespace_only", 

311 "newlines_only", 

312 "tab_only", 

313 "empty_string", 

314 ], 

315) 

316def test_build_output_whitespace_returns_none( 

317 output: str, 

318 expected: str | None, 

319) -> None: 

320 """Build output returns None for whitespace-only outputs. 

321 

322 Args: 

323 output: The whitespace-only output string to test. 

324 expected: The expected result (None for whitespace-only). 

325 """ 

326 aggregated = AggregatedResult() 

327 aggregated.all_outputs = [output] 

328 

329 result = aggregated.build_output() 

330 

331 assert_that(result).is_equal_to(expected)