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     print(f"::set-output name={name}::{value}")
  56
  57
  58 def http_get(url: str, *, is_json: bool = True, **kwargs: Any) -> Any:
  59     headers = kwargs.get("headers") or {}
  60     headers["User-Agent"] = USER_AGENT
  61     if "github" in url:
  62         if GH_API_TOKEN:
  63             headers["Authorization"] = f"token {GH_API_TOKEN}"
  64         headers["Accept"] = "application/vnd.github.v3+json"
  65     kwargs["headers"] = headers
  66
  67     r = http.request("GET", url, **kwargs)
  68     if is_json:
  69         data = json.loads(r.data.decode("utf-8"))
  70     else:
  71         data = r.data
  72     print(f"[INFO]: issued GET request for {r.geturl()}")
  73     if not (200 <= r.status < 300):
  74         pprint.pprint(dict(r.info()))
  75         pprint.pprint(data)
  76         raise RuntimeError(f"unexpected status code: {r.status}")
  77
  78     return data
  79
  80
  81 def get_main_revision() -> str:
  82     data = http_get(
  83         f"https://api.github.com/repos/{REPO}/commits",
  84         fields={"per_page": "1", "sha": "main"},
  85     )
  86     assert isinstance(data[0]["sha"], str)
  87     return data[0]["sha"]
  88
  89
  90 def get_pr_revision(pr: int) -> str:
  91     data = http_get(f"https://api.github.com/repos/{REPO}/pulls/{pr}")
  92     assert isinstance(data["head"]["sha"], str)
  93     return data["head"]["sha"]
  94
  95
  96 def get_pypi_version() -> Version:
  97     data = http_get("https://pypi.org/pypi/black/json")
  98     versions = [Version(v) for v in data["releases"]]
  99     sorted_versions = sorted(versions, reverse=True)
 100     return sorted_versions[0]
 101
 102
 103 @click.group()
 104 def main() -> None:
 105     pass
 106
 107
 108 @main.command("config", help="Acquire run configuration and metadata.")
 109 @click.argument("event", type=click.Choice(["push", "pull_request"]))
 110 def config(event: Literal["push", "pull_request"]) -> None:
 111     import diff_shades
 112
 113     if event == "push":
 114         jobs = [{"mode": "preview-changes", "force-flag": "--force-preview-style"}]
 115         # Push on main, let's use PyPI Black as the baseline.
 116         baseline_name = str(get_pypi_version())
 117         baseline_cmd = f"git checkout {baseline_name}"
 118         target_rev = os.getenv("GITHUB_SHA")
 119         assert target_rev is not None
 120         target_name = "main-" + target_rev[:SHA_LENGTH]
 121         target_cmd = f"git checkout {target_rev}"
 122
 123     elif event == "pull_request":
 124         jobs = [
 125             {"mode": "preview-changes", "force-flag": "--force-preview-style"},
 126             {"mode": "assert-no-changes", "force-flag": "--force-stable-style"},
 127         ]
 128         # PR, let's use main as the baseline.
 129         baseline_rev = get_main_revision()
 130         baseline_name = "main-" + baseline_rev[:SHA_LENGTH]
 131         baseline_cmd = f"git checkout {baseline_rev}"
 132         pr_ref = os.getenv("GITHUB_REF")
 133         assert pr_ref is not None
 134         pr_num = int(pr_ref[10:-6])
 135         pr_rev = get_pr_revision(pr_num)
 136         target_name = f"pr-{pr_num}-{pr_rev[:SHA_LENGTH]}"
 137         target_cmd = f"gh pr checkout {pr_num} && git merge origin/main"
 138
 139     env = f"{platform.system()}-{platform.python_version()}-{diff_shades.__version__}"
 140     for entry in jobs:
 141         entry["baseline-analysis"] = f"{entry['mode']}-{baseline_name}.json"
 142         entry["baseline-setup-cmd"] = baseline_cmd
 143         entry["target-analysis"] = f"{entry['mode']}-{target_name}.json"
 144         entry["target-setup-cmd"] = target_cmd
 145         entry["baseline-cache-key"] = f"{env}-{baseline_name}-{entry['mode']}"
 146         if event == "pull_request":
 147             # These are only needed for the PR comment.
 148             entry["baseline-sha"] = baseline_rev
 149             entry["target-sha"] = pr_rev
 150
 151     set_output("matrix", json.dumps(jobs, indent=None))
 152     pprint.pprint(jobs)
 153
 154
 155 @main.command("comment-body", help="Generate the body for a summary PR comment.")
 156 @click.argument("baseline", type=click.Path(exists=True, path_type=Path))
 157 @click.argument("target", type=click.Path(exists=True, path_type=Path))
 158 @click.argument("baseline-sha")
 159 @click.argument("target-sha")
 160 @click.argument("pr-num", type=int)
 161 def comment_body(
 162     baseline: Path, target: Path, baseline_sha: str, target_sha: str, pr_num: int
 163 ) -> None:
 164     # fmt: off
 165     cmd = [
 166         sys.executable, "-m", "diff_shades", "--no-color",
 167         "compare", str(baseline), str(target), "--quiet", "--check"
 168     ]
 169     # fmt: on
 170     proc = subprocess.run(cmd, stdout=subprocess.PIPE, encoding="utf-8")
 171     if not proc.returncode:
 172         body = (
 173             f"**diff-shades** reports zero changes comparing this PR ({target_sha}) to"
 174             f" main ({baseline_sha}).\n\n---\n\n"
 175         )
 176     else:
 177         body = (
 178             f"**diff-shades** results comparing this PR ({target_sha}) to main"
 179             f" ({baseline_sha}). The full diff is [available in the logs]"
 180             f'($job-diff-url) under the "{DIFF_STEP_NAME}" step.'
 181         )
 182         body += "\n```text\n" + proc.stdout.strip() + "\n```\n"
 183     body += (
 184         f"[**What is this?**]({DOCS_URL}) | [Workflow run]($workflow-run-url) |"
 185         " [diff-shades documentation](https://github.com/ichard26/diff-shades#readme)"
 186     )
 187     print(f"[INFO]: writing comment details to {COMMENT_FILE}")
 188     with open(COMMENT_FILE, "w", encoding="utf-8") as f:
 189         json.dump({"body": body, "pr-number": pr_num}, f)
 190
 191
 192 @main.command("comment-details", help="Get PR comment resources from a workflow run.")
 193 @click.argument("run-id")
 194 def comment_details(run_id: str) -> None:
 195     data = http_get(f"https://api.github.com/repos/{REPO}/actions/runs/{run_id}")
 196     if data["event"] != "pull_request" or data["conclusion"] == "cancelled":
 197         set_output("needs-comment", "false")
 198         return
 199
 200     set_output("needs-comment", "true")
 201     jobs = http_get(data["jobs_url"])["jobs"]
 202     job = next(j for j in jobs if j["name"] == "analysis / preview-changes")
 203     diff_step = next(s for s in job["steps"] if s["name"] == DIFF_STEP_NAME)
 204     diff_url = job["html_url"] + f"#step:{diff_step['number']}:1"
 205
 206     artifacts = http_get(data["artifacts_url"])["artifacts"]
 207     comment_artifact = next(a for a in artifacts if a["name"] == COMMENT_FILE)
 208     comment_url = comment_artifact["archive_download_url"]
 209     comment_zip = BytesIO(http_get(comment_url, is_json=False))
 210     with zipfile.ZipFile(comment_zip) as zfile:
 211         with zfile.open(COMMENT_FILE) as rf:
 212             comment_data = json.loads(rf.read().decode("utf-8"))
 213
 214     set_output("pr-number", str(comment_data["pr-number"]))
 215     body = comment_data["body"]
 216     # It's more convenient to fill in these fields after the first workflow is done
 217     # since this command can access the workflows API (doing it in the main workflow
 218     # while it's still in progress seems impossible).
 219     body = body.replace("$workflow-run-url", data["html_url"])
 220     body = body.replace("$job-diff-url", diff_url)
 221     # https://github.community/t/set-output-truncates-multiline-strings/16852/3
 222     escaped = body.replace("%", "%25").replace("\n", "%0A").replace("\r", "%0D")
 223     set_output("comment-body", escaped)
 224
 225
 226 if __name__ == "__main__":
 227     main()