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
|
from __future__ import annotations
from typing import Any
from unittest import main, TestCase
from tools.alerts.create_alerts import filter_job_names, JobStatus
JOB_NAME = "periodic / linux-xenial-cuda10.2-py3-gcc7-slow-gradcheck / test (default, 2, 2, linux.4xlarge.nvidia.gpu)"
MOCK_TEST_DATA = [
{
"sha": "f02f3046571d21b48af3067e308a1e0f29b43af9",
"id": 7819529276,
"conclusion": "failure",
"htmlUrl": "https://github.com/pytorch/pytorch/runs/7819529276?check_suite_focus=true",
"logUrl": "https://ossci-raw-job-status.s3.amazonaws.com/log/7819529276",
"durationS": 14876,
"failureLine": "##[error]The action has timed out.",
"failureContext": "",
"failureCaptures": ["##[error]The action has timed out."],
"failureLineNumber": 83818,
"repo": "pytorch/pytorch",
},
{
"sha": "d0d6b1f2222bf90f478796d84a525869898f55b6",
"id": 7818399623,
"conclusion": "failure",
"htmlUrl": "https://github.com/pytorch/pytorch/runs/7818399623?check_suite_focus=true",
"logUrl": "https://ossci-raw-job-status.s3.amazonaws.com/log/7818399623",
"durationS": 14882,
"failureLine": "##[error]The action has timed out.",
"failureContext": "",
"failureCaptures": ["##[error]The action has timed out."],
"failureLineNumber": 72821,
"repo": "pytorch/pytorch",
},
]
class TestGitHubPR(TestCase):
# Should fail when jobs are ? ? Fail Fail
def test_alert(self) -> None:
modified_data: list[Any] = [{}]
modified_data.append({})
modified_data.extend(MOCK_TEST_DATA)
status = JobStatus(JOB_NAME, modified_data)
self.assertTrue(status.should_alert())
# test filter job names
def test_job_filter(self) -> None:
job_names = [
"pytorch_linux_xenial_py3_6_gcc5_4_test",
"pytorch_linux_xenial_py3_6_gcc5_4_test2",
]
self.assertListEqual(
filter_job_names(job_names, ""),
job_names,
"empty regex should match all jobs",
)
self.assertListEqual(filter_job_names(job_names, ".*"), job_names)
self.assertListEqual(filter_job_names(job_names, ".*xenial.*"), job_names)
self.assertListEqual(
filter_job_names(job_names, ".*xenial.*test2"),
["pytorch_linux_xenial_py3_6_gcc5_4_test2"],
)
self.assertListEqual(filter_job_names(job_names, ".*xenial.*test3"), [])
self.assertRaises(
Exception,
lambda: filter_job_names(job_names, "["),
msg="malformed regex should throw exception",
)
if __name__ == "__main__":
main()
|
