Coverage for tests / unit / utils / test_streaming_output.py: 100%

1"""Tests for lintro.utils.streaming_output module.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import tempfile 

7from pathlib import Path 

8from unittest.mock import MagicMock 

9 

10import pytest 

11from assertpy import assert_that 

12 

13from lintro.enums.action import Action 

14from lintro.models.core.tool_result import ToolResult 

15from lintro.utils.streaming_output import ( 

16 StreamingResultHandler, 

17 create_streaming_handler, 

18) 

19 

20 

21@pytest.fixture 

22def mock_tool_result() -> ToolResult: 

23 """Create a mock ToolResult for testing. 

24 

25 Returns: 

26 A ToolResult with success=True and 2 issues. 

27 """ 

28 return ToolResult( 

29 name="test_tool", 

30 success=True, 

31 issues_count=2, 

32 output="Test output", 

33 ) 

34 

35 

36@pytest.fixture 

37def mock_tool_result_with_issues() -> ToolResult: 

38 """Create a mock ToolResult with issues. 

39 

40 Returns: 

41 A ToolResult with success=False and 1 issue. 

42 """ 

43 mock_issue = MagicMock() 

44 mock_issue.file = "test.py" 

45 mock_issue.line = 10 

46 mock_issue.column = 5 

47 mock_issue.message = "Test error" 

48 

49 return ToolResult( 

50 name="test_tool", 

51 success=False, 

52 issues_count=1, 

53 output="Error output", 

54 issues=[mock_issue], 

55 ) 

56 

57 

58@pytest.fixture 

59def mock_fix_result() -> ToolResult: 

60 """Create a mock ToolResult for fix action. 

61 

62 Returns: 

63 A ToolResult with fix counts set. 

64 """ 

65 return ToolResult( 

66 name="test_tool", 

67 success=True, 

68 issues_count=0, 

69 output="Fixed", 

70 fixed_issues_count=3, 

71 remaining_issues_count=1, 

72 ) 

73 

74 

75def test_handler_stores_output_format() -> None: 

76 """Handler stores output format.""" 

77 handler = StreamingResultHandler(output_format="json", action=Action.CHECK) 

78 assert_that(handler.output_format).is_equal_to("json") 

79 

80 

81def test_handler_stores_action() -> None: 

82 """Handler stores action.""" 

83 handler = StreamingResultHandler(output_format="grid", action=Action.FIX) 

84 assert_that(handler.action).is_equal_to(Action.FIX) 

85 

86 

87def test_handler_initializes_totals() -> None: 

88 """Handler initializes totals dictionary.""" 

89 handler = StreamingResultHandler(output_format="grid", action=Action.CHECK) 

90 totals = handler.get_totals() 

91 

92 assert_that(totals).contains_key("issues", "fixed", "remaining") 

93 assert_that(totals["issues"]).is_equal_to(0) 

94 

95 

96def test_handle_result_updates_totals(mock_tool_result: ToolResult) -> None: 

97 """Handle result updates totals. 

98 

99 Args: 

100 mock_tool_result: Fixture providing a mock ToolResult. 

101 """ 

102 handler = StreamingResultHandler(output_format="grid", action=Action.CHECK) 

103 handler.handle_result(mock_tool_result) 

104 

105 totals = handler.get_totals() 

106 assert_that(totals["tools_run"]).is_equal_to(1) 

107 assert_that(totals["issues"]).is_equal_to(2) 

108 

109 

110def test_handle_result_tracks_failures( 

111 mock_tool_result_with_issues: ToolResult, 

112) -> None: 

113 """Handle result tracks failed tools. 

114 

115 Args: 

116 mock_tool_result_with_issues: Fixture providing a failed ToolResult. 

117 """ 

118 handler = StreamingResultHandler(output_format="grid", action=Action.CHECK) 

119 handler.handle_result(mock_tool_result_with_issues) 

120 

121 totals = handler.get_totals() 

122 assert_that(totals["tools_failed"]).is_equal_to(1) 

123 

124 

125def test_handle_result_tracks_fix_counts(mock_fix_result: ToolResult) -> None: 

126 """Handle result tracks fix counts for FIX action. 

127 

128 Args: 

129 mock_fix_result: Fixture providing a ToolResult with fix counts. 

130 """ 

131 handler = StreamingResultHandler(output_format="grid", action=Action.FIX) 

132 handler.handle_result(mock_fix_result) 

133 

134 totals = handler.get_totals() 

135 assert_that(totals["fixed"]).is_equal_to(3) 

136 assert_that(totals["remaining"]).is_equal_to(1) 

137 

138 

139def test_handle_result_buffers_results(mock_tool_result: ToolResult) -> None: 

140 """Handle result buffers results. 

