Coverage for src / ocarina / dsl / testing / oc_test_campaign.py: 21.21%

1"""Campaigns: orchestrates multiple TestSuites instances with shared config. 

2 

3Test hierarchy: 

4 TestCycle → Campaigns → TestSuites → Tests 

5""" 

6 

7from typing import TYPE_CHECKING 

8 

9from ocarina.aggregates.tests_layers import is_test_result_fail 

10from ocarina.custom_invariants.testing.oc_test_suites_names import ( 

11 validate_test_suites_names, 

12) 

13 

14if TYPE_CHECKING: 

15 from collections.abc import Sequence 

16 

17 from ocarina.custom_types.oc_test_layers import TestCampaignResults 

18 from ocarina.dsl.testing.oc_test_suite import TestSuite 

19 

20 

21class TestCampaign[Driver]: 

22 """Orchestrator for a sequence of related suites with shared worker config. 

23 

24 Campaigns run sequentially; tests within each suite run in parallel. 

25 Injects campaign name into each suite for logging taxonomy. 

26 """ 

27 

28 def __init__( 

29 self, 

30 *, 

31 name: str, 

32 suites: Sequence[TestSuite[Driver]], 

33 max_workers: int, 

34 saturate_workers: bool | None = None, 

35 ) -> None: 

36 """Initialize the sequence. 

37 

38 Args: 

39 name: Campaign name used in logs and reports. 

40 suites: Suites to execute. Must have unique names. 

41 max_workers: Concurrent workers applied to all campaigns. 

42 saturate_workers: Duplicate tests to reach max_workers. Default: None. 

43 

44 Raises: 

45 AggregateInvariantViolationError: If suite names are not unique. 

46 

47 """ 

48 validate_test_suites_names( 

49 suites=suites, name="suites" 

50 ).execute().raise_if_invalid() 

51 

52 self.name = name 

53 self._suites = suites 

54 self._results: TestCampaignResults = {} 

55 self._max_workers = max_workers 

56 self._saturate_workers = saturate_workers 

57 

58 for suite in self._suites: 

59 suite._campaign_name = self.name # noqa: SLF001 

60 

61 def run_all( 

62 self, *, skip_all: bool = False, saturate_workers: bool = True 

63 ) -> TestCampaignResults: 

64 """Execute all suites sequentially. 

65 

66 Args: 

67 skip_all: If True, mark all tests as skipped without executing. 

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

69 

70 Returns: 

71 Dict mapping suite name → campaign results. 

72 

73 """ 

74 self._results.clear() 

75 

76 resolved_saturate_workers = ( 

77 self._saturate_workers 

78 if self._saturate_workers is not None 

79 else saturate_workers 

80 ) 

81 

82 if skip_all: 

83 for suite in self._suites: 

84 self._results[suite.name] = { 

85 name: (None, -1, test_id) 

86 for name, test_id in suite.test_names_and_ids 

87 } 

88 return self._results 

89 

90 for suite in self._suites: 

91 self._results[suite.name] = suite.run( 

92 max_workers=self._max_workers, 

93 saturate_workers=resolved_saturate_workers, 

94 ) 

95 

96 return self._results 

97 

98 

99def campaign_has_failed(results: TestCampaignResults) -> bool: 

100 """Return True if any test in the campaign results has a Fail outcome. 

101 

102 Skipped (None) and passed (Ok) tests return False. 

103 """ 

104 return any( 

105 is_test_result_fail(outcome) 

106 for campaign_results in results.values() 

107 for outcome, _, _ in campaign_results.values() 

108 )