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