scripts/diff_shades_gha_helper.py

   1 """Helper script for psf/black's diff-shades Github Actions integration.
   2
   3 diff-shades is a tool for analyzing what happens when you run Black on
   4 OSS code capturing it for comparisons or other usage. It's used here to
   5 help measure the impact of a change *before* landing it (in particular
   6 posting a comment on completion for PRs).
   7
   8 This script exists as a more maintainable alternative to using inline
   9 Javascript in the workflow YAML files. The revision configuration and
  10 resolving, caching, and PR comment logic is contained here.
  11
  12 For more information, please see the developer docs:
  13
  14 https://black.readthedocs.io/en/latest/contributing/gauging_changes.html#diff-shades
  15 """
  16
  17 import json
  18 import os
  19 import platform
  20 import pprint
  21 import subprocess
  22 import sys
  23 import zipfile
  24 from io import BytesIO
  25 from pathlib import Path
  26 from typing import Any
  27
  28 import click
  29 import urllib3
  30 from packaging.version import Version
  31
  32 if sys.version_info >= (3, 8):
  33     from typing import Final, Literal
  34 else:
  35     from typing_extensions import Final, Literal
  36
  37 COMMENT_FILE: Final = ".pr-comment.json"
  38 DIFF_STEP_NAME: Final = "Generate HTML diff report"
  39 DOCS_URL: Final = (
  40     "https://black.readthedocs.io/en/latest/"
  41     "contributing/gauging_changes.html#diff-shades"
  42 )
  43 USER_AGENT: Final = f"psf/black diff-shades workflow via urllib3/{urllib3.__version__}"
  44 SHA_LENGTH: Final = 10
  45 GH_API_TOKEN: Final = os.getenv("GITHUB_TOKEN")
  46 REPO: Final = os.getenv("GITHUB_REPOSITORY", default="psf/black")
  47 http = urllib3.PoolManager()
  48
  49
  50 def set_output(name: str, value: str) -> None:
  51     if len(value) < 200:
  52         print(f"[INFO]: setting '{name}' to '{value}'")
  53     else:
  54         print(f"[INFO]: setting '{name}' to [{len(value)} chars]")
  55
  56     # Originally the `set-output` workflow command was used here, now replaced
  57     # by setting variables through the `GITHUB_OUTPUT` environment variable
  58     # to stay up to date with GitHub's update.
  59     if "GITHUB_OUTPUT" in os.environ:
  60         with open(os.environ["GITHUB_OUTPUT"], "a") as f:
  61             print(f"{name}={value}", file=f)
  62
  63
  64 def http_get(url: str, *, is_json: bool = True, **kwargs: Any) -> Any:
  65     headers = kwargs.get("headers") or {}
  66     headers["User-Agent"] = USER_AGENT
  67     if "github" in url:
  68         if GH_API_TOKEN:
  69             headers["Authorization"] = f"token {GH_API_TOKEN}"
  70         headers["Accept"] = "application/vnd.github.v3+json"
  71     kwargs["headers"] = headers
  72
  73     r = http.request("GET", url, **kwargs)
  74     if is_json:
  75         data = json.loads(r.data.decode("utf-8"))
  76     else:
  77         data = r.data
  78     print(f"[INFO]: issued GET request for {r.geturl()}")
  79     if not (200 <= r.status < 300):
  80         pprint.pprint(dict(r.info()))
  81         pprint.pprint(data)
  82         raise RuntimeError(f"unexpected status code: {r.status}")
  83
  84     return data
  85
  86
  87 def get_main_revision() -> str:
  88     data = http_get(
  89         f"https://api.github.com/repos/{REPO}/commits",
  90         fields={"per_page": "1", "sha": "main"},
  91     )
  92     assert isinstance(data[0]["sha"], str)
  93     return data[0]["sha"]
  94
  95
  96 def get_pr_revision(pr: int) -> str:
  97     data = http_get(f"https://api.github.com/repos/{REPO}/pulls/{pr}")
  98     assert isinstance(data["head"]["sha"], str)
  99     return data["head"]["sha"]
 100
 101
 102 def get_pypi_version() -> Version:
 103     data = http_get("https://pypi.org/pypi/black/json")
 104     versions = [Version(v) for v in data["releases"]]
 105     sorted_versions = sorted(versions, reverse=True)
 106     return sorted_versions[0]
 107
 108
 109 @click.group()
 110 def main() -> None:
 111     pass
 112
 113
 114 @main.command("config", help="Acquire run configuration and metadata.")
 115 @click.argument("event", type=click.Choice(["push", "pull_request"]))
 116 def config(event: Literal["push", "pull_request"]) -> None:
 117     import diff_shades
 118
 119     if event == "push":
 120         jobs = [{"mode": "preview-changes", "force-flag": "--force-preview-style"}]
 121         # Push on main, let's use PyPI Black as the baseline.
 122         baseline_name = str(get_pypi_version())
 123         baseline_cmd = f"git checkout {baseline_name}"
 124         target_rev = os.getenv("GITHUB_SHA")
 125         assert target_rev is not None
 126         target_name = "main-" + target_rev[:SHA_LENGTH]
 127         target_cmd = f"git checkout {target_rev}"
 128
 129     elif event == "pull_request":
 130         jobs = [
 131             {"mode": "preview-changes", "force-flag": "--force-preview-style"},
 132             {"mode": "assert-no-changes", "force-flag": "--force-stable-style"},
 133         ]
 134         # PR, let's use main as the baseline.
 135         baseline_rev = get_main_revision()
 136         baseline_name = "main-" + baseline_rev[:SHA_LENGTH]
 137         baseline_cmd = f"git checkout {baseline_rev}"
 138         pr_ref = os.getenv("GITHUB_REF")
 139         assert pr_ref is not None
 140         pr_num = int(pr_ref[10:-6])
 141         pr_rev = get_pr_revision(pr_num)
 142         target_name = f"pr-{pr_num}-{pr_rev[:SHA_LENGTH]}"
 143         target_cmd = f"gh pr checkout {pr_num} && git merge origin/main"
 144
 145     env = f"{platform.system()}-{platform.python_version()}-{diff_shades.__version__}"
 146     for entry in jobs:
 147         entry["baseline-analysis"] = f"{entry['mode']}-{baseline_name}.json"
 148         entry["baseline-setup-cmd"] = baseline_cmd
 149         entry["target-analysis"] = f"{entry['mode']}-{target_name}.json"
 150         entry["target-setup-cmd"] = target_cmd
 151         entry["baseline-cache-key"] = f"{env}-{baseline_name}-{entry['mode']}"
 152         if event == "pull_request":
 153             # These are only needed for the PR comment.
 154             entry["baseline-sha"] = baseline_rev
 155             entry["target-sha"] = pr_rev
 156
 157     set_output("matrix", json.dumps(jobs, indent=None))
 158     pprint.pprint(jobs)
 159
 160
 161 @main.command("comment-body", help="Generate the body for a summary PR comment.")
 162 @click.argument("baseline", type=click.Path(exists=True, path_type=Path))
 163 @click.argument("target", type=click.Path(exists=True, path_type=Path))
 164 @click.argument("baseline-sha")
 165 @click.argument("target-sha")
 166 @click.argument("pr-num", type=int)
 167 def comment_body(
 168     baseline: Path, target: Path, baseline_sha: str, target_sha: str, pr_num: int
 169 ) -> None:
 170     # fmt: off
 171     cmd = [
 172         sys.executable, "-m", "diff_shades", "--no-color",
 173         "compare", str(baseline), str(target), "--quiet", "--check"
 174     ]
 175     # fmt: on
 176     proc = subprocess.run(cmd, stdout=subprocess.PIPE, encoding="utf-8")
 177     if not proc.returncode:
 178         body = (
 179             f"**diff-shades** reports zero changes comparing this PR ({target_sha}) to"
 180             f" main ({baseline_sha}).\n\n---\n\n"
 181         )
 182     else:
 183         body = (
 184             f"**diff-shades** results comparing this PR ({target_sha}) to main"
 185             f" ({baseline_sha}). The full diff is [available in the logs]"
 186             f'($job-diff-url) under the "{DIFF_STEP_NAME}" step.'
 187         )
 188         body += "\n```text\n" + proc.stdout.strip() + "\n```\n"
 189     body += (
 190         f"[**What is this?**]({DOCS_URL}) | [Workflow run]($workflow-run-url) |"
 191         " [diff-shades documentation](https://github.com/ichard26/diff-shades#readme)"
 192     )
 193     print(f"[INFO]: writing comment details to {COMMENT_FILE}")
 194     with open(COMMENT_FILE, "w", encoding="utf-8") as f:
 195         json.dump({"body": body, "pr-number": pr_num}, f)
 196
 197
 198 @main.command("comment-details", help="Get PR comment resources from a workflow run.")
 199 @click.argument("run-id")
 200 def comment_details(run_id: str) -> None:
 201     data = http_get(f"https://api.github.com/repos/{REPO}/actions/runs/{run_id}")
 202     if data["event"] != "pull_request" or data["conclusion"] == "cancelled":
 203         set_output("needs-comment", "false")
 204         return
 205
 206     set_output("needs-comment", "true")
 207     jobs = http_get(data["jobs_url"])["jobs"]
 208     job = next(j for j in jobs if j["name"] == "analysis / preview-changes")
 209     diff_step = next(s for s in job["steps"] if s["name"] == DIFF_STEP_NAME)
 210     diff_url = job["html_url"] + f"#step:{diff_step['number']}:1"
 211
 212     artifacts = http_get(data["artifacts_url"])["artifacts"]
 213     comment_artifact = next(a for a in artifacts if a["name"] == COMMENT_FILE)
 214     comment_url = comment_artifact["archive_download_url"]
 215     comment_zip = BytesIO(http_get(comment_url, is_json=False))
 216     with zipfile.ZipFile(comment_zip) as zfile:
 217         with zfile.open(COMMENT_FILE) as rf:
 218             comment_data = json.loads(rf.read().decode("utf-8"))
 219
 220     set_output("pr-number", str(comment_data["pr-number"]))
 221     body = comment_data["body"]
 222     # It's more convenient to fill in these fields after the first workflow is done
 223     # since this command can access the workflows API (doing it in the main workflow
 224     # while it's still in progress seems impossible).
 225     body = body.replace("$workflow-run-url", data["html_url"])
 226     body = body.replace("$job-diff-url", diff_url)
 227     # https://github.community/t/set-output-truncates-multiline-strings/16852/3
 228     escaped = body.replace("%", "%25").replace("\n", "%0A").replace("\r", "%0D")
 229     set_output("comment-body", escaped)
 230
 231
 232 if __name__ == "__main__":
 233     main()