141 

142 Args: 

143 mock_tool_result: Fixture providing a mock ToolResult. 

144 """ 

145 handler = StreamingResultHandler(output_format="grid", action=Action.CHECK) 

146 handler.handle_result(mock_tool_result) 

147 

148 results = handler.get_results() 

149 assert_that(results).is_length(1) 

150 assert_that(results[0].name).is_equal_to("test_tool") 

151 

152 

153def test_get_exit_code_returns_zero_on_success() -> None: 

154 """Get exit code returns 0 when all tools pass.""" 

155 success_result = ToolResult( 

156 name="test_tool", 

157 success=True, 

158 issues_count=0, 

159 ) 

160 handler = StreamingResultHandler(output_format="grid", action=Action.CHECK) 

161 handler.handle_result(success_result) 

162 

163 assert_that(handler.get_exit_code()).is_equal_to(0) 

164 

165 

166def test_get_exit_code_returns_one_on_failure( 

167 mock_tool_result_with_issues: ToolResult, 

168) -> None: 

169 """Get exit code returns 1 when tools fail. 

170 

171 Args: 

172 mock_tool_result_with_issues: Fixture providing a failed ToolResult. 

173 """ 

174 handler = StreamingResultHandler(output_format="grid", action=Action.CHECK) 

175 handler.handle_result(mock_tool_result_with_issues) 

176 

177 assert_that(handler.get_exit_code()).is_equal_to(1) 

178 

179 

180def test_get_exit_code_returns_one_on_issues(mock_tool_result: ToolResult) -> None: 

181 """Get exit code returns 1 when issues found in check mode. 

182 

183 Args: 

184 mock_tool_result: Fixture providing a mock ToolResult. 

185 """ 

186 handler = StreamingResultHandler(output_format="grid", action=Action.CHECK) 

187 handler.handle_result(mock_tool_result) 

188 

189 assert_that(handler.get_exit_code()).is_equal_to(1) 

190 

191 

192def test_context_manager_opens_file() -> None: 

193 """Context manager opens output file.""" 

194 with tempfile.NamedTemporaryFile(delete=False, suffix=".json") as f: 

195 temp_path = f.name 

196 

197 try: 

198 handler = StreamingResultHandler( 

199 output_format="jsonl", 

200 action=Action.CHECK, 

201 output_file=temp_path, 

202 ) 

203 with handler: 

204 assert_that(handler._file_handle).is_not_none() 

205 finally: 

206 Path(temp_path).unlink() 

207 

208 

209def test_context_manager_closes_file() -> None: 

210 """Context manager closes output file.""" 

211 with tempfile.NamedTemporaryFile(delete=False, suffix=".json") as f: 

212 temp_path = f.name 

213 

214 try: 

215 handler = StreamingResultHandler( 

216 output_format="jsonl", 

217 action=Action.CHECK, 

218 output_file=temp_path, 

219 ) 

220 with handler: 

221 file_handle = handler._file_handle 

222 assert_that(file_handle).is_not_none() 

223 assert file_handle is not None 

224 assert_that(file_handle.closed).is_true() 

225 finally: 

226 Path(temp_path).unlink() 

227 

228 

229def test_json_format_writes_array_brackets() -> None: 

230 """JSON format writes array brackets.""" 

231 with tempfile.NamedTemporaryFile(delete=False, suffix=".json") as f: 

232 temp_path = f.name 

233 

234 try: 

235 handler = StreamingResultHandler( 

236 output_format="json", 

237 action=Action.CHECK, 

238 output_file=temp_path, 

239 ) 

240 with handler: 

241 pass 

242 

243 content = Path(temp_path).read_text() 

244 assert_that(content).starts_with("[") 

245 assert_that(content).ends_with("]") 

246 finally: 

247 Path(temp_path).unlink() 

248 

249 

250def test_handles_file_open_error() -> None: 

251 """Handler handles file open errors gracefully.""" 

252 handler = StreamingResultHandler( 

253 output_format="jsonl", 

254 action=Action.CHECK, 

255 output_file="/nonexistent/directory/file.json", 

256 ) 

257 with handler: 

258 assert_that(handler._file_handle).is_none() 

259 

260 

261def test_writes_jsonl_format(mock_tool_result: ToolResult) -> None: 

262 """Handler writes JSONL format. 

263 

264 Args: 

265 mock_tool_result: Fixture providing a mock ToolResult. 

