Added a function that prints the check statuses run on a given commit SHA (#78663)

Relates to #76700

Gets the commit SHAs from the past M minutes, and prints out the SHAs along with the status checks for all of the jobs for each commit. The current output prints out each SHA and a list of all of the workflow job conclusions.

Example output:
![Screen Shot 2022-06-01 at 4 51 07 PM](https://user-images.githubusercontent.com/24441980/171499216-59f6d2f2-01b3-4d01-a7ae-5215b4ac4e5c.png)

**Test Plan:** compare output with HUD

Pull Request resolved: https://github.com/pytorch/pytorch/pull/78663
Approved by: https://github.com/seemethere
This commit is contained in:
swang392
2022-06-02 15:05:14 +00:00
committed by PyTorch MergeBot
parent 4a5381ab40
commit 7fc73285da

View File

@ -2,29 +2,45 @@ from typing import Any
from datetime import datetime, timedelta
from gitutils import _check_output
from rockset import Client, ParamDict # type: ignore[import]
import os
rs = Client(api_key=os.getenv("ROCKSET_API_KEY", None))
qlambda = rs.QueryLambda.retrieve(
'commit_jobs_query',
version='c2a4dbce081d0144',
workspace='commons')
def parse_args() -> Any:
from argparse import ArgumentParser
parser = ArgumentParser("Print latest commits")
parser.add_argument("--minutes", type=int, default=60, help="duration in minutes of last commits")
parser.add_argument("--minutes", type=int, default=30, help="duration in minutes of last commits")
return parser.parse_args()
def print_latest_commits(minutes: int = 60) -> None:
def print_latest_commits(minutes: int = 30) -> None:
current_time = datetime.now()
time_since = current_time - timedelta(minutes=minutes)
timestamp_since = datetime.timestamp(time_since)
commits = _check_output(
[
"git",
"rev-list",
f"--max-age={timestamp_since}",
"--all",
"--remotes=*origin/master",
],
encoding="ascii",
).splitlines()
for commit in commits:
print(commit)
print_commit_status(commit)
def print_commit_status(sha: str) -> None:
params = ParamDict()
params['sha'] = sha
results = qlambda.execute(parameters=params)
for check in results['results']:
print(f"\t{check['conclusion']:>10}: {check['name']}")
def main() -> None:
args = parse_args()