Coverage for scripts / live_release_validation / context.py: 100.00%

1"""Run-context helpers: git identity, profile, and Region topology.""" 

2 

3from __future__ import annotations 

4 

5import subprocess 

6from pathlib import Path 

7 

8from .models import RunContext 

9 

10 

11def _run_git(repo_root: Path, *arguments: str, check: bool = True) -> str: 

12 result = subprocess.run( 

13 ["git", *arguments], 

14 cwd=repo_root, 

15 capture_output=True, 

16 text=True, 

17 check=False, 

18 ) 

19 if check and result.returncode != 0: 

20 message = result.stderr.strip() or result.stdout.strip() or "unknown git error" 

21 raise RuntimeError(f"git {' '.join(arguments)} failed: {message}") 

22 return result.stdout.strip() 

23 

24 

25def _resolve_branch(repo_root: Path) -> str: 

26 branch = _run_git(repo_root, "symbolic-ref", "--short", "HEAD", check=False) 

27 if branch: 

28 return branch 

29 raise RuntimeError("HEAD is detached; local live validation requires a checked-out branch") 

30 

31 

32def _validate_profile(ctx: RunContext) -> None: 

33 count = len(ctx.deployment_regions) 

34 profile = ctx.settings.profile 

35 if profile == "single-region" and count != 1: 

36 raise RuntimeError( 

37 f"single-region profile requires exactly one regional Region; cdk.json has {count}" 

38 ) 

39 if profile == "multi-region" and count < 2: 

40 raise RuntimeError( 

41 f"multi-region profile requires at least two regional Regions; cdk.json has {count}" 

42 ) 

43 if profile not in {"configured", "single-region", "multi-region"}: 

44 raise RuntimeError(f"Unknown validation profile: {profile}") 

45 

46 

47def _topology_regions(ctx: RunContext) -> tuple[str, ...]: 

48 regions = ctx.cdk_context["deployment_regions"] 

49 return tuple( 

50 dict.fromkeys( 

51 ( 

52 str(regions["global"]), 

53 str(regions["api_gateway"]), 

54 str(regions["monitoring"]), 

55 *ctx.deployment_regions, 

56 ) 

57 ) 

58 ) 

59 

60 

61def _direct_regional_access_enabled(ctx: RunContext) -> bool: 

62 partition = ctx.session.get_partition_for_region(ctx.config.global_region) 

63 if not partition: 

64 raise RuntimeError(f"Could not resolve AWS partition for {ctx.config.global_region}") 

65 configured = bool((ctx.cdk_context.get("api_gateway") or {}).get("regional_api_enabled", False)) 

66 return partition != "aws" or configured 

67 

68 

69def _job_transport_region(ctx: RunContext, execution_region: str) -> str | None: 

70 """Choose authorized transport without probing a denied regional bridge.""" 

71 if _direct_regional_access_enabled(ctx): 

72 return execution_region 

73 if len(ctx.deployment_regions) == 1 and execution_region == ctx.deployment_regions[0]: 

74 return None 

75 raise RuntimeError( 

76 "Multi-Region workload validation requires api_gateway.regional_api_enabled=true " 

77 "so each Job can be observed and deleted in its exact execution Region" 

78 ) 

79 

80 

81def _project_ecr_name(name: str, project_name: str) -> bool: 

82 return name == project_name or name.startswith((f"{project_name}/", f"{project_name}-"))