ClickHouse/tests/ci/rerun_helper.py
2022-03-22 17:39:58 +01:00

39 lines
1.1 KiB
Python

#!/usr/bin/env python3
from commit_status_helper import get_commit
def _filter_statuses(statuses):
"""
Squash statuses to latest state
1. context="first", state="success", update_time=1
2. context="second", state="success", update_time=2
3. context="first", stat="failure", update_time=3
=========>
1. context="second", state="success"
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",
):
return True
return False