Coverage for /home/crpier/Projects/snektest/snektest/cli.py: 42%

1import asyncio 

2import json 

3import sys 

4import threading 

5import traceback 

6from collections.abc import Callable, Coroutine 

7from dataclasses import dataclass, field 

8from typing import Literal, cast 

9 

10from snektest.agent_docs import ( 

11 get_agent_docs, 

12 get_example_source, 

13 get_examples_listing, 

14) 

15from snektest.collection import TestsQueue, load_tests_from_filters 

16from snektest.execution import run_tests 

17from snektest.models import ( 

18 ArgsError, 

19 BadRequestError, 

20 CollectionError, 

21 ErrorResult, 

22 FailedResult, 

23 FilterItem, 

24 PassedResult, 

25 TeardownFailure, 

26 TestResult, 

27 UnreachableError, 

28) 

29from snektest.presenter import print_error 

30 

31 

32def _json_result_status(result: TestResult) -> str: 

33 match result.result: 

34 case PassedResult(): 

35 return "passed" 

36 case FailedResult(): 

37 return "failed" 

38 case ErrorResult(): 

39 return "error" 

40 

41 

42def build_json_summary(summary: TestRunSummary) -> dict[str, object]: 

43 return { 

44 "passed": summary.passed, 

45 "failed": summary.failed, 

46 "errors": summary.errors, 

47 "fixture_teardown_failed": summary.fixture_teardown_failed, 

48 "session_teardown_failed": summary.session_teardown_failed, 

49 "tests": [ 

50 { 

51 "name": str(result.name), 

52 "duration": result.duration, 

53 "markers": list(result.markers), 

54 "status": _json_result_status(result), 

55 } 

56 for result in summary.test_results 

57 ], 

58 } 

59 

60 

61@dataclass 

62class TestRunSummary: 

63 """Summary of test run results.""" 

64 

65 total_tests: int 

66 passed: int 

67 failed: int 

68 errors: int 

69 fixture_teardown_failed: int 

70 session_teardown_failed: int 

71 test_results: list[TestResult] 

72 session_teardown_failures: list[TeardownFailure] 

73 

74 

75type CliAction = Literal["agent_docs", "help", "list_examples", "show_example"] 

76 

77 

78@dataclass(frozen=True) 

79class CliOptions: 

80 action: CliAction | None = None 

81 capture_output: bool = True 

82 example_name: str | None = None 

83 json_output: bool = False 

84 pdb_on_failure: bool = False 

85 mark: str | None = None 

86 

87 

88@dataclass 

89class CliParseState: 

90 action: CliAction | None = None 

91 capture_output: bool = True 

92 example_name: str | None = None 

93 json_output: bool = False 

94 mark: str | None = None 

95 pdb_on_failure: bool = False 

96 potential_filter: list[str] = field(default_factory=list[str]) 

97 

98 

99PARSE_ERROR = -1 

100VALID_MARKER_VALUES = {"fast", "medium", "slow"} 

101 

102HELP_TEXT = """Usage: snektest [OPTIONS] [FILTER ...] 

103 

104Run snektest tests. 

105 

106Filters: 

107 . Run tests below current directory 

108 tests/test_file.py Run one test file 

109 tests/test_file.py::test_name Run one test function 

110 tests/test_file.py::test_name[param] Run one parameterized case 

111 

112Options: 

113 -h, --help Show this help message 

114 -s Disable stdout/stderr capture 

115 --agent-docs Print AI-agent usage guide 

116 --llms Alias for --agent-docs 

117 --examples List bundled examples 

118 --example NAME Print a bundled example 

119 --json-output Print machine-readable JSON summary 

120 --mark MARK Run tests marked fast, medium, or slow; marking tests is recommended 

121 --pdb Drop into post-mortem debugger on first failure 

122 

123Example commands: 

124 snektest --agent-docs 

125 snektest --examples 

126 snektest --example async 

127 snektest examples 

128 snektest example async 

129 python -m snektest --agent-docs 

130""" 

131 

132 

133def _is_valid_mark_value(mark_value: str) -> bool: 

134 return mark_value in VALID_MARKER_VALUES 

135 

136 

137def _invalid_mark_message(mark_value: str) -> str: 

138 allowed = ", ".join(sorted(VALID_MARKER_VALUES)) 

139 return f"Invalid --mark value: `{mark_value}`. Use one of: {allowed}" 

140 

141 

142def _parse_mark_value( 

143 argv: list[str], index: int, mark: str | None 

144) -> tuple[str, int] | None: 

145 if mark is not None: 

146 print_error("Only one --mark value is supported") 

147 return None 

148 if index + 1 >= len(argv): 

149 print_error("Missing value for --mark") 

150 return None 

151 mark_value = argv[index + 1] 

152 if mark_value.startswith("-"): 

153 print_error("Missing value for --mark") 

154 return None 

155 if not _is_valid_mark_value(mark_value): 

156 print_error(_invalid_mark_message(mark_value)) 

157 return None 

158 return mark_value, index + 1 

159 

160 

161def _parse_example_value(argv: list[str], index: int) -> tuple[str, int] | None: 

