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

1"""api and sqs: authenticated API and direct regional SQS Job lifecycles.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from ..checks.jobs import ( 

8 _complete_job_lifecycle, 

9 _get_owned_job, 

10 _job_appearance_timeout, 

11 _load_manifest, 

12 _register_job, 

13 _run_api_transport_lifecycle, 

14 _run_token, 

15 _wait_for_ambiguous_job_reconciliation, 

16 _wait_for_owned_job_appearance, 

17) 

18from ..constants import ( 

19 _RUN_JOB_LABEL, 

20) 

21from ..models import RunContext 

22 

23 

24def action_api_lifecycle(ctx: RunContext) -> dict[str, Any]: 

25 """Submit, observe, read logs, and delete an authenticated API Job. 

26 

27 The full crash-safe lifecycle lives in 

28 :func:`..checks.jobs._run_api_transport_lifecycle`, shared with the 

29 scheduler probes so the transport dance cannot drift between them. 

30 """ 

31 return _run_api_transport_lifecycle( 

32 ctx, 

33 manifest_filename="api-smoke-job.yaml", 

34 path="api", 

35 marker_prefix="API", 

36 ) 

37 

38 

39def action_sqs_lifecycle(ctx: RunContext) -> dict[str, Any]: 

40 """Submit, observe, read logs, and delete a direct regional-SQS Job.""" 

41 manifests, name, namespace = _load_manifest(ctx, "sqs-smoke-job.yaml") 

42 token = _run_token(ctx.settings.run_id) 

43 marker = f"GCO_LIVE_SQS_{token}" 

44 region = ctx.deployment_regions[0] 

45 record = _register_job( 

46 ctx, 

47 name=name, 

48 namespace=namespace, 

49 execution_region=region, 

50 path="sqs", 

51 ) 

52 envelope = { 

53 "transport": "direct-sqs", 

54 "manifests": manifests, 

55 "region": region, 

56 "namespace": namespace, 

57 "labels": {_RUN_JOB_LABEL: token}, 

58 "priority": 100, 

59 } 

60 ctx.prepare_job_submission(record, envelope=envelope, resumable=False) 

61 

62 existing = _get_owned_job(ctx, record) 

63 state = str(record.get("submission_state") or "") 

64 if existing is None and state == "submitting": 

65 existing = _wait_for_ambiguous_job_reconciliation(ctx, record) 

66 if existing is None: 

67 reason = ( 

68 "Direct SQS submission crossed a non-idempotent boundary but no Job appeared; " 

69 "automatic replay is forbidden" 

70 ) 

71 ctx.block_job_submission(record, reason) 

72 raise RuntimeError(reason) 

73 elif existing is None and state == "submitted": 

74 existing = _wait_for_owned_job_appearance(ctx, record) 

75 elif existing is None and state == "blocked": 

76 raise RuntimeError(str(record.get("submission_blocked_reason") or "SQS submission blocked")) 

77 

78 if existing is None: 

79 if state != "prepared": 

80 raise RuntimeError(f"Cannot submit SQS Job from state {state!r}") 

81 ctx.begin_job_submission( 

82 record, 

83 reconciliation_timeout_seconds=_job_appearance_timeout(ctx), 

84 ) 

85 submission = ctx.job_manager.submit_job_sqs( 

86 manifests, 

87 region=region, 

88 namespace=namespace, 

89 labels={_RUN_JOB_LABEL: token}, 

90 priority=100, 

91 ) 

92 if submission.get("job_name") != name: 

93 raise RuntimeError( 

94 f"SQS submission returned unexpected job name: {submission.get('job_name')}" 

95 ) 

96 ctx.finish_job_submission( 

97 record, 

98 submission, 

99 appearance_timeout_seconds=_job_appearance_timeout(ctx), 

100 ) 

101 ctx.checkpoint.state["sqs_submission"] = submission 

102 ctx.persist() 

103 else: 

104 submission = {"reconciled_existing_job": True} 

105 

106 lifecycle = _complete_job_lifecycle(ctx, record=record, marker=marker) 

107 lifecycle["submission"] = submission 

108 return lifecycle