(ELI-466) pulling out shared code

Author: TOEL2

scripts/workflow/ci_utils.py

@@ -0,0 +1,97 @@
+"""
+CI shared utilities:
+- light wrappers around subprocess and GitHub CLI
+- git/gh helpers used by both resolver scripts
+- consistent failure handling and environment access
+"""
+
+from __future__ import annotations
+import json
+import os
+import subprocess
+import sys
+from typing import List, Optional, NoReturn
+
+BRANCH = os.getenv("BRANCH", "main")
+
+def fail(msg: str) -> NoReturn:
+    print(f"::error::{msg}", file=sys.stderr)
+    sys.exit(1)
+
+def run(cmd: List[str], *, check: bool = True) -> subprocess.CompletedProcess:
+    # cp = completed process (will use this to refer)
+    return subprocess.run(cmd, check=check, capture_output=True, text=True)
+
+def ensure_token() -> None:
+    if not (os.getenv("GH_TOKEN") or os.getenv("GITHUB_TOKEN")):
+        fail("GH_TOKEN/GITHUB_TOKEN is required")
+
+def fetch_latest_from_remote() -> None:
+    run(["git", "fetch", "origin", BRANCH, "--quiet"])
+    run(["git", "fetch", "--tags", "--force", "--quiet"])
+
+def git_ok(args: List[str]) -> bool:
+    return subprocess.run(["git", *args]).returncode == 0
+
+def is_ancestor(older: str, newer: str) -> bool:
+    return subprocess.run(["git", "merge-base", "--is-ancestor", older, newer]).returncode == 0
+
+def gh_json(args: List[str]) -> list:
+    cp = run(["gh", *args])
+    raw = cp.stdout.strip()
+    return json.loads(raw) if raw else []
+
+def gh_api(path: str, jq: Optional[str] = None) -> List[str]:
+    """
+    A simple python wrapper around the GitHub API
+    to make it a callable function.
+    """
+    args = ["gh", "api", path]
+    if jq:
+        args += ["--jq", jq]
+    cp = run(args, check=True)
+    return [x for x in cp.stdout.splitlines() if x]
+
+def dev_tag_for_sha(sha: str) -> Optional[str]:
+    cp = run(["git", "tag", "--points-at", sha], check=False)
+    for t in cp.stdout.splitlines():
+        if t.startswith("dev-"):
+            return t
+    return None
+
+def sha_for_tag(tag: str) -> Optional[str]:
+    cp = run(["git", "rev-list", "-n1", tag], check=False)
+    return cp.stdout.strip() or None
+
+def latest_final_tag() -> Optional[str]:
+    cp = run(["git", "tag", "--list", "v[0-9]*.[0-9]*.[0-9]*", "--sort=-v:refname"], check=True)
+    tags = cp.stdout.splitlines()
+    return tags[0] if tags else None
+
+def first_commit() -> str:
+    """
+    Returns the first commit of the current branch.
+
+    We will never use this for our project since we
+    already have a release but can be used as a
+    fallback for new projects.
+    """
+    return run(["git", "rev-list", "--max-parents=0", "HEAD"], check=True).stdout.strip()
+
+def list_merged_pr_commits(base: str, head: str) -> List[str]:
+    rng = f"{base}..{head}"
+    cp = run(["git", "rev-list", "--merges", "--first-parent", rng], check=False)
+    return [x for x in cp.stdout.splitlines() if x]
+
+def prs_for_commit(sha: str) -> List[int]:
+    nums = gh_api(
+        f"/repos/{os.getenv('GITHUB_REPOSITORY')}/commits/{sha}/pulls",
+        jq=".[].number",
+    )
+    return [int(n) for n in nums]
+
+def labels_for_pr(pr: int) -> List[str]:
+    return gh_api(
+        f"/repos/{os.getenv('GITHUB_REPOSITORY')}/issues/{pr}/labels",
+        jq=".[].name",
+    )

scripts/workflow/pre-release_resolver.py

