Source code
Revision control
Copy as Markdown
Other Tools
import argparse
import os
import time
from typing import Any, Iterator, Tuple
from tools.wpt.testfiles import get_git_cmd
here = os.path.dirname(__file__)
wpt_root = os.path.abspath(os.path.join(here, os.pardir, os.pardir))
def calculate_cutoff_date(until: int, epoch: int, offset: int) -> int:
return (((until - offset) // epoch) * epoch) + offset
def parse_epoch(string: str) -> int:
UNIT_DICT = {"h": 3600, "d": 86400, "w": 604800}
base = string[:-1]
unit = string[-1:]
if base.isdigit() and unit in UNIT_DICT:
return int(base) * UNIT_DICT[unit]
raise argparse.ArgumentTypeError('must be digits followed by h/d/w')
def get_tagged_revisions(pattern: str) -> Iterator[Tuple[str, str, int]]:
'''
Iterates the tagged revisions as (tag name, commit sha, committer date) tuples.
'''
git = get_git_cmd(wpt_root)
args = [
pattern,
'--sort=-committerdate',
'--format=%(refname:lstrip=2) %(objectname) %(committerdate:raw)',
'--count=100000'
]
ref_list = git("for-each-ref", *args) # type: ignore
for line in ref_list.splitlines():
if not line:
continue
tag, commit, date, _ = line.split(" ")
date = int(date)
yield tag, commit, date
def get_epoch_revisions(epoch: int, until: int, max_count: int) -> Iterator[str]:
# Set an offset to start to count the the weekly epoch from
# Monday 00:00:00. This is particularly important for the weekly epoch
# because fix the start of the epoch to Monday. This offset is calculated
# from Thursday, 1 January 1970 0:00:00 to Monday, 5 January 1970 0:00:00
epoch_offset = 345600
count = 0
# Iterates the tagged revisions in descending order finding the more
# recent commit still older than a "cutoff_date" value.
# When a commit is found "cutoff_date" is set to a new value multiplier of
# "epoch" but still below of the date of the current commit found.
# This needed to deal with intervals where no candidates were found
# for the current "epoch" and the next candidate found is yet below
# the lower values of the interval (it is the case of J and I for the
# interval between Wed and Tue, in the example). The algorithm fix
# the next "cutoff_date" value based on the date value of the current one
# skipping the intermediate values.
# The loop ends once we reached the required number of revisions to return
# or the are no more tagged revisions or the cutoff_date reach zero.
#
# Fri Sat Sun Mon Tue Wed Thu Fri Sat
# | | | | | | | | |
# -A---B-C---DEF---G---H--IJ----------K-----L-M----N--O--
# ^
# now
# Expected result: N,M,K,J,H,G,F,C,A
cutoff_date = calculate_cutoff_date(until, epoch, epoch_offset)
for _, commit, date in get_tagged_revisions("refs/tags/merge_pr_*"):
if count >= max_count:
return
if date < cutoff_date:
yield commit
count += 1
cutoff_date = calculate_cutoff_date(date, epoch, epoch_offset)
def get_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser()
parser.add_argument("--epoch",
default="1d",
type=parse_epoch,
help="regular interval of time selected to get the "
"tagged revisions. Valid values are digits "
"followed by h/d/w (e.x. 9h, 9d, 9w ...) where "
"the mimimun selectable interval is one hour "
"(1h)")
parser.add_argument("--max-count",
default=1,
type=int,
help="maximum number of revisions to be returned by "
"the command")
return parser
def run_rev_list(**kwargs: Any) -> None:
# "epoch_threshold" is a safety margin. After this time it is fine to
# assume that any tags are created and pushed.
epoch_threshold = 600
until = int(time.time()) - epoch_threshold
for line in get_epoch_revisions(kwargs["epoch"], until, kwargs["max_count"]):
print(line)