diff --git a/.github/scripts/cleanup_prek_update_branches.py b/.github/scripts/cleanup_prek_update_branches.py deleted file mode 100644 index 17154d73..00000000 --- a/.github/scripts/cleanup_prek_update_branches.py +++ /dev/null @@ -1,533 +0,0 @@ -"""Clean up stale prek autoupdate pull requests and branches.""" - -from __future__ import annotations - -import argparse -from collections.abc import Mapping, Sequence -from dataclasses import dataclass, field -import json -import logging -import os -import sys -from typing import Protocol -from urllib.error import HTTPError -from urllib.parse import quote -from urllib.request import Request, urlopen - -GITHUB_API_URL = "https://api.github.com" -CLOSED_PULL_PAGE_LIMIT = 5 -LOGGER = logging.getLogger(__name__) - - -@dataclass -class CleanupResult: - """Result of cleaning workflow-owned pull requests and branches. - - Attributes: - closed_prs: Pull request numbers closed as stale. - deleted_branches: Branch names deleted from the repository. - """ - - closed_prs: list[int] = field(default_factory=list) - deleted_branches: list[str] = field(default_factory=list) - - -class GithubCleanupClient(Protocol): - """Protocol for GitHub operations needed by the cleanup routine.""" - - def list_pulls(self, *, state: str, max_pages: int | None = None) -> list[dict[str, object]]: - """List pull requests by state.""" - - def list_refs(self, *, ref_prefix: str) -> list[str]: - """List git refs by ref prefix.""" - - def close_pull(self, pull_number: int) -> None: - """Close a pull request by number.""" - - def delete_ref(self, ref: str) -> None: - """Delete a git ref by name.""" - - -class GithubClient: - """Small GitHub REST client for the workflow cleanup task.""" - - def __init__(self, *, repository: str, token: str) -> None: - """Initialize the client. - - Args: - repository: Repository in ``owner/name`` format. - token: GitHub token for API calls. - """ - self.repository = repository - self.token = token - - def list_pulls(self, *, state: str, max_pages: int | None = None) -> list[dict[str, object]]: - """List pull requests by state. - - Args: - state: Pull request state to request. - max_pages: Optional pagination cap. - - Returns: - Pull request objects from GitHub. - """ - pulls: list[dict[str, object]] = [] - pages_read = 0 - url: str | None = ( - f"{GITHUB_API_URL}/repos/{self.repository}/pulls?state={state}&per_page=100" - ) - while url is not None: - payload, link_header = self._request("GET", url) - if not isinstance(payload, list): - raise TypeError(f"Expected pull request list from {url}") - pulls.extend(pull for pull in payload if isinstance(pull, dict)) - pages_read += 1 - next_url = _next_link(link_header) - if max_pages is not None and pages_read >= max_pages: - if next_url is not None: - LOGGER.info( - "Stopped listing %s pull requests after %s pages.", - state, - max_pages, - ) - break - url = next_url - return pulls - - def list_refs(self, *, ref_prefix: str) -> list[str]: - """List git refs by ref prefix. - - Args: - ref_prefix: Ref prefix after ``refs/`` such as ``heads/branch``. - - Returns: - Full git ref names matching the prefix. - """ - refs: list[str] = [] - safe_ref_prefix = quote(ref_prefix, safe="/") - url: str | None = ( - f"{GITHUB_API_URL}/repos/{self.repository}/git/matching-refs/{safe_ref_prefix}" - ) - while url is not None: - try: - payload, link_header = self._request("GET", url) - except HTTPError as err: - if err.code == 404: - return refs - raise - if not isinstance(payload, list): - raise TypeError(f"Expected ref list from {url}") - refs.extend( - ref - for item in payload - if isinstance(item, dict) - and isinstance(ref := item.get("ref"), str) - and ref.startswith(f"refs/{ref_prefix}") - ) - url = _next_link(link_header) - return refs - - def close_pull(self, pull_number: int) -> None: - """Close a pull request. - - Args: - pull_number: Pull request number to close. - """ - url = f"{GITHUB_API_URL}/repos/{self.repository}/pulls/{pull_number}" - self._request("PATCH", url, payload={"state": "closed"}) - - def delete_ref(self, ref: str) -> None: - """Delete a git ref if it exists. - - Args: - ref: Ref path such as ``heads/branch-name``. - """ - url = f"{GITHUB_API_URL}/repos/{self.repository}/git/refs/{ref}" - try: - self._request("DELETE", url) - except HTTPError as err: - if err.code == 404 or (err.code == 422 and _is_missing_ref_error(err)): - LOGGER.info("Ref %s was already deleted.", ref) - return - raise - - def _request( - self, - method: str, - url: str, - *, - payload: Mapping[str, object] | None = None, - ) -> tuple[object, str | None]: - """Send a GitHub REST request. - - Args: - method: HTTP method. - url: Request URL. - payload: Optional JSON payload. - - Returns: - Decoded payload and Link header. - """ - data = json.dumps(payload).encode() if payload is not None else None - request = Request(url, data=data, headers=_github_headers(self.token), method=method) # noqa: S310 - with urlopen(request, timeout=30) as response: # noqa: S310 - if response.status == 204: - return {}, response.headers.get("Link") - return json.load(response), response.headers.get("Link") - - -def cleanup_update_branches( - *, - client: GithubCleanupClient, - repository: str, - branch: str, - branch_prefix: str, - label_name: str, - author_login: str | None, - body_marker: str | None, - keep_pr_number: int | None, - close_stale_prs: bool, - delete_merged_branches: bool, - delete_stale_branches: bool = False, - keep_latest_open_pr: bool = False, -) -> CleanupResult: - """Clean workflow-owned stale pull requests and branches. - - Args: - client: GitHub client with list, close, and delete methods. - repository: Repository in ``owner/name`` format. - branch: Current workflow update branch. - branch_prefix: Prefix for workflow-owned update branches. - label_name: Label identifying workflow-created PRs. - author_login: Optional author login identifying workflow-created PRs. - body_marker: Optional body text identifying workflow-created PRs. - keep_pr_number: Optional PR number to preserve. - close_stale_prs: Whether to close open stale update PRs. - delete_stale_branches: Whether to delete stale branch-prefix refs. - delete_merged_branches: Whether to delete branches from merged update PRs. - keep_latest_open_pr: Whether to preserve the newest open workflow PR. - - Returns: - Summary of cleanup actions. - """ - result = CleanupResult() - protected_branches: set[str] = set() - branches_to_delete: set[str] = set() - - all_open_pulls = client.list_pulls(state="open") - workflow_open_pulls = [ - pull - for pull in all_open_pulls - if _is_workflow_pull( - pull, - repository=repository, - branch=branch, - branch_prefix=branch_prefix, - label_name=label_name, - author_login=author_login, - body_marker=body_marker, - ) - ] - workflow_open_pull_numbers = {_pull_number(pull) for pull in workflow_open_pulls} - protected_branches.update( - head_ref - for pull in all_open_pulls - if _pull_number(pull) not in workflow_open_pull_numbers - and (head_ref := _same_repo_head_ref(pull, repository=repository)) is not None - and head_ref.startswith(branch_prefix) - ) - latest_open_pr_number = ( - max(workflow_open_pull_numbers, default=None) if keep_latest_open_pr else None - ) - for pull in workflow_open_pulls: - pull_number = _pull_number(pull) - head_ref = _head_ref(pull) - - if pull_number in {keep_pr_number, latest_open_pr_number}: - protected_branches.add(head_ref) - continue - - if not close_stale_prs: - protected_branches.add(head_ref) - continue - - client.close_pull(pull_number) - result.closed_prs.append(pull_number) - branches_to_delete.add(head_ref) - - if delete_stale_branches: - branches_to_delete.update( - branch_name - for ref in client.list_refs(ref_prefix=f"heads/{branch_prefix}") - if (branch_name := _branch_from_ref(ref)).startswith(branch_prefix) - ) - - if delete_merged_branches: - closed_pulls = client.list_pulls(state="closed", max_pages=CLOSED_PULL_PAGE_LIMIT) - for pull in closed_pulls: - if pull.get("merged_at") is not None and _is_workflow_pull( - pull, - repository=repository, - branch=branch, - branch_prefix=branch_prefix, - label_name=label_name, - author_login=author_login, - body_marker=body_marker, - ): - branches_to_delete.add(_head_ref(pull)) - - branches_to_delete -= protected_branches - for branch_name in sorted(branches_to_delete): - client.delete_ref(f"heads/{branch_name}") - result.deleted_branches.append(branch_name) - - return result - - -def _same_repo_head_ref(pull: Mapping[str, object], *, repository: str) -> str | None: - """Return the head ref when a pull request head belongs to this repository. - - Args: - pull: Pull request object. - repository: Repository in ``owner/name`` format. - - Returns: - Same-repository head ref, or None for malformed or forked PR heads. - """ - head = pull.get("head", {}) - if not isinstance(head, dict): - return None - head_ref = head.get("ref") - if not isinstance(head_ref, str): - return None - head_repo = head.get("repo", {}) - if not isinstance(head_repo, dict) or head_repo.get("full_name") != repository: - return None - return head_ref - - -def _branch_from_ref(ref: str) -> str: - """Return a branch name from a full heads ref. - - Args: - ref: Full git ref such as ``refs/heads/branch-name``. - - Returns: - Branch name. - - Raises: - ValueError: If the ref is not a branch ref. - """ - prefix = "refs/heads/" - if not ref.startswith(prefix): - raise ValueError(f"Expected a branch ref, got {ref}") - return ref.removeprefix(prefix) - - -def _is_workflow_pull( - pull: Mapping[str, object], - *, - repository: str, - branch: str, - branch_prefix: str, - label_name: str, - author_login: str | None, - body_marker: str | None, -) -> bool: - """Return whether a pull request belongs to this workflow. - - Args: - pull: Pull request object. - repository: Repository in ``owner/name`` format. - branch: Current workflow update branch. - branch_prefix: Prefix for workflow-owned update branches. - label_name: Label identifying workflow-created PRs. - author_login: Optional author login identifying workflow-created PRs. - body_marker: Optional body text identifying workflow-created PRs. - - Returns: - True when the PR head branch is owned by this workflow. - """ - labels = pull.get("labels", []) - if not isinstance(labels, list) or not any( - isinstance(label, dict) and label.get("name") == label_name for label in labels - ): - return False - - if author_login is not None: - user = pull.get("user", {}) - if not isinstance(user, dict) or user.get("login") != author_login: - return False - - if body_marker is not None: - body = pull.get("body") - if not isinstance(body, str) or body_marker not in body: - return False - - head = pull.get("head", {}) - if not isinstance(head, dict): - return False - head_ref = head.get("ref") - if not isinstance(head_ref, str): - return False - if head_ref != branch and not head_ref.startswith(branch_prefix): - return False - - head_repo = head.get("repo", {}) - if not isinstance(head_repo, dict): - return False - return head_repo.get("full_name") == repository - - -def _head_ref(pull: Mapping[str, object]) -> str: - """Return a pull request head ref. - - Args: - pull: Pull request object. - - Returns: - Pull request head ref. - """ - head = pull["head"] - if not isinstance(head, dict) or not isinstance(head.get("ref"), str): - raise TypeError("Pull request is missing a head ref") - return head["ref"] - - -def _pull_number(pull: Mapping[str, object]) -> int: - """Return a pull request number. - - Args: - pull: Pull request object. - - Returns: - Pull request number. - """ - number = pull["number"] - if isinstance(number, int): - return number - if isinstance(number, str): - return int(number) - raise TypeError("Pull request is missing a numeric number") - - -def _github_headers(token: str) -> dict[str, str]: - """Build GitHub API request headers. - - Args: - token: GitHub token. - - Returns: - Request headers. - """ - return { - "Accept": "application/vnd.github+json", - "Authorization": f"Bearer {token}", - "Content-Type": "application/json", - "X-GitHub-Api-Version": "2022-11-28", - "User-Agent": "prek-autoupdate-cleanup", - } - - -def _is_missing_ref_error(err: HTTPError) -> bool: - """Return whether a GitHub 422 error reports a missing ref. - - Args: - err: HTTP error from the GitHub API. - - Returns: - True when the error body says the reference does not exist. - """ - body = err.read().decode(errors="replace") - if not body: - return False - try: - payload = json.loads(body) - except json.JSONDecodeError: - return "Reference does not exist" in body - if not isinstance(payload, dict): - return False - message = payload.get("message") - return isinstance(message, str) and "Reference does not exist" in message - - -def _next_link(link_header: str | None) -> str | None: - """Return the next pagination URL from a GitHub Link header. - - Args: - link_header: Raw Link header value. - - Returns: - Next URL when present. - """ - if not link_header: - return None - for link in link_header.split(","): - url_part, *params = link.split(";") - if any(param.strip() == 'rel="next"' for param in params): - return url_part.strip()[1:-1] - return None - - -def _parse_args(argv: Sequence[str]) -> argparse.Namespace: - """Parse command-line arguments. - - Args: - argv: Command-line arguments excluding the executable name. - - Returns: - Parsed arguments. - """ - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--repository", required=True) - parser.add_argument("--branch", required=True) - parser.add_argument("--branch-prefix", required=True) - parser.add_argument("--label-name", required=True) - parser.add_argument("--author-login") - parser.add_argument("--body-marker") - parser.add_argument("--keep-pr-number", type=int) - parser.add_argument("--keep-latest-open-pr", action="store_true") - parser.add_argument("--close-stale-prs", action="store_true") - parser.add_argument("--delete-stale-branches", action="store_true") - parser.add_argument("--delete-merged-branches", action="store_true") - return parser.parse_args(argv) - - -def main(argv: Sequence[str] | None = None) -> int: - """Run workflow branch cleanup. - - Args: - argv: Optional command-line arguments excluding the executable name. - - Returns: - Process exit code. - """ - logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") - args = _parse_args(sys.argv[1:] if argv is None else argv) - token = os.environ.get("GITHUB_TOKEN") - if not token: - LOGGER.error("GITHUB_TOKEN is required for cleanup.") - return 1 - - client = GithubClient(repository=args.repository, token=token) - result = cleanup_update_branches( - client=client, - repository=args.repository, - branch=args.branch, - branch_prefix=args.branch_prefix, - label_name=args.label_name, - author_login=args.author_login, - body_marker=args.body_marker, - keep_pr_number=args.keep_pr_number, - keep_latest_open_pr=args.keep_latest_open_pr, - close_stale_prs=args.close_stale_prs, - delete_stale_branches=args.delete_stale_branches, - delete_merged_branches=args.delete_merged_branches, - ) - LOGGER.info("Closed stale PRs: %s", result.closed_prs or "none") - LOGGER.info("Deleted branches: %s", result.deleted_branches or "none") - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/.github/workflows/prek_autoupdate.yml b/.github/workflows/prek_autoupdate.yml index cddecf1e..34cb883b 100644 --- a/.github/workflows/prek_autoupdate.yml +++ b/.github/workflows/prek_autoupdate.yml @@ -1,153 +1,14 @@ name: prek Autoupdate + on: schedule: - - cron: '0 2 * * *' + - cron: '41 5 * * 3' workflow_dispatch: -permissions: - contents: write - pull-requests: write - jobs: - update-hooks: - name: Update prek Hooks - runs-on: ubuntu-latest - steps: - - name: Check weekly update window - id: weekly - env: - EVENT_NAME: ${{ github.event_name }} - run: | - set -euo pipefail - - if [[ "$EVENT_NAME" == "workflow_dispatch" ]] || [[ "$(date -u +%u)" == "1" ]]; then - echo "should_update=true" >> "$GITHUB_OUTPUT" - else - echo "should_update=false" >> "$GITHUB_OUTPUT" - fi - - - name: Checkout Repository - if: steps.weekly.outputs.should_update == 'true' - uses: actions/checkout@v7 - with: - persist-credentials: false - - - name: Set up Python - if: steps.weekly.outputs.should_update == 'true' - uses: actions/setup-python@v6 - with: - python-version: '3.14' - cache: 'pip' - - - name: Install prek - if: steps.weekly.outputs.should_update == 'true' - uses: taiki-e/install-action@v2 - with: - tool: prek - - - name: Run prek auto-update - if: steps.weekly.outputs.should_update == 'true' - id: prek - run: | - set -euo pipefail - body_file="$GITHUB_WORKSPACE/prek-autoupdate-body.md" - { - echo "Automated update of \`prek\` hooks." - echo - echo '```text' - prek auto-update --cooldown-days 7 2>&1 - echo '```' - } | tee "$body_file" - - - name: Create Pull Request - if: steps.weekly.outputs.should_update == 'true' - uses: peter-evans/create-pull-request@v8 - id: cpr - with: - commit-message: 'chore: update prek hooks' - title: 'Bump prek Hooks' - branch: chore/prek-updates - author: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> - labels: dependencies - delete-branch: true - body-path: prek-autoupdate-body.md - add-paths: | - prek.toml - - - name: Close extra auto-generated PRs - if: steps.weekly.outputs.should_update == 'true' && steps.cpr.outputs.pull-request-number != '' - env: - BODY_MARKER: Automated update of `prek` hooks. - GITHUB_TOKEN: ${{ github.token }} - KEEP_PR_NUMBER: ${{ steps.cpr.outputs.pull-request-number }} - REPOSITORY: ${{ github.repository }} - run: | - set -euo pipefail - - python .github/scripts/cleanup_prek_update_branches.py \ - --repository "$REPOSITORY" \ - --branch chore/prek-updates \ - --branch-prefix chore/prek-updates \ - --label-name dependencies \ - --author-login 'github-actions[bot]' \ - --body-marker "$BODY_MARKER" \ - --keep-pr-number "$KEEP_PR_NUMBER" \ - --close-stale-prs \ - --delete-stale-branches \ - --delete-merged-branches - - - name: Close stale prek update PRs - if: steps.weekly.outputs.should_update == 'true' && steps.cpr.outputs.pull-request-number == '' - env: - BODY_MARKER: Automated update of `prek` hooks. - GITHUB_TOKEN: ${{ github.token }} - REPOSITORY: ${{ github.repository }} - run: | - set -euo pipefail - - python .github/scripts/cleanup_prek_update_branches.py \ - --repository "$REPOSITORY" \ - --branch chore/prek-updates \ - --branch-prefix chore/prek-updates \ - --label-name dependencies \ - --author-login 'github-actions[bot]' \ - --body-marker "$BODY_MARKER" \ - --close-stale-prs \ - --delete-stale-branches \ - --delete-merged-branches - - nightly-cleanup: - name: Nightly Cleanup - needs: update-hooks - if: always() && (github.event_name == 'workflow_dispatch' || github.event_name == 'schedule') - runs-on: ubuntu-latest - steps: - - name: Checkout Repository - uses: actions/checkout@v7 - with: - persist-credentials: false - - - name: Set up Python - uses: actions/setup-python@v6 - with: - python-version: '3.14' - - - name: Close stale prek update PRs - env: - BODY_MARKER: Automated update of `prek` hooks. - GITHUB_TOKEN: ${{ github.token }} - REPOSITORY: ${{ github.repository }} - run: | - set -euo pipefail - - python .github/scripts/cleanup_prek_update_branches.py \ - --repository "$REPOSITORY" \ - --branch chore/prek-updates \ - --branch-prefix chore/prek-updates \ - --label-name dependencies \ - --author-login 'github-actions[bot]' \ - --body-marker "$BODY_MARKER" \ - --keep-latest-open-pr \ - --close-stale-prs \ - --delete-stale-branches \ - --delete-merged-branches + prek-autoupdate: + uses: Snuffy2/prek-autoupdate/.github/workflows/prek_autoupdate.yml@v1 + permissions: + contents: write + pull-requests: write + actions: write diff --git a/tests/test_prek_autoupdate_workflow.py b/tests/test_prek_autoupdate_workflow.py deleted file mode 100644 index bef912c2..00000000 --- a/tests/test_prek_autoupdate_workflow.py +++ /dev/null @@ -1,576 +0,0 @@ -"""Tests for the prek autoupdate workflow.""" - -from collections.abc import Generator -from importlib import util -from io import BytesIO -from pathlib import Path -import re -import sys -from types import ModuleType -from urllib.error import HTTPError - -import pytest - -REPO_ROOT = Path(__file__).resolve().parents[1] -WORKFLOW_SCRIPT_PATH = ".github/scripts/cleanup_prek_update_branches.py" -WORKFLOW_PATH = REPO_ROOT / ".github/workflows/prek_autoupdate.yml" -CLEANUP_SCRIPT_PATH = REPO_ROOT / WORKFLOW_SCRIPT_PATH -WORKFLOW_BRANCH = "chore/prek-updates" -WORKFLOW_LABEL = "dependencies" -WORKFLOW_AUTHOR = "github-actions[bot]" -WORKFLOW_BODY_MARKER = "Automated update of `prek` hooks." -REPOSITORY = "o/r" - - -class FakeCleanupClient: - """Fake GitHub cleanup client that records mutating calls.""" - - def __init__( - self, - *, - open_pulls: list[dict[str, object]], - closed_pulls: list[dict[str, object]], - branch_refs: list[str] | None = None, - fail_on_close: bool = False, - ) -> None: - """Initialize fake pull request state. - - Args: - open_pulls: Pull requests to return for open PR lookups. - closed_pulls: Pull requests to return for closed PR lookups. - branch_refs: Branch refs to return for matching branch lookups. - fail_on_close: Whether closing a PR should fail the test. - """ - self.open_pulls = open_pulls - self.closed_pulls = closed_pulls - self.branch_refs = [] if branch_refs is None else branch_refs - self.fail_on_close = fail_on_close - self.closed_prs: list[int] = [] - self.deleted_refs: list[str] = [] - self.max_pages_by_state: dict[str, int | None] = {} - - def list_pulls(self, *, state: str, max_pages: int | None = None) -> list[dict[str, object]]: - """Return fake pull requests by state.""" - self.max_pages_by_state[state] = max_pages - return self.open_pulls if state == "open" else self.closed_pulls - - def list_refs(self, *, ref_prefix: str) -> list[str]: - """Return fake git refs by prefix.""" - return [ref for ref in self.branch_refs if ref.startswith(f"refs/{ref_prefix}")] - - def close_pull(self, pull_number: int) -> None: - """Record or reject a closed pull request.""" - if self.fail_on_close: - raise AssertionError(f"Unexpected close for PR {pull_number}") - self.closed_prs.append(pull_number) - - def delete_ref(self, ref: str) -> None: - """Record a deleted git ref.""" - self.deleted_refs.append(ref) - - -def _workflow_pull( - *, - number: int, - ref: str = WORKFLOW_BRANCH, - label: str = WORKFLOW_LABEL, - author: str = WORKFLOW_AUTHOR, - body: str = WORKFLOW_BODY_MARKER, - repository: str = REPOSITORY, - merged_at: str | None = None, -) -> dict[str, object]: - """Return a fake workflow pull request object. - - Args: - number: Pull request number. - ref: Pull request head ref. - label: Pull request label name. - author: Pull request author login. - body: Pull request body text. - repository: Head repository full name. - merged_at: Optional merge timestamp for closed PRs. - - Returns: - Fake pull request object shaped like the GitHub REST API response. - """ - return { - "number": number, - "merged_at": merged_at, - "body": body, - "user": {"login": author}, - "head": {"ref": ref, "repo": {"full_name": repository}}, - "labels": [{"name": label}], - } - - -@pytest.fixture -def cleanup_script() -> Generator[ModuleType]: - """Load the prek cleanup script as a test module.""" - spec = util.spec_from_file_location("cleanup_prek_update_branches", CLEANUP_SCRIPT_PATH) - assert spec is not None - assert spec.loader is not None - module = util.module_from_spec(spec) - previous_module = sys.modules.get("cleanup_prek_update_branches") - sys.modules["cleanup_prek_update_branches"] = module - spec.loader.exec_module(module) - try: - yield module - finally: - if previous_module is None: - sys.modules.pop("cleanup_prek_update_branches", None) - else: - sys.modules["cleanup_prek_update_branches"] = previous_module - - -@pytest.mark.parametrize( - ("needle", "reason"), - [ - ("actions/setup-python@v6", "pins a Python runtime for cleanup"), - ("python-version: '3.14'", "uses the same runtime as local tooling"), - ("persist-credentials: false", "disables persisted checkout credentials"), - (WORKFLOW_SCRIPT_PATH, "runs the checked-in cleanup helper"), - ("Close extra auto-generated PRs", "closes duplicate generated PRs"), - ("Close stale prek update PRs", "closes stale PRs when no diff remains"), - (f"--branch {WORKFLOW_BRANCH}", "uses the workflow update branch"), - ( - f"--branch-prefix {WORKFLOW_BRANCH}", - "limits cleanup to prek workflow branches", - ), - (f"--label-name {WORKFLOW_LABEL}", "limits cleanup to workflow-labeled PRs"), - ( - f"--author-login '{WORKFLOW_AUTHOR}'", - "limits cleanup to workflow-authored PRs", - ), - ( - "BODY_MARKER: Automated update of `prek` hooks.", - "exports the workflow body marker outside shell expansion", - ), - ( - '--body-marker "$BODY_MARKER"', - "limits cleanup to PRs generated by this workflow", - ), - ("--keep-latest-open-pr", "preserves the current update PR during nightly cleanup"), - ("--delete-stale-branches", "deletes orphaned workflow branches"), - ("--delete-merged-branches", "cleans merged workflow-owned branches"), - ("REPOSITORY: ${{ github.repository }}", "exports repository before shell use"), - ('--repository "$REPOSITORY"', "avoids inline repository template expansion"), - ], -) -def test_workflow_contains_expected_cleanup_logic(needle: str, reason: str) -> None: - """Workflow should include standalone cleanup logic for generated prek PRs.""" - del reason - - assert needle in WORKFLOW_PATH.read_text() - - -def test_workflow_avoids_inline_repository_template_expansion() -> None: - """Workflow should not expand the repository context inside shell scripts.""" - assert '--repository "${{ github.repository }}"' not in WORKFLOW_PATH.read_text() - - -def test_workflow_uses_plural_stale_branch_cleanup_only() -> None: - """Workflow should use prefix branch cleanup instead of fixed branch cleanup.""" - workflow_text = WORKFLOW_PATH.read_text() - - assert "--delete-stale-branches" in workflow_text - assert re.search(r"--delete-stale-branch(?!es)\b", workflow_text) is None - - -def test_workflow_runs_updates_weekly_and_cleanup_daily() -> None: - """Workflow should only update prek weekly while running cleanup every night.""" - workflow_text = WORKFLOW_PATH.read_text() - - assert "cron: '0 2 * * *'" in workflow_text - assert "date -u +%u" in workflow_text - assert "github.event.schedule == '0 2 * * 1'" not in workflow_text - assert "should_update=true" in workflow_text - assert "name: Nightly Cleanup" in workflow_text - - -def test_cleanup_script_closes_stale_prs_and_deletes_workflow_branches( - cleanup_script: ModuleType, -) -> None: - """Cleanup script should close stale PRs and remove workflow-created branches.""" - client = FakeCleanupClient( - open_pulls=[ - _workflow_pull(number=10), - _workflow_pull(number=9, ref=f"{WORKFLOW_BRANCH}-old"), - _workflow_pull(number=11, ref="feature/manual"), - ], - closed_pulls=[ - _workflow_pull(number=8, merged_at="2026-05-28T00:00:00Z"), - _workflow_pull(number=7, ref=f"{WORKFLOW_BRANCH}-old"), - ], - ) - - result = cleanup_script.cleanup_update_branches( - client=client, - repository=REPOSITORY, - branch=WORKFLOW_BRANCH, - branch_prefix=WORKFLOW_BRANCH, - label_name=WORKFLOW_LABEL, - author_login=WORKFLOW_AUTHOR, - body_marker=WORKFLOW_BODY_MARKER, - keep_pr_number=None, - close_stale_prs=True, - delete_merged_branches=True, - ) - - assert client.closed_prs == [10, 9] - assert client.deleted_refs == [ - f"heads/{WORKFLOW_BRANCH}", - f"heads/{WORKFLOW_BRANCH}-old", - ] - assert client.max_pages_by_state == { - "open": None, - "closed": cleanup_script.CLOSED_PULL_PAGE_LIMIT, - } - assert result.closed_prs == [10, 9] - assert result.deleted_branches == [ - WORKFLOW_BRANCH, - f"{WORKFLOW_BRANCH}-old", - ] - - -def test_cleanup_script_keeps_active_update_branch(cleanup_script: ModuleType) -> None: - """Cleanup script should not delete the branch for the kept update PR.""" - client = FakeCleanupClient( - open_pulls=[_workflow_pull(number=12)], - closed_pulls=[ - _workflow_pull(number=8, merged_at="2026-05-28T00:00:00Z"), - _workflow_pull( - number=6, - ref=f"{WORKFLOW_BRANCH}-old", - merged_at="2026-05-20T00:00:00Z", - ), - ], - fail_on_close=True, - ) - - result = cleanup_script.cleanup_update_branches( - client=client, - repository=REPOSITORY, - branch=WORKFLOW_BRANCH, - branch_prefix=WORKFLOW_BRANCH, - label_name=WORKFLOW_LABEL, - author_login=WORKFLOW_AUTHOR, - body_marker=WORKFLOW_BODY_MARKER, - keep_pr_number=12, - close_stale_prs=True, - delete_merged_branches=True, - ) - - assert client.deleted_refs == [f"heads/{WORKFLOW_BRANCH}-old"] - assert result.closed_prs == [] - assert result.deleted_branches == [f"{WORKFLOW_BRANCH}-old"] - - -def test_cleanup_script_can_keep_latest_open_workflow_pr(cleanup_script: ModuleType) -> None: - """Cleanup script should preserve the newest open PR during nightly cleanup.""" - client = FakeCleanupClient( - open_pulls=[ - _workflow_pull(number=17, ref=f"{WORKFLOW_BRANCH}-old"), - _workflow_pull(number=18), - ], - closed_pulls=[ - _workflow_pull( - number=16, ref=f"{WORKFLOW_BRANCH}-merged", merged_at="2026-05-28T00:00:00Z" - ), - ], - ) - - result = cleanup_script.cleanup_update_branches( - client=client, - repository=REPOSITORY, - branch=WORKFLOW_BRANCH, - branch_prefix=WORKFLOW_BRANCH, - label_name=WORKFLOW_LABEL, - author_login=WORKFLOW_AUTHOR, - body_marker=WORKFLOW_BODY_MARKER, - keep_pr_number=None, - keep_latest_open_pr=True, - close_stale_prs=True, - delete_merged_branches=True, - ) - - assert client.closed_prs == [17] - assert client.deleted_refs == [ - f"heads/{WORKFLOW_BRANCH}-merged", - f"heads/{WORKFLOW_BRANCH}-old", - ] - assert result.closed_prs == [17] - assert result.deleted_branches == [ - f"{WORKFLOW_BRANCH}-merged", - f"{WORKFLOW_BRANCH}-old", - ] - - -def test_cleanup_script_deletes_orphaned_workflow_branches(cleanup_script: ModuleType) -> None: - """Cleanup script should delete prefixed workflow branches without open PRs.""" - client = FakeCleanupClient( - open_pulls=[_workflow_pull(number=18)], - closed_pulls=[], - branch_refs=[ - f"refs/heads/{WORKFLOW_BRANCH}", - f"refs/heads/{WORKFLOW_BRANCH}-orphan", - f"refs/heads/{WORKFLOW_BRANCH}-manual", - ], - ) - - result = cleanup_script.cleanup_update_branches( - client=client, - repository=REPOSITORY, - branch=WORKFLOW_BRANCH, - branch_prefix=WORKFLOW_BRANCH, - label_name=WORKFLOW_LABEL, - author_login=WORKFLOW_AUTHOR, - body_marker=WORKFLOW_BODY_MARKER, - keep_pr_number=None, - keep_latest_open_pr=True, - close_stale_prs=True, - delete_stale_branches=True, - delete_merged_branches=False, - ) - - assert client.deleted_refs == [ - f"heads/{WORKFLOW_BRANCH}-manual", - f"heads/{WORKFLOW_BRANCH}-orphan", - ] - assert result.deleted_branches == [ - f"{WORKFLOW_BRANCH}-manual", - f"{WORKFLOW_BRANCH}-orphan", - ] - - -def test_cleanup_script_protects_open_workflow_prs_when_not_closing_stale_prs( - cleanup_script: ModuleType, -) -> None: - """Cleanup script should not sweep open workflow branches when not closing PRs.""" - client = FakeCleanupClient( - open_pulls=[ - _workflow_pull(number=18), - _workflow_pull(number=17, ref=f"{WORKFLOW_BRANCH}-old"), - ], - closed_pulls=[], - branch_refs=[ - f"refs/heads/{WORKFLOW_BRANCH}", - f"refs/heads/{WORKFLOW_BRANCH}-old", - f"refs/heads/{WORKFLOW_BRANCH}-orphan", - ], - ) - - result = cleanup_script.cleanup_update_branches( - client=client, - repository=REPOSITORY, - branch=WORKFLOW_BRANCH, - branch_prefix=WORKFLOW_BRANCH, - label_name=WORKFLOW_LABEL, - author_login=WORKFLOW_AUTHOR, - body_marker=WORKFLOW_BODY_MARKER, - keep_pr_number=None, - keep_latest_open_pr=False, - close_stale_prs=False, - delete_stale_branches=True, - delete_merged_branches=False, - ) - - assert client.closed_prs == [] - assert client.deleted_refs == [f"heads/{WORKFLOW_BRANCH}-orphan"] - assert result.deleted_branches == [f"{WORKFLOW_BRANCH}-orphan"] - - -def test_cleanup_script_closes_stale_canonical_branch_when_newer_pr_exists( - cleanup_script: ModuleType, -) -> None: - """Cleanup script should keep only the latest generated workflow PR.""" - client = FakeCleanupClient( - open_pulls=[ - _workflow_pull(number=17), - _workflow_pull(number=18, ref=f"{WORKFLOW_BRANCH}-newer"), - ], - closed_pulls=[], - branch_refs=[ - f"refs/heads/{WORKFLOW_BRANCH}", - f"refs/heads/{WORKFLOW_BRANCH}-newer", - ], - ) - - result = cleanup_script.cleanup_update_branches( - client=client, - repository=REPOSITORY, - branch=WORKFLOW_BRANCH, - branch_prefix=WORKFLOW_BRANCH, - label_name=WORKFLOW_LABEL, - author_login=WORKFLOW_AUTHOR, - body_marker=WORKFLOW_BODY_MARKER, - keep_pr_number=None, - keep_latest_open_pr=True, - close_stale_prs=True, - delete_stale_branches=True, - delete_merged_branches=False, - ) - - assert client.closed_prs == [17] - assert client.deleted_refs == [f"heads/{WORKFLOW_BRANCH}"] - assert result.closed_prs == [17] - assert result.deleted_branches == [WORKFLOW_BRANCH] - - -def test_cleanup_script_preserves_open_non_workflow_pr_branches( - cleanup_script: ModuleType, -) -> None: - """Cleanup script should not delete branches used by open non-workflow PRs.""" - client = FakeCleanupClient( - open_pulls=[ - _workflow_pull(number=18), - _workflow_pull( - number=19, - ref=f"{WORKFLOW_BRANCH}-manual-fix", - author="maintainer", - body="Manual maintenance PR.", - ), - _workflow_pull( - number=20, - ref=f"{WORKFLOW_BRANCH}-external", - author="maintainer", - body="External fork PR.", - repository="fork/r", - ), - ], - closed_pulls=[], - branch_refs=[ - f"refs/heads/{WORKFLOW_BRANCH}", - f"refs/heads/{WORKFLOW_BRANCH}-manual-fix", - f"refs/heads/{WORKFLOW_BRANCH}-orphan", - ], - ) - - result = cleanup_script.cleanup_update_branches( - client=client, - repository=REPOSITORY, - branch=WORKFLOW_BRANCH, - branch_prefix=WORKFLOW_BRANCH, - label_name=WORKFLOW_LABEL, - author_login=WORKFLOW_AUTHOR, - body_marker=WORKFLOW_BODY_MARKER, - keep_pr_number=None, - keep_latest_open_pr=True, - close_stale_prs=True, - delete_stale_branches=True, - delete_merged_branches=False, - ) - - assert client.deleted_refs == [f"heads/{WORKFLOW_BRANCH}-orphan"] - assert result.deleted_branches == [f"{WORKFLOW_BRANCH}-orphan"] - - -def test_cleanup_script_preserves_human_prs_with_matching_label_and_prefix( - cleanup_script: ModuleType, -) -> None: - """Cleanup script should not mutate human PRs that share labels and prefixes.""" - client = FakeCleanupClient( - open_pulls=[ - _workflow_pull( - number=13, - ref=f"{WORKFLOW_BRANCH}-manual-fix", - author="maintainer", - ), - ], - closed_pulls=[ - _workflow_pull( - number=14, - ref=f"{WORKFLOW_BRANCH}-manual-merged", - author="maintainer", - merged_at="2026-05-29T00:00:00Z", - ), - ], - ) - - result = cleanup_script.cleanup_update_branches( - client=client, - repository=REPOSITORY, - branch=WORKFLOW_BRANCH, - branch_prefix=WORKFLOW_BRANCH, - label_name=WORKFLOW_LABEL, - author_login=WORKFLOW_AUTHOR, - body_marker=WORKFLOW_BODY_MARKER, - keep_pr_number=None, - close_stale_prs=True, - delete_merged_branches=True, - ) - - assert client.closed_prs == [] - assert client.deleted_refs == [] - assert result.closed_prs == [] - assert result.deleted_branches == [] - - -def test_cleanup_script_preserves_bot_prs_without_workflow_body_marker( - cleanup_script: ModuleType, -) -> None: - """Cleanup script should not mutate bot PRs without the workflow body marker.""" - client = FakeCleanupClient( - open_pulls=[ - _workflow_pull( - number=15, - ref=f"{WORKFLOW_BRANCH}-v2", - body="Automated dependency update from another workflow.", - ), - ], - closed_pulls=[ - _workflow_pull( - number=16, - ref=f"{WORKFLOW_BRANCH}-docs", - body="Automated dependency update from another workflow.", - merged_at="2026-05-29T00:00:00Z", - ), - ], - ) - - result = cleanup_script.cleanup_update_branches( - client=client, - repository=REPOSITORY, - branch=WORKFLOW_BRANCH, - branch_prefix=WORKFLOW_BRANCH, - label_name=WORKFLOW_LABEL, - author_login=WORKFLOW_AUTHOR, - body_marker=WORKFLOW_BODY_MARKER, - keep_pr_number=None, - close_stale_prs=True, - delete_merged_branches=True, - ) - - assert client.closed_prs == [] - assert client.deleted_refs == [] - assert result.closed_prs == [] - assert result.deleted_branches == [] - - -def test_github_headers_include_json_content_type(cleanup_script: ModuleType) -> None: - """GitHub client headers should describe JSON request bodies correctly.""" - headers = cleanup_script._github_headers("token") - - assert headers["Content-Type"] == "application/json" - - -def test_github_client_list_refs_treats_missing_prefix_as_empty( - cleanup_script: ModuleType, - monkeypatch: pytest.MonkeyPatch, -) -> None: - """GitHub client should treat a missing matching-ref prefix as no refs.""" - - def fake_urlopen(*_args: object, **_kwargs: object) -> object: - """Raise a GitHub-style 404 response for missing refs.""" - raise HTTPError( - url="https://api.github.test/repos/o/r/git/matching-refs/heads/missing", - code=404, - msg="Not Found", - hdrs={}, - fp=BytesIO(b'{"message": "Not Found"}'), - ) - - monkeypatch.setattr(cleanup_script, "urlopen", fake_urlopen) - - client = cleanup_script.GithubClient(repository=REPOSITORY, token="token") # noqa: S106 - - assert client.list_refs(ref_prefix="heads/missing") == []