@@ -11,74 +11,37 @@
 """
 
 from __future__ import annotations
-import json
 import os
 import re
-import subprocess
 import sys
 from dataclasses import dataclass
 from typing import List, Optional, NoReturn
 
-WORKFLOW_NAME = os.getenv("WORKFLOW_NAME", "3. CD | Deploy to Test")
-BRANCH = os.getenv("BRANCH", "main")
-LIMIT = int(os.getenv("LIMIT", "30")) #how many previous successful test deployed commits to consider
+from ci_utils import (
+    BRANCH,
+    fail,
+    ensure_token,
+    fetch_latest_from_remote,
+    gh_json,
+    dev_tag_for_sha,
+    sha_for_tag,
+    is_ancestor,
+    git_ok,
+)
 
+WORKFLOW_NAME = os.getenv("WORKFLOW_NAME", "3. CD | Deploy to Test")
+LIMIT = int(os.getenv("LIMIT", "30"))
 EVENT_NAME = os.getenv("EVENT_NAME", "")
 HEAD_SHA_AUTO = os.getenv("WORKFLOW_RUN_HEAD_SHA", "")
 MANUAL_REF = os.getenv("MANUAL_REF", "")
 ALLOW_OLDER = os.getenv("ALLOW_OLDER", "true").lower()
 
-
-def fail(msg: str) -> "NoReturn":
-    print(f"::error::{msg}", file=sys.stderr)
-    sys.exit(1)
-
-def run(cmd: List[str], *, check: bool = True) -> subprocess.CompletedProcess:
-    # cp = completed process (will use this to refer)
-    return subprocess.run(cmd, check=check, capture_output=True, text=True)
-
-def git_ok(args: List[str]) -> bool:
-    return subprocess.run(["git", *args]).returncode == 0
-
-def is_ancestor(older: str, newer: str) -> bool:
-    return subprocess.run(["git", "merge-base", "--is-ancestor", older, newer]).returncode == 0
-
-def fetch_latest_from_remote() -> None:
-    run(["git", "fetch", "origin", BRANCH, "--quiet"])
-    run(["git", "fetch", "--tags", "--force", "--quiet"])
-
-def gh_json(args: List[str]) -> list:
-    cp = run(["gh", *args])
-    raw = cp.stdout.strip()
-    return json.loads(raw) if raw else []
-
-def dev_tag_for_sha(sha: str) -> Optional[str]:
-    cp = run(["git", "tag", "--points-at", sha], check=False)
-    for tag in cp.stdout.splitlines():
-        if tag.startswith("dev-"):
-            return tag
-    return None
-
-def sha_for_tag(tag: str) -> Optional[str]:
-    cp = run(["git", "rev-list", "-n1", tag], check=False)
-    return cp.stdout.strip() or None
-
-
 @dataclass(frozen=True)
 class RefInfo:
     sha: str
     ref: str
 
-
-def require_token() -> None:
-    if not (os.getenv("GH_TOKEN") or os.getenv("GITHUB_TOKEN")):
-        fail("GH_TOKEN/GITHUB_TOKEN is required")
-
 def list_successful_test_shas() -> List[str]:
-    """
-    These are commits that have been deployed into
-    the test env successfully
-    """
     data = gh_json([
         "run", "list",
         "--workflow", WORKFLOW_NAME,
@@ -87,7 +50,7 @@ def list_successful_test_shas() -> List[str]:
         "--json", "headSha",
         "--limit", str(LIMIT),
     ])
-    return [commit["headSha"] for commit in data if commit.get("headSha")]
+    return [c["headSha"] for c in data if c.get("headSha")]
 
 def pick_furthest_ahead(shas: List[str]) -> str:
     latest: Optional[str] = None
@@ -111,7 +74,7 @@ def resolve_latest_test() -> RefInfo:
         fail(f"No dev-* tag found on latest TEST SHA ({latest_sha})")
     return RefInfo(sha=latest_sha, ref=latest_ref)
 
-def resolve_this_run() -> RefInfo | None:
+def resolve_this_run() -> RefInfo:
     if EVENT_NAME == "workflow_run":
         if not HEAD_SHA_AUTO:
             fail("WORKFLOW_RUN_HEAD_SHA missing for workflow_run event")
@@ -123,6 +86,8 @@ def resolve_this_run() -> RefInfo | None:
     if EVENT_NAME == "workflow_dispatch":
         if not MANUAL_REF:
             fail("MANUAL_REF (inputs.ref) is required for manual dispatch")
+        if not re.match(r"^dev-\d{14}$", MANUAL_REF):
+            fail(f"Invalid dev-* tag format: {MANUAL_REF}")
         if not git_ok(["rev-parse", "-q", "--verify", f"refs/tags/{MANUAL_REF}"]):
             fail(f"Tag not found: {MANUAL_REF}")
         sha = sha_for_tag(MANUAL_REF)
@@ -132,9 +97,10 @@ def resolve_this_run() -> RefInfo | None:
             fail(f"Chosen tag {MANUAL_REF} is not on origin/{BRANCH} history")
         return RefInfo(sha=sha, ref=MANUAL_REF)
 
+    fail(f"Unsupported EVENT_NAME: {EVENT_NAME}")
+
 def enforce_guard(current: RefInfo, latest: RefInfo) -> None:
     older_than_latest = (current.sha != latest.sha) and is_ancestor(current.sha, latest.sha)
-
     if EVENT_NAME == "workflow_run":
         if older_than_latest:
             fail(
@@ -155,9 +121,8 @@ def write_outputs(current: RefInfo, latest: RefInfo) -> None:
         f.write(f"latest_test_sha={latest.sha}\n")
         f.write(f"latest_test_ref={latest.ref}\n")
 
-
 def main() -> int:
-    require_token()
+    ensure_token()
     fetch_latest_from_remote()
     latest = resolve_latest_test()
     current = resolve_this_run()
@@ -167,6 +132,5 @@ def main() -> int:
     print(f"LATEST TEST: {latest.ref} ({latest.sha})")
     return 0
 
-
 if __name__ == "__main__":
     sys.exit(main())

scripts/workflow/release_type_resolver.py

@@ -1,86 +1,42 @@
+#!/usr/bin/env python3
 """
 Resolve release_type from PR labels with safe defaults.
 
 Modes
