Add typing to rerun and commit_status helpers

This commit is contained in:
Mikhail f. Shiryaev 2022-07-21 18:12:07 +02:00
parent 142f7d4b44
commit db0653758e
2 changed files with 40 additions and 24 deletions

View File

@ -1,10 +1,14 @@
#!/usr/bin/env python3
import time
import os
import csv
from env_helper import GITHUB_REPOSITORY, GITHUB_RUN_URL
import os
import time
from typing import Optional
from ci_config import CI_CONFIG
from env_helper import GITHUB_REPOSITORY, GITHUB_RUN_URL
from github import Github
from github.Commit import Commit
from pr_info import SKIP_SIMPLE_CHECK_LABEL
RETRY = 5
@ -22,7 +26,9 @@ def override_status(status, check_name, invert=False):
return status
def get_commit(gh, commit_sha, retry_count=RETRY):
def get_commit(
gh: Github, commit_sha: str, retry_count: int = RETRY
) -> Optional[Commit]:
for i in range(retry_count):
try:
repo = gh.get_repo(GITHUB_REPOSITORY)

View File

@ -1,9 +1,42 @@
#!/usr/bin/env python3
from typing import List, Optional
from commit_status_helper import get_commit
from github import Github
from github.CommitStatus import CommitStatus
from pr_info import PRInfo
CommitStatuses = List[CommitStatus]
def _filter_statuses(statuses):
class RerunHelper:
def __init__(self, gh: Github, pr_info: PRInfo, check_name: str):
self.gh = gh
self.pr_info = pr_info
self.check_name = check_name
commit = get_commit(gh, self.pr_info.sha)
if commit is None:
raise ValueError(f"unable to receive commit for {pr_info.sha}")
self.pygh_commit = commit
self.statuses = self.ger_filtered_statuses()
def is_already_finished_by_status(self) -> bool:
# currently we agree even for failed statuses
for status in self.statuses:
if self.check_name in status.context and status.state in (
"success",
"failure",
):
return True
return False
def get_finished_status(self) -> Optional[CommitStatus]:
for status in self.statuses:
if self.check_name in status.context:
return status
return None
def ger_filtered_statuses(self) -> CommitStatuses:
"""
Squash statuses to latest state
1. context="first", state="success", update_time=1
@ -14,31 +47,8 @@ def _filter_statuses(statuses):
2. context="first", stat="failure"
"""
filt = {}
for status in sorted(statuses, key=lambda x: x.updated_at):
filt[status.context] = status
return filt.values()
class RerunHelper:
def __init__(self, gh, pr_info, check_name):
self.gh = gh
self.pr_info = pr_info
self.check_name = check_name
self.pygh_commit = get_commit(gh, self.pr_info.sha)
self.statuses = _filter_statuses(self.pygh_commit.get_statuses())
def is_already_finished_by_status(self):
# currently we agree even for failed statuses
for status in self.statuses:
if self.check_name in status.context and status.state in (
"success",
"failure",
for status in sorted(
self.pygh_commit.get_statuses(), key=lambda x: x.updated_at
):
return True
return False
def get_finished_status(self):
for status in self.statuses:
if self.check_name in status.context:
return status
return None
filt[status.context] = status
return list(filt.values())