(ELI-466) refactoring

Author: TOEL2

.act/auto_preprod_trigger_latest.json

@@ -12,7 +12,7 @@
     "status": "completed",
     "conclusion": "success",
     "head_branch": "main",
-    "head_sha": "166abc418e62d986484c8336f06223fc9135e864"
+    "head_sha": "758a8d751ca1885695e9ac0766fcc0007dfb2995"
   },
   "repository": {
     "full_name": "NHSDigital/eligibility-signposting-api",

scripts/workflow/ci_utils.py

@@ -15,6 +15,10 @@
 BRANCH = os.getenv("BRANCH", "main")
 REPO_FALLBACK = "NHSDigital/eligibility-signposting-api"
 
+
+def _repo() -> str:
+    return os.environ.get("GITHUB_REPOSITORY") or REPO_FALLBACK
+
 def fail(msg: str) -> NoReturn:
     print(f"::error::{msg}", file=sys.stderr)
     sys.exit(1)
@@ -41,34 +45,6 @@ def git_ok(args: List[str]) -> bool:
 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]) -> Any:
-    # Map GITHUB_TOKEN -> GH_TOKEN (gh prefers GH_TOKEN)
-    if "GH_TOKEN" not in os.environ and os.environ.get("GITHUB_TOKEN"):
-        os.environ["GH_TOKEN"] = os.environ["GITHUB_TOKEN"]
-
-    # Ensure repo is explicit (act containers often need this)
-    repo = os.environ.get("GITHUB_REPOSITORY") or REPO_FALLBACK
-    base = ["gh", *args]
-    if "--repo" not in args and "-R" not in args:
-        base.extend(["--repo", repo])
-
-    cp = run(base, check=True)
-    try:
-        return json.loads(cp.stdout or "null")
-    except json.JSONDecodeError as e:
-        raise RuntimeError(f"Failed to parse gh JSON: {e}\nSTDOUT:\n{cp.stdout}\nSTDERR:\n{cp.stderr}")
-
-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():
@@ -80,7 +56,7 @@ 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]:
+def latest_release_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
@@ -91,24 +67,100 @@ def first_commit() -> str:
 
     We will never use this for our project since we
     already have a release but can be used as a
-    fallback for new projects.
+    fallback.
     """
     return run(["git", "rev-list", "--max-parents=0", "HEAD"], check=True).stdout.strip()
 
-def list_merged_pr_commits(base: str, head: str) -> List[str]:
+def labels_for_pr(pr: int) -> List[str]:
+    """Return all labels on a PR (or issue)."""
+    args = [
+        "gh", "api",
+        f"/repos/{_repo()}/issues/{pr}/labels",
+        "-H", "X-GitHub-Api-Version: 2022-11-28",
+        "--jq", ".[].name",
+    ]
+    cp = run(args, check=False)
+    if cp.returncode not in (0, 1):
+        print(f"Warning: gh api exit {cp.returncode} for PR #{pr}", file=sys.stderr)
+    if cp.stdout.strip():
+        return [x.strip() for x in cp.stdout.splitlines() if x.strip()]
+    return []
+
+
+def commit_subject(sha: str) -> str:
+    """Return the one-line subject for a commit SHA."""
+    cp = run(["git", "log", "-1", "--pretty=%s", sha], check=False)
+    return (cp.stdout or "").strip()
+
+def parse_merge_subject_for_pr_numbers(subject: str) -> List[int]:
+    """
+    Extract PR numbers from common merge subjects, e.g.:
+    - 'Merge pull request #123 from ...'
+    - 'Some feature (#456)'
+    """
+    import re
+    nums = set()
+    for m in re.finditer(r"(?:#|\bPR\s*#)(\d+)", subject, flags=re.IGNORECASE):
+        try:
+            nums.add(int(m.group(1)))
+        except ValueError:
+            pass
+    # Also match explicit 'Merge pull request #123'
+    m2 = re.search(r"Merge pull request #(\d+)", subject, flags=re.IGNORECASE)
+    if m2:
+        try:
+            nums.add(int(m2.group(1)))
+        except ValueError:
+            pass
+    return sorted(nums)
+
+def list_merge_commits(base: str, head: str) -> List[str]:
+    """
+    Merge commits on the first-parent path from base..head.
+    """
     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 list_all_commits(base: str, head: str) -> List[str]:
+    """
+    All commits on the first-parent path base..head (includes squash merges as single commits).
+    """
+    rng = f"{base}..{head}"
+    # We choose first-parent so we're scanning the mainline history only.
+    cp = run(["git", "rev-list", "--first-parent", rng], check=False)
+    return [x for x in cp.stdout.splitlines() if x]
 
-def labels_for_pr(pr: int) -> List[str]:
-    return gh_api(
-        f"/repos/{os.getenv('GITHUB_REPOSITORY')}/issues/{pr}/labels",
-        jq=".[].name",
-    )
+def prs_for_commit_via_api(sha: str) -> List[int]:
+    """
+    Use the official endpoint linking any commit to associated PRs.
+    Works for merge commits and for commits that landed via rebase merges.
+    """
+    args = [
+        "gh", "api",
+        f"/repos/{_repo()}/commits/{sha}/pulls",
+        "-H", "Accept: application/vnd.github.groot-preview+json",
+        "-H", "X-GitHub-Api-Version: 2022-11-28",
+        "--jq", ".[].number"
+    ]
+    try:
+        cp = run(args, check=True)
+        return [int(x) for x in cp.stdout.splitlines() if x]
+    except subprocess.CalledProcessError as e:
+        print(f"Warning: commit→PR lookup failed for {sha[:7]} ({e.returncode})", file=sys.stderr)
+        return []
+
+def title_for_pr(pr: int) -> str:
+    """
+    Return the title of a pull request.
+    """
+    args = [
+        "gh", "pr", "view", str(pr),
+        "--repo", _repo(),
+        "--json", "title",
+        "--jq", ".title",
+    ]
+    cp = run(args, check=False)
+    if cp.returncode not in (0, 1):
+        print(f"Warning: gh pr view exit {cp.returncode} for PR #{pr}", file=sys.stderr)
+    return cp.stdout.strip()

scripts/workflow/pre-release_resolver.py

@@ -44,7 +44,8 @@ class RefInfo:
     ref: str
 
 def get_event_name() -> str:
-    """Determine the effective event name, correcting for act quirks."""
+    """Determine the effective event name,
+     correcting for act quirks."""
     evt_env = os.getenv("GITHUB_EVENT_NAME", "")
     evt_payload = None
     path = os.getenv("GITHUB_EVENT_PATH")
@@ -81,7 +82,6 @@ def _run_gh(args: list[str]) -> str:
         raise RuntimeError(f"Command failed: {' '.join(cmd)}")
     return cp.stdout
 
-
 def list_successful_test_shas() -> List[str]:
     """
     Return SHAs for successful runs of the test deploy workflow on BRANCH.