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

1"""Run settings for the upgrade validation harness.""" 

2 

3from __future__ import annotations 

4 

5import os 

6import re 

7from dataclasses import dataclass 

8from pathlib import Path 

9from typing import Any, ClassVar 

10 

11from cli.upgrade import RELEASE_TAG_RE 

12from scripts.live_release_validation.models import RunSettings 

13 

14#: CDK context that makes the regional stack retain its custom-resource 

15#: provider log groups, so the harness deletes them once the stack is gone 

16#: (``gco/stacks/regional_stack.py``). The release harness adds it to the change 

17#: sets it prepares; here it lives in the base checkout's cdk.json. 

18PROVIDER_LOG_CONTEXT = "gco_live_validation_retain_provider_log_groups" 

19 

20_COMMIT_RE = re.compile(r"[0-9a-f]{40}") 

21 

22 

23def _real(path: Path) -> Path: 

24 return Path(os.path.realpath(os.fspath(path))) 

25 

26 

27def _overlaps(first: Path, second: Path) -> bool: 

28 return first == second or first in second.parents or second in first.parents 

29 

30 

31@dataclass(frozen=True) 

32class UpgradeRunSettings(RunSettings): 

33 """Operator inputs for one upgrade-validation run. 

34 

35 Extends the live-release-validation settings with the release the run 

36 deploys first (``base_ref``, pinned to ``base_commit`` so a moved tag cannot 

37 change what a resumed run deploys) and the private workspace that holds 

38 that release's clone, its git mirror, and its virtual environment. The 

39 workspace must not overlap the checkout (the preflight's clean-tree rule 

40 and the CDK asset staging both read it) or the report directory (which 

41 may hold only report files, and outlives the workspace). 

42 """ 

43 

44 allows_run_tag_adoption: ClassVar[bool] = True 

45 

46 #: The release tag deployed first and upgraded from (``vMAJOR.MINOR.PATCH``). 

47 base_ref: str = "" 

48 #: The commit ``base_ref`` named when the run started. 

49 base_commit: str = "" 

50 #: Private directory for the base clone, its mirror, its venv, and logs. 

51 workspace_dir: Path = Path() 

52 #: Wall-clock caps for the two long ``gco`` subprocesses. 

53 deploy_timeout_seconds: int = 3 * 60 * 60 

54 upgrade_timeout_seconds: int = 4 * 60 * 60 

55 #: How long the sentinel calls wait for the API to answer. 

56 api_ready_timeout_seconds: int = 15 * 60 

57 

58 def __post_init__(self) -> None: 

59 super().__post_init__() 

60 if not RELEASE_TAG_RE.fullmatch(self.base_ref): 

61 raise ValueError(f"base_ref must be a release tag vX.Y.Z, not {self.base_ref!r}") 

62 if not _COMMIT_RE.fullmatch(self.base_commit): 

63 raise ValueError("base_commit must be a lowercase 40-character commit SHA") 

64 if self.base_commit == self.expected_sha.lower(): 

65 raise ValueError( 

66 f"{self.base_ref} is the checked-out commit, so there is nothing to upgrade" 

67 ) 

68 workspace = Path(os.path.abspath(os.fspath(self.workspace_dir))) 

69 object.__setattr__(self, "workspace_dir", workspace) 

70 for label, directory in ( 

71 ("the checkout", self.repo_root), 

72 ("the report directory", self.report_dir), 

73 ): 

74 if _overlaps(_real(workspace), _real(directory)): 

75 raise ValueError(f"The workspace {workspace} must not overlap {label}") 

76 for name in ( 

77 "deploy_timeout_seconds", 

78 "upgrade_timeout_seconds", 

79 "api_ready_timeout_seconds", 

80 ): 

81 if getattr(self, name) <= 0: 

82 raise ValueError(f"{name} must be positive") 

83 

84 def base_cdk_context(self) -> dict[str, str]: 

85 """Run-scoped context the base checkout's cdk.json carries. 

86 

87 ``gco`` in the base checkout deploys, and ``gco upgrade`` redeploys, 

88 without the harness's ``--context`` flags, so the context has to live 

89 in the one file the upgrade preserves byte for byte. 

90 """ 

91 return {**self.extra_cdk_context(), PROVIDER_LOG_CONTEXT: "true"} 

92 

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

94 identity = super().identity() 

95 identity.update( 

96 { 

97 "base_ref": self.base_ref, 

98 "base_commit": self.base_commit, 

99 "workspace_dir": str(self.workspace_dir), 

100 "base_cdk_context": self.base_cdk_context(), 

101 "stack_ownership": "run-tag-adoption", 

102 } 

103 ) 

104 return identity