1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
|
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
)
|