-
 - Manual override (MANUAL_RELEASE_TYPE): emit that and exit.
 - Single-PR mode (default): inspect PR labels for THIS_SHA; default "rc".
-- Aggregate mode (AGGREGATE=true): consider TEST deployed PRs merged since latest final tag*
-   up to LATEST_TEST_SHA (BOUNDARY), and pick highest of major > minor > patch > rc
+- Aggregate mode (AGGREGATE=true): consider TEST-deployed PRs merged since latest final tag
+  up to LATEST_TEST_SHA (BOUNDARY), and pick highest of major > minor > patch > rc.
 
 Env inputs
-
-GH_TOKEN / GITHUB_TOKEN: required
-THIS_SHA: SHA being promoted (required unless manual override)
-LATEST_TEST_SHA: required when AGGREGATE=true
-MANUAL_RELEASE_TYPE: (rc|patch|minor|major)
-AGGREGATE: "true"|"false" (default "false")
-BRANCH: branch (default "main")
+- GH_TOKEN / GITHUB_TOKEN: required
+- THIS_SHA: SHA being promoted (required unless manual override)
+- LATEST_TEST_SHA: required when AGGREGATE=true
+- MANUAL_RELEASE_TYPE: (rc|patch|minor|major)
+- AGGREGATE: "true"|"false" (default "false")
+- BRANCH: branch (default "main")
 
 Outputs