266 """ 

267 with tempfile.NamedTemporaryFile(delete=False, suffix=".jsonl") as f: 

268 temp_path = f.name 

269 

270 try: 

271 handler = StreamingResultHandler( 

272 output_format="jsonl", 

273 action=Action.CHECK, 

274 output_file=temp_path, 

275 ) 

276 with handler: 

277 handler.handle_result(mock_tool_result) 

278 

279 content = Path(temp_path).read_text() 

280 lines = content.strip().split("\n") 

281 assert_that(lines).is_length(1) 

282 

283 data = json.loads(lines[0]) 

284 assert_that(data["tool"]).is_equal_to("test_tool") 

285 finally: 

286 Path(temp_path).unlink() 

287 

288 

289def test_writes_json_array_format(mock_tool_result: ToolResult) -> None: 

290 """Handler writes JSON array format. 

291 

292 Args: 

293 mock_tool_result: Fixture providing a mock ToolResult. 

294 """ 

295 with tempfile.NamedTemporaryFile(delete=False, suffix=".json") as f: 

296 temp_path = f.name 

297 

298 try: 

299 handler = StreamingResultHandler( 

300 output_format="json", 

301 action=Action.CHECK, 

302 output_file=temp_path, 

303 ) 

304 with handler: 

305 handler.handle_result(mock_tool_result) 

306 

307 content = Path(temp_path).read_text() 

308 data = json.loads(content) 

309 assert_that(data).is_instance_of(list) 

310 assert_that(data[0]["tool"]).is_equal_to("test_tool") 

311 finally: 

312 Path(temp_path).unlink() 

313 

314 

315def test_writes_multiple_json_array_results() -> None: 

316 """Handler writes multiple results in JSON array format.""" 

317 with tempfile.NamedTemporaryFile(delete=False, suffix=".json") as f: 

318 temp_path = f.name 

319 

320 try: 

321 result1 = ToolResult(name="tool1", success=True, issues_count=0) 

322 result2 = ToolResult(name="tool2", success=True, issues_count=1) 

323 

324 handler = StreamingResultHandler( 

325 output_format="json", 

326 action=Action.CHECK, 

327 output_file=temp_path, 

328 ) 

329 with handler: 

330 handler.handle_result(result1) 

331 handler.handle_result(result2) 

332 

333 content = Path(temp_path).read_text() 

334 data = json.loads(content) 

335 assert_that(data).is_instance_of(list) 

336 assert_that(data).is_length(2) 

337 assert_that(data[0]["tool"]).is_equal_to("tool1") 

338 assert_that(data[1]["tool"]).is_equal_to("tool2") 

339 finally: 

340 Path(temp_path).unlink() 

341 

342 

343def test_result_to_dict_includes_basic_fields(mock_tool_result: ToolResult) -> None: 

344 """Result dict includes basic fields. 

345 

346 Args: 

347 mock_tool_result: Fixture providing a mock ToolResult. 

348 """ 

349 handler = StreamingResultHandler(output_format="json", action=Action.CHECK) 

350 data = handler._result_to_dict(mock_tool_result) 

351 

352 assert_that(data).contains_key("tool", "success", "issues_count") 

353 assert_that(data["tool"]).is_equal_to("test_tool") 

354 

355 

356def test_result_to_dict_includes_fix_counts(mock_fix_result: ToolResult) -> None: 

357 """Result dict includes fix counts when present. 

358 

359 Args: 

360 mock_fix_result: Fixture providing a ToolResult with fix counts. 

361 """ 

362 handler = StreamingResultHandler(output_format="json", action=Action.FIX) 

363 data = handler._result_to_dict(mock_fix_result) 

364 

365 assert_that(data).contains_key("fixed_issues_count", "remaining_issues_count") 

366 assert_that(data["fixed_issues_count"]).is_equal_to(3) 

367 

368 

369def test_create_streaming_handler_with_format() -> None: 

370 """Create handler with specified format.""" 

371 handler = create_streaming_handler("json", Action.CHECK) 

372 assert_that(handler.output_format).is_equal_to("json") 

373 

374 

375def test_create_streaming_handler_with_action() -> None: 

376 """Create handler with specified action.""" 

377 handler = create_streaming_handler("grid", Action.FIX) 

378 assert_that(handler.action).is_equal_to(Action.FIX) 

379 

380 

381def test_create_streaming_handler_with_output_file(tmp_path: Path) -> None: 

382 """Create handler with output file. 

383 

384 Args: 

385 tmp_path: Temporary path fixture. 

386 """ 

387 output_file = tmp_path / "output.json" 

388 handler = create_streaming_handler("json", Action.CHECK, str(output_file)) 

389 assert_that(handler.output_file).is_equal_to(str(output_file))