| import os |
| import re |
| from typing import List, Pattern, Tuple, Optional |
| |
| |
| BOT_COMMANDS_WIKI = "https://github.com/pytorch/pytorch/wiki/Bot-commands" |
| |
| CIFLOW_LABEL = re.compile(r"^ciflow/.+") |
| CIFLOW_TRUNK_LABEL = re.compile(r"^ciflow/trunk") |
| |
| OFFICE_HOURS_LINK = "https://github.com/pytorch/pytorch/wiki/Dev-Infra-Office-Hours" |
| CONTACT_US = f"Please reach out to the [PyTorch DevX Team]({OFFICE_HOURS_LINK}) with feedback or questions!" |
| ALTERNATIVES = ( |
| "If this is not the intended behavior, feel free to use some " |
| + f"of the other merge options in the [wiki]({BOT_COMMANDS_WIKI})." |
| ) |
| LAND_CHECK_ROLLOUT = "https://github.com/pytorch/test-infra/blob/main/torchci/lib/bot/rolloutUtils.ts#L1-L34" |
| |
| |
| def has_label(labels: List[str], pattern: Pattern[str] = CIFLOW_LABEL) -> bool: |
| return len(list(filter(pattern.match, labels))) > 0 |
| |
| |
| class TryMergeExplainer(object): |
| force: bool |
| on_green: bool |
| land_checks: bool |
| labels: List[str] |
| pr_num: int |
| org: str |
| project: str |
| |
| has_trunk_label: bool |
| has_ciflow_label: bool |
| |
| def __init__( |
| self, |
| force: bool, |
| on_green: bool, |
| land_checks: bool, |
| labels: List[str], |
| pr_num: int, |
| org: str, |
| project: str, |
| ): |
| self.force = force |
| self.on_green = on_green |
| self.land_checks = land_checks |
| self.labels = labels |
| self.pr_num = pr_num |
| self.org = org |
| self.project = project |
| self.get_flags() |
| |
| def get_flags(self) -> Tuple[bool, bool]: |
| self.has_trunk_label = has_label(self.labels, CIFLOW_TRUNK_LABEL) |
| self.has_ciflow_label = has_label(self.labels, CIFLOW_LABEL) |
| should_check_land_branch = self.land_checks and not self.has_trunk_label |
| should_check_green = self.on_green or self.has_ciflow_label |
| |
| return (should_check_green, should_check_land_branch) |
| |
| def _get_flag_msg(self) -> str: |
| if self.force: |
| return " the force (-f) flag." |
| elif self.on_green: |
| return " the green (-g) flag." |
| elif self.land_checks: |
| return ( |
| " the land checks (-l) flag." |
| + " If you did not specify this flag yourself, " |
| + f" you are likely enrolled in the [land checks rollout]({LAND_CHECK_ROLLOUT})." |
| ) |
| else: |
| return "out a flag." |
| |
| def _get_land_check_progress(self, commit: Optional[str]) -> str: |
| if commit is not None: |
| return ( |
| " and land check " |
| + f"progress [here](https://hud.pytorch.org/{self.org}/{self.project}/commit/{commit})" |
| ) |
| else: |
| return "" |
| |
| def _get_flag_explanation_message(self) -> str: |
| if self.force: |
| return "This means your change will be merged **immediately**, bypassing any CI checks (ETA: 1-5 minutes)." |
| elif self.on_green: |
| return "This means that your change will be merged once all checks on your PR have passed (ETA: 0-4 Hours)." |
| elif self.land_checks: |
| if self.has_trunk_label: |
| land_check_msg_suffix = "have passed since you have added the `ciflow/trunk` label to your PR (ETA 0-4 Hours)." |
| else: |
| land_check_msg_suffix = ( |
| "and the land checks have passed (**ETA 4 Hours**). " |
| ) |
| land_check_msg_suffix += "If you need to coordinate lands between different changes and cannot risk a land race, " |
| land_check_msg_suffix += "please add the `ciflow/trunk` label to your PR and wait for signal to complete, " |
| land_check_msg_suffix += "and then land your changes in proper order." |
| land_check_msg_suffix += ( |
| " Having `trunk`, `pull`, and `Lint` pre-run on a " |
| ) |
| land_check_msg_suffix += ( |
| "PR will bypass land checks and the ETA should be immediate." |
| ) |
| |
| return ( |
| "This means that your change will be merged once all checks on your PR " |
| + land_check_msg_suffix |
| ) |
| else: |
| return "This means that your change will be merged once all checks on your PR have passed (ETA: 0-4 Hours)." |
| |
| def get_merge_message(self, commit: Optional[str] = None) -> str: |
| message_prefix = "@pytorchbot successfully started a merge job." |
| progress_links = f"Check the current status [here]({os.getenv('GH_RUN_URL')}){self._get_land_check_progress(commit)}." |
| flag_message = f"The merge job was triggered with{self._get_flag_msg()}" |
| explanation_message = self._get_flag_explanation_message() |
| |
| msg = message_prefix + " " |
| msg += progress_links + "\n" |
| msg += flag_message + " " |
| msg += explanation_message + " " |
| msg += ALTERNATIVES + "\n" |
| msg += CONTACT_US |
| return msg |
| |
| |
| def get_revert_message(org: str, project: str, pr_num: int) -> str: |
| msg = ( |
| "@pytorchbot successfully started a revert job." |
| + f" Check the current status [here]({os.getenv('GH_RUN_URL')}).\n" |
| ) |
| msg += CONTACT_US |
| return msg |
| |
| |
| def get_land_check_troubleshooting_message() -> str: |
| return ( |
| " If you believe this is an error, you can use the old behavior with `@pytorchbot merge -g`" |
| + " (optionally with the `ciflow/trunk` to get land checks)" |
| + ' or use `@pytorchbot merge -f "some reason here"`.' |
| + f" For more information, see the [bot wiki]({BOT_COMMANDS_WIKI}). \n\n" |
| + CONTACT_US |
| ) |