# Copyright Kevin Deldycke <[email protected]> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
"""Runner timings aggregate honestly, and sort as durations rather than strings."""
from __future__ import annotations
import json
from unittest.mock import patch
import pytest
from click_extra.testing import CliRunner
from repomatic.cli.main import repomatic
from repomatic.github.job_timings import (
UNATTRIBUTED,
JobTiming,
_duration,
fetch_job_timings,
format_duration,
match_runner,
render_markdown,
summarize,
)
[docs]
@pytest.mark.parametrize(
("job_name", "expected"),
[
("⁉️ ubuntu-26.04-arm / py3.15-dev", "ubuntu-26.04-arm"),
# The bare image must not shadow its longer sibling: a prefix scan in
# registry order reports `ubuntu-26.04` for the Arm job.
("✅ ubuntu-26.04 / py3.10", "ubuntu-26.04"),
("release / ⁉️ windows-11-arm, abc1234 build", "windows-11-arm"),
("📦 Package install", UNATTRIBUTED),
],
)
def test_match_runner(job_name: str, expected: str) -> None:
"""A job is attributed to the image its name carries, or to nothing."""
assert match_runner(job_name) == expected
[docs]
@pytest.mark.parametrize(
("job", "expected"),
[
# The form GitHub actually sends, with a trailing `Z`. A parser that
# rejects it returns `None` for every job, and the command then reports
# a fleet with no measurable work rather than failing.
(
{
"startedAt": "2026-08-15T10:00:00Z",
"completedAt": "2026-08-15T10:02:30Z",
},
150.0,
),
# The explicit-offset form, which the `Z` normalisation must not touch:
# appending an offset unconditionally yields `+00:00+00:00`, which
# parses nowhere.
(
{
"startedAt": "2026-08-15T10:00:00+00:00",
"completedAt": "2026-08-15T10:02:30+00:00",
},
150.0,
),
# Cancelled in the queue: no start, so no duration to read.
({"startedAt": "", "completedAt": "2026-08-15T10:02:30Z"}, None),
# Still running: no end.
({"startedAt": "2026-08-15T10:00:00Z", "completedAt": ""}, None),
# Unparsable timestamps are skipped rather than guessed at.
({"startedAt": "yesterday", "completedAt": "today"}, None),
],
)
def test_duration_skips_what_it_cannot_measure(job: dict, expected: float) -> None:
"""A job with no usable pair is skipped, never counted as zero.
Counting it as zero would drag a median toward a runner's queue behaviour
instead of its speed, which is the opposite of what this measures.
"""
assert _duration(job) == expected
[docs]
def test_fetch_job_timings_samples_dated_successful_runs() -> None:
"""Runs come from the dated listing of successful runs, never `gh run list`.
That listing command has served runs weeks old as the newest ones, which
would time an image on builds nobody runs any more.
"""
jobs = {
"jobs": [
{
"name": "✅ ubuntu-26.04 / py3.10",
"startedAt": "2026-08-15T10:00:00Z",
"completedAt": "2026-08-15T10:02:30Z",
},
# Cancelled in the queue: skipped rather than counted as zero.
{"name": "✅ macos-26 / py3.10", "startedAt": "", "completedAt": ""},
]
}
def gh(args):
if args[0] == "api":
return json.dumps({"workflow_runs": [{"id": 8}, {"id": 9}]})
return json.dumps(jobs)
with patch("repomatic.github.gh.run_gh_command", side_effect=gh) as mock_gh:
timings = fetch_job_timings("tests.yaml", "main", limit=2)
assert (
timings
== [
JobTiming("✅ ubuntu-26.04 / py3.10", "ubuntu-26.04", 150.0),
]
* 2
)
commands = [" ".join(call.args[0]) for call in mock_gh.call_args_list]
assert "actions/workflows/tests.yaml/runs?" in commands[0]
assert "status=success" in commands[0]
assert "created=%3E%3D" in commands[0]
assert commands[1:] == ["run view 8 --json jobs", "run view 9 --json jobs"]
[docs]
def test_render_markdown_carries_its_own_caveat() -> None:
"""The table says how it was sampled, since the numbers drift."""
table = render_markdown(summarize([JobTiming("x", "img", 42.0)]), "tests.yaml", 5)
assert "| `img` | 1 | 00m42s | x | 00m42s |" in table
assert "5 most recent successful" in table
assert table.endswith("\n")
[docs]
def test_job_timings_command_prints_the_table() -> None:
"""Each runner image gets a row, from the default sample of `tests.yaml` runs."""
with patch(
"repomatic.cli.ci.fetch_job_timings",
return_value=[JobTiming("x", "img", 42.0)],
) as fetch:
result = CliRunner().invoke(repomatic, ["--table-format", "csv", "job-timings"])
assert result.exit_code == 0, result.output
assert "img,1,00m42s,x,00m42s" in result.stdout
fetch.assert_called_once_with("tests.yaml", "main", 5)
[docs]
def test_job_timings_command_writes_the_report(tmp_path) -> None:
"""`--output` also writes the table as Markdown, caveat included."""
report = tmp_path / "timings.md"
with patch(
"repomatic.cli.ci.fetch_job_timings",
return_value=[JobTiming("x", "img", 42.0)],
):
result = CliRunner().invoke(repomatic, ["job-timings", "--output", str(report)])
assert result.exit_code == 0, result.output
assert "img" in result.stdout
text = report.read_text(encoding="UTF-8")
assert "| `img` | 1 | 00m42s | x | 00m42s |" in text
assert "5 most recent successful" in text