Repository: ClickHouse/ClickHouse
File: tests/ci/cherry_pick.py (Python)

Code before:
                    self.pr.head.ref,
                )
        original_pr = f"Original pull-request {self.pr.html_url}\n"
        cherrypick_pr = ""
        if self.cherrypick_pr is not None:
            cherrypick_pr = f"Cherry-pick pull-request {self.cherrypick_pr.html_url}\n"
        return f"{upstream_pr}{original_pr}{cherrypick_pr}\n"

    def ping_cherry_pick_assignees(self, dry_run: bool) -> None:
        assert self.cherrypick_pr is not None
        logging.info(
            "Checking if cherry-pick PR #%s needs to be pinged or closed",
            self.cherrypick_pr.number,
        )
        # The `updated_at` is Optional[datetime]
        cherrypick_updated_ts = (
            self.cherrypick_pr.updated_at or datetime.now()
        ).timestamp()
        since_updated = int(datetime.now().timestamp() - cherrypick_updated_ts)
        since_updated_str = (
            f"{since_updated // 86400}d{since_updated // 3600 % 24}h"
            f"{since_updated // 60 % 60}m{since_updated % 60}s"
        )

        PING_THRESHOLD = 3 * 24 * 3600  # 3 days
        CLOSE_THRESHOLD = 7 * 24 * 3600  # 7 days

        if since_updated < PING_THRESHOLD:
            logging.info(
                "The cherry-pick PR was updated %s ago, waiting for the next run",
                since_updated_str,
            )
            return

        if since_updated >= CLOSE_THRESHOLD:
            # Close the PR after 7 days
            assignees = ", ".join(
                f"@{user.login}" for user in self.cherrypick_pr.assignees
            )
            comment_body = (
                f"Dear {assignees}, this cherry-pick PR has not been updated for {since_updated_str}. "
                f"Closing automatically. If you still want to backport #{self.pr.number}, "
                "please resolve the conflicts and reopen this PR."
            )
            if dry_run:
                logging.info(
                    "DRY RUN: would close cherry-pick PR #%s with comment:\n%s",
                    self.cherrypick_pr.number,
                    comment_body,
                )
                return

Review comment (suggestion):
These thresholds are new policy constants but are defined as local magic numbers inside the method. Consider moving them to `ReleaseBranch` class constants (or a config/env-driven setting) so they’re easier to discover/change and remain consistent if reused elsewhere (e.g., logs/messages/tests).

Code after:
                    self.pr.head.ref,
                )
        original_pr = f"Original pull-request {self.pr.html_url}\n"
        cherrypick_pr = ""
        if self.cherrypick_pr is not None:
            cherrypick_pr = f"Cherry-pick pull-request {self.cherrypick_pr.html_url}\n"
        return f"{upstream_pr}{original_pr}{cherrypick_pr}\n"

    def ping_cherry_pick_assignees(self, dry_run: bool) -> None:
        assert self.cherrypick_pr is not None
        logging.info(
            "Checking if cherry-pick PR #%s needs to be pinged or closed",
            self.cherrypick_pr.number,
        )
        # The `updated_at` is Optional[datetime]
        cherrypick_updated_ts = (
            self.cherrypick_pr.updated_at or datetime.now()
        ).timestamp()
        since_updated = int(datetime.now().timestamp() - cherrypick_updated_ts)
        since_updated_str = (
            f"{since_updated // 86400}d{since_updated // 3600 % 24}h"
            f"{since_updated // 60 % 60}m{since_updated % 60}s"
        )

        PING_THRESHOLD = 3 * 24 * 3600  # 3 days
        CLOSE_THRESHOLD = 7 * 24 * 3600  # 7 days

        if since_updated < PING_THRESHOLD:
            logging.info(
                "The cherry-pick PR was updated %s ago, waiting for the next run",
                since_updated_str,
            )
            return

        if since_updated >= CLOSE_THRESHOLD:
            # Close the PR after 7 days
            if self.cherrypick_pr.assignees:
                assignees = ", ".join(
                    f"@{user.login}" for user in self.cherrypick_pr.assignees
                )
                comment_body = (
                    f"Dear {assignees}, this cherry-pick PR has not been updated for {since_updated_str}. "
                    f"Closing automatically. If you still want to backport #{self.pr.number}, "
                    "please resolve the conflicts and reopen this PR."
                )
            else:
                logging.warning(
                    "Cherry-pick PR #%s has no assignees when closing",
                    self.cherrypick_pr.number,
                )
                comment_body = (