Log in
benchmarksPublic

Viewing files at commit 2f5107c20e85Commit detailsView latest

Browse files
tests/test_tperf_runner.py 194 lines · 8.0 KB · python
1
"""Golden-fixture tests for the t/perf external cross-check runner.
2
3
These tests run WITHOUT a git source tree: they exercise only the parse path
4
(--parse-only against a synthetic t/test-results directory), the skip-honesty
5
path (missing --git-src), and the sanity_band helper. No t/perf shell is
6
copied here — the fixture is hand-written numeric output in the documented
7
``real user sys`` triplet format.
8
"""
9
10
from __future__ import annotations
11
12
import json
13
import subprocess
14
import sys
15
import tempfile
16
import unittest
17
from pathlib import Path
18
19
ROOT = Path(__file__).resolve().parents[1]
20
sys.path.insert(0, str(ROOT / "scripts"))
21
22
from tperf_runner import (  # noqa: E402
23
    EXIT_OK,
24
    EXIT_SKIP,
25
    MEASUREMENT_SOURCE,
26
    RESULTS_FILENAME,
27
    parse_tperf_results,
28
    sanity_band,
29
)
30
31
SCRIPT = ROOT / "scripts" / "tperf_runner.py"
32
33
34
def make_golden_results(root: Path) -> Path:
35
    """A temp dir mimicking <git-src>/t/test-results for p0001-rev-list."""
36
    results = root / "test-results"
37
    results.mkdir()
38
    (results / "p0001-rev-list.1.result").write_text("0.12 0.10 0.02\n0.15 0.11 0.03\n0.11 0.09 0.02\n")
39
    (results / "p0001-rev-list.subtests").write_text("1\n")
40
    (results / "p0001-rev-list.1.descr").write_text("rev-list --all\n")
41
    return results
42
43
44
def run_cli(*argv: str) -> subprocess.CompletedProcess:
45
    return subprocess.run(
46
        [sys.executable, str(SCRIPT), *argv],
47
        capture_output=True,
48
        text=True,
49
        cwd=str(ROOT),
50
    )
51
52
53
class ParseTperfResultsTest(unittest.TestCase):
54
    def test_golden_fixture_repetition_and_min_rows(self) -> None:
55
        with tempfile.TemporaryDirectory() as tmp:
56
            results = make_golden_results(Path(tmp))
57
            rows = parse_tperf_results(results)
58
59
        repetitions = [row for row in rows if "repetition" in row]
60
        minima = [row for row in rows if row.get("tperf_statistic") == "min_of_n"]
61
        self.assertEqual(len(repetitions), 3)
62
        self.assertEqual(len(minima), 1)
63
64
        self.assertEqual([row["elapsed_ms"] for row in repetitions], [120.0, 150.0, 110.0])
65
        self.assertEqual([row["user_ms"] for row in repetitions], [100.0, 110.0, 90.0])
66
        self.assertEqual([row["sys_ms"] for row in repetitions], [20.0, 30.0, 20.0])
67
        self.assertEqual([row["repetition"] for row in repetitions], [1, 2, 3])
68
69
        for row in rows:
70
            self.assertEqual(row["measurement_source"], "git_tperf_external")
71
            self.assertEqual(row["measurement_source"], MEASUREMENT_SOURCE)
72
            self.assertEqual(row["tperf_script"], "p0001-rev-list")
73
            self.assertEqual(row["tperf_test"], "p0001-rev-list.1")
74
            self.assertEqual(row["tperf_description"], "rev-list --all")
75
            self.assertEqual(row["subject"], "git")
76
            self.assertEqual(row["subject_kind"], "git")
77
78
        self.assertEqual(minima[0]["elapsed_ms"], 110.0)
79
        self.assertEqual(minima[0]["n"], 3)
80
81
    def test_malformed_lines_are_skipped_not_fatal(self) -> None:
82
        with tempfile.TemporaryDirectory() as tmp:
83
            results = Path(tmp)
84
            (results / "p1234-broken.1.result").write_text(
85
                "not a number line\n0.20 0.10 0.05\n\n0.30 garbage 0.01\n0.25\n"
86
            )
87
            rows = parse_tperf_results(results)
88
89
        repetitions = [row for row in rows if "repetition" in row]
90
        minima = [row for row in rows if row.get("tperf_statistic") == "min_of_n"]
91
        self.assertEqual([row["elapsed_ms"] for row in repetitions], [200.0, 250.0])
92
        # Single-float line is accepted liberally; user/sys are unmeasured.
93
        self.assertIsNone(repetitions[1]["user_ms"])
94
        self.assertIsNone(repetitions[1]["sys_ms"])
95
        self.assertEqual(minima[0]["elapsed_ms"], 200.0)
96
        self.assertEqual(minima[0]["n"], 2)
97
98
    def test_missing_descr_yields_null_description(self) -> None:
99
        with tempfile.TemporaryDirectory() as tmp:
100
            results = Path(tmp)
101
            (results / "p0002-read-cache.1.result").write_text("0.05 0.04 0.01\n")
102
            rows = parse_tperf_results(results)
103
        self.assertTrue(rows)
104
        for row in rows:
105
            self.assertIsNone(row["tperf_description"])
106
107
108
class CliParseOnlyTest(unittest.TestCase):
109
    def test_parse_only_writes_stamped_jsonl(self) -> None:
110
        with tempfile.TemporaryDirectory() as tmp:
111
            root = Path(tmp)
112
            results = make_golden_results(root)
113
            out_dir = root / "out"
114
            proc = run_cli("--parse-only", str(results), "--results", str(out_dir))
115
            self.assertEqual(proc.returncode, EXIT_OK, proc.stderr)
116
117
            out_path = out_dir / RESULTS_FILENAME
118
            rows = [json.loads(line) for line in out_path.read_text().splitlines()]
119
120
        repetitions = [row for row in rows if "repetition" in row]
121
        minima = [row for row in rows if row.get("tperf_statistic") == "min_of_n"]
122
        self.assertEqual(len(repetitions), 3)
123
        self.assertEqual(len(minima), 1)
124
        self.assertEqual([row["elapsed_ms"] for row in repetitions], [120.0, 150.0, 110.0])
125
        self.assertEqual(minima[0]["elapsed_ms"], 110.0)
126
        self.assertEqual(minima[0]["n"], 3)
127
        for row in rows:
128
            self.assertEqual(row["measurement_source"], "git_tperf_external")
129
            # Runner identity is stamped on every row (ADR-0007).
130
            self.assertIsInstance(row.get("runner_id"), str)
131
            self.assertTrue(row["runner_id"])
132
            self.assertIn("runner_class", row)
133
134
    def test_parse_only_missing_dir_is_skip(self) -> None:
135
        with tempfile.TemporaryDirectory() as tmp:
136
            missing = Path(tmp) / "nope"
137
            proc = run_cli("--parse-only", str(missing), "--results", str(Path(tmp) / "out"))
138
        self.assertEqual(proc.returncode, EXIT_SKIP)
139
        payload = json.loads(proc.stdout)
140
        self.assertEqual(payload["status"], "skipped")
141
        self.assertEqual(payload["reason"], f"results_dir_missing:{missing}")
142
143
144
class CliRunModeSkipTest(unittest.TestCase):
145
    def test_missing_git_src_flag_exits_3_with_skip_json(self) -> None:
146
        with tempfile.TemporaryDirectory() as tmp:
147
            proc = run_cli("--results", str(Path(tmp) / "out"))
148
        self.assertEqual(proc.returncode, EXIT_SKIP)
149
        payload = json.loads(proc.stdout)
150
        self.assertEqual(payload["status"], "skipped")
151
        self.assertTrue(payload["reason"].startswith("git_src_missing:"))
152
153
    def test_git_src_without_tperf_run_exits_3_with_skip_json(self) -> None:
154
        with tempfile.TemporaryDirectory() as tmp:
155
            not_a_git_tree = Path(tmp) / "src"
156
            not_a_git_tree.mkdir()
157
            proc = run_cli("--git-src", str(not_a_git_tree), "--results", str(Path(tmp) / "out"))
158
            self.assertEqual(proc.returncode, EXIT_SKIP)
159
            payload = json.loads(proc.stdout)
160
            self.assertEqual(payload["status"], "skipped")
161
            self.assertEqual(payload["reason"], f"git_src_missing:{not_a_git_tree}")
162
163
164
class SanityBandTest(unittest.TestCase):
165
    def test_within_band(self) -> None:
166
        band = sanity_band(100.0, 110.0, tolerance_pct=50)
167
        self.assertTrue(band["within_band"])
168
        self.assertAlmostEqual(band["ratio"], 1.1)
169
        self.assertEqual(band["tolerance_pct"], 50)
170
171
    def test_outside_band(self) -> None:
172
        band = sanity_band(100.0, 200.0, tolerance_pct=50)
173
        self.assertFalse(band["within_band"])
174
        self.assertAlmostEqual(band["ratio"], 2.0)
175
176
    def test_boundary_is_inclusive(self) -> None:
177
        band = sanity_band(100.0, 150.0, tolerance_pct=50)
178
        self.assertTrue(band["within_band"])
179
180
    def test_zero_min_guard_ratio_none_within_false(self) -> None:
181
        # Documented guard: tperf_min_ms <= 0 -> ratio None, within_band False.
182
        band = sanity_band(0.0, 100.0)
183
        self.assertIsNone(band["ratio"])
184
        self.assertFalse(band["within_band"])
185
186
    def test_none_inputs_guard(self) -> None:
187
        for tperf_min, our_median in ((None, 100.0), (100.0, None), (None, None)):
188
            band = sanity_band(tperf_min, our_median)
189
            self.assertIsNone(band["ratio"])
190
            self.assertFalse(band["within_band"])
191
192
193
if __name__ == "__main__":
194
    unittest.main()