-
-release_type: rc|patch|minor|major
-basis: manual|single-pr|aggregate
-pr_numbers: comma-separated PR numbers considered
+- release_type: rc|patch|minor|major
+- basis: manual|single-pr|aggregate
+- pr_numbers: comma-separated PR numbers considered
 """
 
-import os, subprocess, sys
+from __future__ import annotations
+import os
+import sys
 from typing import List, Set
 
-BRANCH = os.getenv("BRANCH", "main")
-
-def run(cmd: List[str], check=True, capture=True) -> subprocess.CompletedProcess:
-    # cp = completed process (will use this to refer)
-    return subprocess.run(cmd, check=check, capture_output=capture, text=True)
-
-def fail(msg: str) -> int:
-    print(f"::error::{msg}", file=sys.stderr)
-    return 1
-
-def fetch_latest_from_remote():
-    run(["git","fetch","origin", BRANCH, "--quiet"], check=True)
-    run(["git","fetch","--tags","--force","--quiet"], check=True)
-
-def gh_api(path: str, jq: str | None = None) -> List[str]:
-    """
-    A simple python wrapper around the GitHub API
-    to make it a callable function.
-    """
-    args = ["gh","api", path]
-    if jq:
-        args += ["--jq", jq]
-    cp = run(args, check=True)
-    return [x for x in cp.stdout.splitlines() if x]
-
-def latest_final_tag() -> str | None:
-    """
-    Grabs the version tags and sorts semantically desc
-    """
-    cp = run(["git","tag","--list","v[0-9]*.[0-9]*.[0-9]*","--sort=-v:refname"], check=True)
-    tags = cp.stdout.splitlines()
-    return tags[0] if tags else None
-
-def first_commit() -> str:
-    """
-    Returns the first commit of the current branch.
-
-    We will never use this for our project since we
-    already have a release but can be used as a
-    fallback for new projects.
-    """
-    return run(["git","rev-list","--max-parents=0","HEAD"], check=True).stdout.strip()
-
-def list_merged_pr_commits(base: str, head: str) -> List[str]:
-    rng = f"{base}..{head}"
-    cp = run(["git","rev-list","--merges","--first-parent", rng], check=False)
-    return [x for x in cp.stdout.splitlines() if x]
-
-def prs_for_commit(sha: str) -> List[int]:
-    nums = gh_api(f"/repos/{os.getenv('GITHUB_REPOSITORY')}/commits/{sha}/pulls", jq=".[].number")
-    return [int(n) for n in nums]
-
-def labels_for_pr(pr: int) -> List[str]:
-    return gh_api(f"/repos/{os.getenv('GITHUB_REPOSITORY')}/issues/{pr}/labels", jq=".[].name")
+from ci_utils import (
+    ensure_token,
+    fetch_latest_from_remote,
+    latest_final_tag,
+    first_commit,
+    list_merged_pr_commits,
+    prs_for_commit,
+    labels_for_pr,
+    fail,
+)
 
 def pick_highest(labels: List[str]) -> str | None:
     has_major = any(l == "release:major" for l in labels)
@@ -94,13 +50,12 @@ def pick_highest(labels: List[str]) -> str | None:
     return None
 
 def main() -> int:
-    if not (os.getenv("GH_TOKEN") or os.getenv("GITHUB_TOKEN")):
-        return fail("GH_TOKEN/GITHUB_TOKEN is required")
+    ensure_token()
 
     manual = (os.getenv("MANUAL_RELEASE_TYPE") or "").strip()
     if manual:
         if manual not in {"rc","patch","minor","major"}:
-            return fail(f"Invalid MANUAL_RELEASE_TYPE: {manual}")
+            fail(f"Invalid MANUAL_RELEASE_TYPE: {manual}")
         out = os.getenv("GITHUB_OUTPUT")
         if out:
             with open(out, "a") as f:
@@ -112,7 +67,7 @@ def main() -> int:
 
     this_sha = (os.getenv("THIS_SHA") or "").strip()
     if not this_sha:
-        return fail("Cannot determine sha")
+        fail("Cannot determine sha")
 
     fetch_latest_from_remote()
 
@@ -124,7 +79,7 @@ def main() -> int:
     if aggregate:
         latest_test_sha = (os.getenv("LATEST_TEST_SHA") or "").strip()
         if not latest_test_sha:
-            return fail("LATEST_TEST_SHA is required when AGGREGATE=true")
+            fail("LATEST_TEST_SHA is required when AGGREGATE=true")
         base = latest_final_tag() or first_commit()
         merges = list_merged_pr_commits(base, latest_test_sha)
         for m in merges: