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.models import RunSettings 

9 

10from .specs import EXAMPLE_SPECS, required_feature_overrides, required_helm_overrides 

11 

12 

13@dataclass(frozen=True) 

14class ExampleRunSettings(RunSettings): 

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

16 

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

18 the required helm/feature enablement is DERIVED from the selection so the 

19 deployed graph always matches exactly what the selected examples need. 

20 """ 

21 

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

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

24 

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

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

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

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

29 max_parallel_examples: int = 0 

30 

31 def __post_init__(self) -> None: 

32 super().__post_init__() 

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

34 if unknown: 

35 raise ValueError( 

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

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

38 ) 

39 if self.max_parallel_examples < 0: 

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

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

42 # same optional_schedulers field the base class already carries so 

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

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

45 object.__setattr__( 

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

47 ) 

48 

49 @property 

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

51 return required_feature_overrides(list(self.selected_examples)) 

52 

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

54 context = super().extra_cdk_context() 

55 if self.feature_overrides: 

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

57 return context 

58 

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

60 identity = super().identity() 

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

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

63 return identity