162 if index + 1 >= len(argv): 

163 print_error("Missing value for --example") 

164 return None 

165 example_name = argv[index + 1] 

166 if example_name.startswith("-"): 

167 print_error("Missing value for --example") 

168 return None 

169 return example_name, index + 1 

170 

171 

172def _set_cli_action(state: CliParseState, new_action: CliAction) -> bool: 

173 if state.action is not None: 

174 print_error("Only one help/docs/examples command is supported") 

175 return False 

176 state.action = new_action 

177 return True 

178 

179 

180def _parse_example_action(argv: list[str], index: int, state: CliParseState) -> int: 

181 next_index = PARSE_ERROR 

182 if _set_cli_action(state, "show_example"): 

183 parsed_example = _parse_example_value(argv, index) 

184 if parsed_example is not None: 

185 state.example_name, next_index = parsed_example 

186 return next_index 

187 

188 

189def _parse_action_option(command: str, state: CliParseState) -> int: 

190 actions: dict[str, CliAction] = { 

191 "--agent-docs": "agent_docs", 

192 "--examples": "list_examples", 

193 "--help": "help", 

194 "--llms": "agent_docs", 

195 "-h": "help", 

196 } 

197 next_index = PARSE_ERROR 

198 action = actions.get(command) 

199 if action is not None and _set_cli_action(state, action): 

200 next_index = 0 

201 return next_index 

202 

203 

204def _parse_option(argv: list[str], index: int, state: CliParseState) -> int: 

205 command = argv[index] 

206 next_index = index 

207 if command in {"-h", "--agent-docs", "--examples", "--help", "--llms"}: 

208 parsed_index = _parse_action_option(command, state) 

209 next_index = index if parsed_index != PARSE_ERROR else PARSE_ERROR 

210 elif command == "--example": 

211 next_index = _parse_example_action(argv, index, state) 

212 elif command == "-s": 

213 state.capture_output = False 

214 elif command == "--json-output": 

215 state.json_output = True 

216 elif command == "--pdb": 

217 state.pdb_on_failure = True 

218 elif command == "--mark": 

219 parsed_mark = _parse_mark_value(argv, index, state.mark) 

220 if parsed_mark is None: 

221 next_index = PARSE_ERROR 

222 else: 

223 state.mark, next_index = parsed_mark 

224 else: 

225 print_error(f"Invalid option: `{command}`") 

226 next_index = PARSE_ERROR 

227 return next_index 

228 

229 

230def _parse_positional(argv: list[str], index: int, state: CliParseState) -> int: 

231 command = argv[index] 

232 next_index = index 

233 if command == "examples": 

234 if not _set_cli_action(state, "list_examples"): 

235 next_index = PARSE_ERROR 

236 elif command == "example": 

237 next_index = _parse_example_action(argv, index, state) 

238 else: 

239 state.potential_filter.append(command) 

240 return next_index 

241 

242 

243def _print_cli_action(options: CliOptions) -> int: 

244 output = "" 

245 if options.action == "help": 

246 output = HELP_TEXT 

247 elif options.action == "agent_docs": 

248 output = get_agent_docs() 

249 elif options.action == "list_examples": 

250 output = get_examples_listing() 

251 elif options.action == "show_example": 

252 if options.example_name is None: 

253 print_error("Missing example name") 

254 return 2 

255 try: 

256 output = get_example_source(options.example_name) 

257 except BadRequestError as e: 

258 print_error(str(e)) 

259 return 2 

260 print(output, end="") 

261 return 0 

262 

263 

264def _finish_parse_state(state: CliParseState) -> tuple[list[str], CliOptions] | int: 

265 potential_filter = state.potential_filter 

266 if state.action is not None and potential_filter: 

267 print_error("Cannot combine help/docs/examples commands with test filters") 

268 return 2 

269 

270 if state.action is None and not potential_filter: 

271 potential_filter.append(".") 

272 

273 options = CliOptions( 

274 action=state.action, 

275 capture_output=state.capture_output, 

276 example_name=state.example_name, 

277 json_output=state.json_output, 

278 pdb_on_failure=state.pdb_on_failure, 

279 mark=state.mark, 

280 ) 

281 return potential_filter, options 

282 

283 

284def parse_cli_args(argv: list[str]) -> tuple[list[str], CliOptions] | int: 

285 state = CliParseState() 

286 

287 index = 0 

288 while index < len(argv): 

289 command = argv[index] 

290 if command.startswith("-"): 

291 index = _parse_option(argv, index, state) 

292 else: 

293 index = _parse_positional(argv, index, state) 

294 if index == PARSE_ERROR: 

295 return 2 

296 index += 1 

297 

298 return _finish_parse_state(state) 

299 

300 

301async def _run_tests_with_producer_thread( 

302 filter_items: list[FilterItem], 

303 *, 

304 capture_output: bool, 

305 pdb_on_failure: bool, 

306 mark: str | None = None, 

307) -> tuple[list[TestResult], list[TeardownFailure]]: 

308 queue = TestsQueue() 

309 collection_exception: list[BaseException] = [] 

310 

