Coverage for scripts / example_job_validation / models.py: 100.00%

1"""Run settings for the example-job validation harness.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from typing import Any 

7 

8from scripts.live_release_validation.checks.eks_capabilities import ( 

9 build_eks_capabilities_overrides, 

10 overrides_json, 

11) 

12from scripts.live_release_validation.models import RunSettings 

13 

14from .specs import ( 

15 EXAMPLE_SPECS, 

16 required_capability_overrides, 

17 required_capability_settings, 

18 required_feature_overrides, 

19 required_helm_overrides, 

20) 

21 

22 

23@dataclass(frozen=True) 

24class ExampleRunSettings(RunSettings): 

25 """Operator inputs for one example-validation run. 

26 

27 Extends the live-release-validation settings with the example selection; 

28 the required helm/feature/EKS Capability enablement is DERIVED from the 

29 selection so the deployed graph always matches exactly what the selected 

30 examples need. 

31 """ 

32 

33 #: Example names (file stems) to validate this run, in registry order. 

34 selected_examples: tuple[str, ...] = () 

35 

36 #: Maximum examples running concurrently inside the examples action. 

37 #: 0 means "all selected at once". Deliberately NOT part of identity(): 

38 #: parallelism changes execution pacing, not what is validated, so a 

39 #: checkpointed run may resume with a different setting. 

40 max_parallel_examples: int = 0 

41 

42 def __post_init__(self) -> None: 

43 super().__post_init__() 

44 unknown = sorted(set(self.selected_examples) - set(EXAMPLE_SPECS)) 

45 if unknown: 

46 raise ValueError( 

47 f"Unknown example name(s): {', '.join(unknown)}. " 

48 f"Valid: {', '.join(sorted(EXAMPLE_SPECS))}" 

49 ) 

50 if self.max_parallel_examples < 0: 

51 raise ValueError("max_parallel_examples must be >= 0 (0 = no limit)") 

52 # The scheduler charts the selection needs are threaded through the 

53 # same optional_schedulers field the base class already carries so 

54 # identity(), resume checks, and extra_cdk_context() stay coherent. 

55 derived = required_helm_overrides(list(self.selected_examples)) 

56 object.__setattr__( 

57 self, "optional_schedulers", tuple(sorted({*self.optional_schedulers, *derived})) 

58 ) 

59 # EKS Capabilities travel as the base class's eks_capabilities_overrides 

60 # JSON (the same context the live harness's --eks-capabilities builds), 

61 # so extra_cdk_context() and the resume identity carry them unchanged. 

62 capability_types = self.capability_overrides 

63 if capability_types: 

64 object.__setattr__( 

65 self, 

66 "eks_capabilities_overrides_json", 

67 overrides_json( 

68 build_eks_capabilities_overrides( 

69 types=capability_types, 

70 extra=required_capability_settings(list(self.selected_examples)), 

71 ) 

72 ), 

73 ) 

74 

75 @property 

76 def capability_overrides(self) -> tuple[str, ...]: 

77 return required_capability_overrides(list(self.selected_examples)) 

78 

79 @property 

80 def feature_overrides(self) -> tuple[str, ...]: 

81 return required_feature_overrides(list(self.selected_examples)) 

82 

83 def extra_cdk_context(self) -> dict[str, str]: 

84 context = super().extra_cdk_context() 

85 if self.feature_overrides: 

86 context["feature_enabled_overrides"] = ",".join(self.feature_overrides) 

87 return context 

88 

89 def identity(self) -> dict[str, Any]: 

90 identity = super().identity() 

91 identity["selected_examples"] = list(self.selected_examples) 

92 identity["feature_overrides"] = list(self.feature_overrides) 

93 identity["capability_overrides"] = list(self.capability_overrides) 

94 return identity