Coverage for scripts / live_release_validation / actions / policy.py: 100.00%

1"""policy: prove the deployed job-validation policy is fully readable. 

2 

3Read-only. Adds no owned resources and mutates nothing, so it needs no cleanup 

4and can run alongside the lifecycle actions. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import Any 

10 

11from ..checks.policy import _validate_region_policy 

12from ..models import RunContext 

13 

14 

15def action_policy(ctx: RunContext) -> dict[str, Any]: 

16 """Require every Region to report all three admission layers. 

17 

18 ``GET /api/v1/policy`` answers "will this cluster admit the job I am about 

19 to pay to run" in three layers: the front-door per-manifest caps, the 

20 per-container ``LimitRange``, and the namespace aggregate ``ResourceQuota``. 

21 

22 The cluster-read layers are fail-soft by design, and the degraded response 

23 is an HTTP 200 carrying a per-namespace ``status`` field. Every 

24 transport-level check in this harness therefore passes while the endpoint 

25 reports nothing usable -- which is exactly what happened on 2026-08-26, when 

26 all ten actions were green and ``cluster_enforcement."gco-jobs"`` was 

27 ``{"status": "unavailable", "reason": "403 Forbidden"}``. So this action 

28 asserts on the body, per Region and per namespace. 

29 

30 It also requires the project's own ECR hostnames to be present in 

31 ``trusted_registries``, since CDK appends them at synth time and their 

32 absence would reject every job pulling a project-built image while no 

33 offline check could predict it. 

34 """ 

35 regions: dict[str, Any] = {} 

36 evidence: dict[str, Any] = {"regions": regions} 

37 try: 

38 for region in ctx.deployment_regions: 

39 regions[region] = _validate_region_policy(ctx, region) 

40 except BaseException as exc: 

41 evidence["result"] = "failed" 

42 evidence["error"] = f"{type(exc).__name__}: {exc}" 

43 with ctx.state_lock: 

44 ctx.checkpoint.state["policy"] = evidence 

45 ctx.persist_callback(ctx.checkpoint) 

46 raise 

47 

48 evidence["result"] = "passed" 

49 with ctx.state_lock: 

50 ctx.checkpoint.state["policy"] = evidence 

51 ctx.persist_callback(ctx.checkpoint) 

52 return evidence