Coverage for src / ocarina / opinionated / launcher / bootstrap.py: 0.00%

1"""Generic bootstrap for running a test cycle, post-run plugins, and reporting.""" 

2 

3from concurrent.futures.thread import ThreadPoolExecutor 

4from typing import TYPE_CHECKING 

5 

6if TYPE_CHECKING: 

7 from collections.abc import Callable 

8 

9 from ocarina.custom_types.effect import Effect 

10 from ocarina.custom_types.oc_test_layers import TestCycleResults 

11 from ocarina.dsl.testing.oc_test_cycle import TestCycle 

12 from ocarina.ports.ilogger import ILogger 

13 

14 

15def run_plugins(*plugins: Effect, exceptions_logger: ILogger) -> None: 

16 """Run plugins sequentially (one plugin) or in parallel (multiple plugins). 

17 

18 Exceptions are caught and logged via `exceptions_logger` so that a failing 

19 plugin never interrupts the others. 

20 

21 Args: 

22 *plugins: Side-effectful callables to execute. 

23 exceptions_logger: Logger used to report plugin failures. 

24 

25 """ 

26 if not plugins: # pragma: no cover 

27 return 

28 

29 def _run_plugin(plugin: Effect, exceptions_logger: ILogger) -> None: 

30 try: 

31 plugin() 

32 except Exception as exc: 

33 exceptions_logger.exception("The plugin failed.", exc=exc) 

34 

35 if len(plugins) == 1: 

36 _run_plugin(plugins[0], exceptions_logger) 

37 return 

38 

39 with ThreadPoolExecutor(max_workers=len(plugins)) as executor: 

40 futures = [ 

41 executor.submit(_run_plugin, plugin, exceptions_logger) 

42 for plugin in plugins 

43 ] 

44 for f in futures: 

45 f.result() 

46 

47 

48def bootstrap[T]( 

49 *, 

50 test_cycle: TestCycle[T], 

51 run_plugins: Callable[[TestCycleResults], None], 

52 post_exec: Callable[[TestCycleResults], None] | None = None, 

53 saturate_workers: bool = True, 

54) -> None: 

55 """Run all tests, execute post-run plugins, and optionally print the report. 

56 

57 Args: 

58 test_cycle: The test cycle to run. 

59 run_plugins: Called with the results once the cycle completes. Receives 

60 ``results`` as its only argument, which allows plugins that depend on 

61 the results (e.g. JSON export) to be declared at the call-site without 

62 forward-referencing an not-yet-existing variable. 

63 post_exec: Post execution callback, taking results. 

64 saturate_workers: Duplicate tests to reach max_workers. Default: True. 

65 

66 Example: 

67 bootstrap( 

68 post_exec=lambda results: pretty_print_results(results), 

69 test_cycle=TestCycle(campaigns=[...]), 

70 run_plugins=lambda results: run_plugins( 

71 lambda: generate_docx_proof( 

72 logs_root=get_default_log_dir(), 

73 logger=logger, 

74 output_root=Path.cwd() / "tests_docx_output", 

75 ), 

76 lambda: generate_json_results( 

77 results=results, 

78 output_dir=Path.cwd() / "tests_json_output", 

79 logger=logger, 

80 ), 

81 exceptions_logger=PrintLogger() 

82 .set_prefix( 

83 lambda: concat_metadata( 

84 format_utc_date_metadata_str, 

85 format_current_thread_metadata_str, 

86 ) 

87 ) 

88 .set_domain_taxonomy(("Post-execution plugins",)), 

89 ), 

90 ) 

91 

92 """ 

93 results = test_cycle.run_all(saturate_workers=saturate_workers) 

94 run_plugins(results) 

95 if post_exec: 

96 post_exec(results)