311 producer_thread = threading.Thread( 

312 target=load_tests_from_filters, 

313 kwargs={ 

314 "filter_items": filter_items, 

315 "queue": queue, 

316 "loop": asyncio.get_running_loop(), 

317 "mark": mark, 

318 "exception_holder": collection_exception, 

319 }, 

320 ) 

321 

322 producer_thread.start() 

323 

324 try: 

325 test_results, session_teardown_failures = await run_tests( 

326 queue=queue, 

327 capture_output=capture_output, 

328 pdb_on_failure=pdb_on_failure, 

329 collection_failed=lambda: bool(collection_exception), 

330 ) 

331 finally: 

332 producer_thread.join() 

333 if collection_exception: 

334 raise collection_exception[0] 

335 

336 return test_results, session_teardown_failures 

337 

338 

339def exit_code_from_summary(summary: TestRunSummary) -> int: 

340 has_failures = ( 

341 summary.failed > 0 

342 or summary.errors > 0 

343 or summary.fixture_teardown_failed > 0 

344 or summary.session_teardown_failed > 0 

345 ) 

346 return 1 if has_failures else 0 

347 

348 

349async def run_tests_programmatic( 

350 filter_items: list[FilterItem], 

351 *, 

352 capture_output: bool = True, 

353 pdb_on_failure: bool = False, 

354 mark: str | None = None, 

355) -> TestRunSummary: 

356 """Run tests and return structured results instead of printing. 

357 

358 This is the programmatic API for testing snektest itself. 

359 Returns structured data instead of just printing and exiting. 

360 

361 Args: 

362 filter_items: List of filter items to run tests from 

363 capture_output: Whether to capture test output 

364 

365 Returns: 

366 TestRunSummary with test results and counts 

367 """ 

368 if mark is not None and not _is_valid_mark_value(mark): 

369 raise BadRequestError(_invalid_mark_message(mark)) 

370 

371 test_results, session_teardown_failures = await _run_tests_with_producer_thread( 

372 filter_items, 

373 capture_output=capture_output, 

374 pdb_on_failure=pdb_on_failure, 

375 mark=mark, 

376 ) 

377 

378 return TestRunSummary( 

379 total_tests=len(test_results), 

380 passed=sum(1 for r in test_results if isinstance(r.result, PassedResult)), 

381 failed=sum(1 for r in test_results if isinstance(r.result, FailedResult)), 

382 errors=sum(1 for r in test_results if isinstance(r.result, ErrorResult)), 

383 fixture_teardown_failed=sum( 

384 1 for r in test_results if r.fixture_teardown_failures 

385 ), 

386 session_teardown_failed=len(session_teardown_failures), 

387 test_results=test_results, 

388 session_teardown_failures=session_teardown_failures, 

389 ) 

390 

391 

392async def run_script( 

393 argv: list[str] | None = None, 

394 *, 

395 run_tests_programmatic_fn: Callable[..., Coroutine[object, object, object]] 

396 | None = None, 

397) -> int: 

398 """Parse arguments and run tests.""" 

399 parsed = parse_cli_args(sys.argv[1:] if argv is None else argv) 

400 if isinstance(parsed, int): 

401 return parsed 

402 

403 potential_filter, options = parsed 

404 

405 if options.action is not None: 

406 return _print_cli_action(options) 

407 

408 try: 

409 filter_items = [FilterItem(item) for item in potential_filter] 

410 except ArgsError as e: 

411 print_error(str(e)) 

412 return 2 

413 

414 runner = run_tests_programmatic_fn or run_tests_programmatic 

415 try: 

416 summary = cast( 

417 "TestRunSummary", 

418 await runner( 

419 filter_items, 

420 capture_output=options.capture_output, 

421 pdb_on_failure=options.pdb_on_failure, 

422 mark=options.mark, 

423 ), 

424 ) 

425 except asyncio.CancelledError: 

426 return 2 

427 

428 if options.json_output: 

429 print(json.dumps(build_json_summary(summary))) 

430 

431 return exit_code_from_summary(summary) 

432 

433 

434def main() -> None: 

435 """Main entry point for the CLI.""" 

436 async_runner = cast("Callable[[object], int]", asyncio.run) 

437 sys.exit(main_inner(async_runner=async_runner)) 

438 

439 

440def main_inner( 

441 *, 

442 async_runner: Callable[[object], int], 

443 argv: list[str] | None = None, 

444) -> int: 

445 try: 

446 coroutine = run_script(argv) 

447 return async_runner(coroutine) 

448 except CollectionError as e: 

449 formatted = "".join(traceback.format_exception(e)).rstrip() 

450 print_error(f"Collection error:\n{formatted}") 

451 return 2 

452 except BadRequestError as e: 

453 print_error(f"Bad request error: {e}") 

454 return 2 

455 except UnreachableError as e: 

456 print_error(f"Internal error: {e}") 

457 return 2 

458 except KeyboardInterrupt: 

459 print_error("Interrupted by user") 

460 return 2 

461 except Exception as e: 

462 print_error(f"Unexpected error: {e}") 

463 return 1 

464 

465 

466if __name__ == "__main__": 

467 main()