Coverage for src / ocarina / dsl / testing / oc_test_cycle.py: 22.06%

1"""Full test suite orchestrator. 

2 

3Hierarchy: TestCycle → Campaigns → TestSuites → Tests 

4 

5Smoke tests run first. If any fail, main campaigns sequences are skipped. 

6Execution modes control whether smoke sequences stop on first failure or all run. 

7""" 

8 

9from typing import TYPE_CHECKING, Literal, final 

10 

11from ocarina.aggregates.tests_layers import is_test_result_fail 

12from ocarina.custom_invariants.testing.oc_test_campaigns_names import ( 

13 validate_campaigns_names, 

14) 

15from ocarina.custom_invariants.testing.oc_test_cycles_names import ( 

16 validate_test_cycle_name, 

17) 

18from ocarina.custom_invariants.testing.oc_test_runners_names import ( 

19 validate_test_runners_names, 

20) 

21from ocarina.custom_invariants.testing.oc_test_suites_names import ( 

22 validate_test_suites_names, 

23) 

24from ocarina.dsl.invariants.internals.validation_chain import chain_validations 

25from ocarina.dsl.testing.oc_test_campaign import ( 

26 TestCampaign, 

27 campaign_has_failed, 

28) 

29 

30if TYPE_CHECKING: 

31 from collections.abc import Sequence 

32 

33 from ocarina.custom_types.oc_test_layers import TestCycleResults 

34 from ocarina.custom_types.thunk import Thunk 

35 

36type Mode = Literal[ 

37 "fail-fast-on-first-smoke-campaigns-sequence-fail", 

38 "wait-for-all-smoke-tests", 

39] 

40"""Smoke test failure handling mode. 

41 

42- fail-fast: Stop smoke sequences on first failure, skip all remaining. 

43- wait-for-all: Run all smoke sequences, skip main tests if any failed. 

44""" 

45 

46 

47@final 

48class TestCycle[Driver]: 

49 """Top-level orchestrator for smoke + main test campaigns sequences. 

50 

51 Smoke sequences run first. Main sequences are skipped if any smoke test failed. 

52 Sequence names must be unique across smoke and main sequences. 

53 """ 

54 

55 def __init__( 

56 self, 

57 *, 

58 name: str, 

59 campaigns: Sequence[TestCampaign[Driver]], 

60 smoke_tests_campaigns: Sequence[TestCampaign[Driver]] | None = None, 

61 mode: Mode | None = None, 

62 ) -> None: 

63 """Initialize test suite. 

64 

65 Args: 

66 name: Cycle name used in logs and reports. 

67 campaigns: Main campaigns, run after smoke tests. 

68 smoke_tests_campaigns: Smoke campaigns, run first. Default: none. 

69 mode: Failure handling mode. Default: fail-fast. 

70 

71 Raises: 

72 AggregateInvariantViolationError: If campaign names are not unique. 

73 

74 """ 

75 if mode is None: 

76 mode = "fail-fast-on-first-smoke-campaigns-sequence-fail" 

77 

78 smoke_campaigns = smoke_tests_campaigns or [] 

79 all_campaigns = [*smoke_campaigns, *campaigns] 

80 

81 for campaign in all_campaigns: 

82 for suite in campaign._suites: # noqa: SLF001 

83 chain_validations( 

84 validate_test_cycle_name(cycle_name=name, name="test_cycle"), 

85 validate_campaigns_names( 

86 campaigns=all_campaigns, 

87 name="all campaigns (smoke + deep tests)", 

88 ), 

89 validate_test_suites_names( 

90 suites=campaign._suites, # noqa: SLF001 

91 name="suites", 

92 ), 

93 validate_test_runners_names(tests=suite._tests, name="tests"), # noqa: SLF001 

94 ).execute().raise_if_invalid() 

95 

96 self.name = name 

97 self._campaigns = campaigns 

98 self._smoke_campaigns = smoke_campaigns 

99 self._results: TestCycleResults = {} 

100 self._mode = mode 

101 

102 for campaign in self._campaigns: 

103 for suite in campaign._suites: # noqa: SLF001 

104 suite._cycle_name = name # noqa: SLF001 

105 

106 for smoke_campaign in self._smoke_campaigns: 

107 for suite in smoke_campaign._suites: # noqa: SLF001 

108 suite._cycle_name = name # noqa: SLF001 

109 

110 def run_all(self, *, saturate_workers: bool = True) -> TestCycleResults: 

111 """Run smoke sequences then main sequences. Returns aggregated results. 

112 

113 Main sequences are skipped if any smoke test failed. 

114 Smoke behavior depends on configured mode. 

115 

116 Args: 

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

118 

119 """ 

120 self._results.clear() 

121 

122 def _run_smoke(*, fail_fast: bool) -> bool: 

123 failed = False 

124 for campaign in self._smoke_campaigns: 

125 results = campaign.run_all( 

126 skip_all=failed if fail_fast else False, 

127 saturate_workers=saturate_workers, 

128 ) 

129 self._results[campaign.name] = results 

130 if campaign_has_failed(results): 

131 failed = True 

132 return failed 

133 

134 dispatch: dict[Mode, Thunk[bool]] = { 

135 "fail-fast-on-first-smoke-campaigns-sequence-fail": lambda: _run_smoke( 

136 fail_fast=True 

137 ), 

138 "wait-for-all-smoke-tests": lambda: _run_smoke(fail_fast=False), 

139 } 

140 

141 skip_all = dispatch[self._mode]() 

142 

143 for campaign in self._campaigns: 

144 self._results[campaign.name] = campaign.run_all( 

145 skip_all=skip_all, saturate_workers=saturate_workers 

146 ) 

147 

148 return self._results 

149 

150 

151def has_test_cycle_failed(results: TestCycleResults) -> bool: 

152 """Is true if any test failed.""" 

153 return any( 

154 is_test_result_fail(outcome) 

155 for campaigns in results.values() 

156 for tests in campaigns.values() 

157 for outcome, _, _ in tests.values() 

158 )