Log in
benchmarksPublic

Viewing files at commit 5dd811a1487aCommit detailsView latest

Browse files
tests/test_fleet_report.py 213 lines · 8.6 KB · python
1
"""Hand-computed expectations for fleet_report's pure computations.
2
3
Synthetic rows cover three worker tiers (2, 8, 32) so the saturation knee,
4
fairness spread, starvation count, and server-error detection are all checked
5
against numbers computed by hand in the comments.
6
"""
7
8
from __future__ import annotations
9
10
import json
11
import sys
12
import tempfile
13
import unittest
14
from pathlib import Path
15
16
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
17
18
import fleet_report
19
20
21
def total_row(workers: int, throughput: float | None, subject: str = "git", mode: str = "merge_train", **extra) -> dict:
22
    row = {
23
        "operation": "parallel.total",
24
        "scenario": f"contention_{mode}_w{workers}",
25
        "contention_mode": mode,
26
        "subject": subject,
27
        "workers": workers,
28
        "returncode": 0,
29
        "elapsed_ms": 100.0,
30
        "parallel_metrics": {"commit_throughput_per_s": throughput},
31
    }
32
    row.update(extra)
33
    return row
34
35
36
def worker_row(workers: int, worker: int, elapsed_ms: float, **extra) -> dict:
37
    row = {
38
        "operation": "parallel.worker",
39
        "scenario": f"contention_merge_train_w{workers}",
40
        "contention_mode": "merge_train",
41
        "subject": "git",
42
        "workers": workers,
43
        "worker": worker,
44
        "returncode": 0,
45
        "elapsed_ms": elapsed_ms,
46
    }
47
    row.update(extra)
48
    return row
49
50
51
class SaturationKneeTest(unittest.TestCase):
52
    def test_knee_is_largest_tier_at_70_percent_of_w2_per_worker_baseline(self) -> None:
53
        # w2: 10.0/s -> 5.0 per worker (baseline)
54
        # w8: 32.0/s -> 4.0 per worker  >= 0.7 * 5.0 = 3.5  (qualifies)
55
        # w32: 64.0/s -> 2.0 per worker < 3.5               (saturated)
56
        # knee = 8
57
        self.assertEqual(fleet_report.saturation_knee({2: 10.0, 8: 32.0, 32: 64.0}), 8)
58
59
    def test_baseline_falls_back_to_smallest_tier_when_w2_absent(self) -> None:
60
        # baseline tier 4: 8.0/4 = 2.0 per worker; w8: 8.0/8 = 1.0 < 1.4
61
        self.assertEqual(fleet_report.saturation_knee({4: 8.0, 8: 8.0}), 4)
62
63
    def test_no_measurable_baseline_yields_none(self) -> None:
64
        self.assertIsNone(fleet_report.saturation_knee({}))
65
        self.assertIsNone(fleet_report.saturation_knee({2: 0.0}))
66
67
    def test_throughput_by_tier_averages_runs_and_ignores_unmeasured(self) -> None:
68
        rows = [
69
            total_row(2, 9.0),
70
            total_row(2, 11.0),  # mean 10.0
71
            total_row(8, 32.0),
72
            total_row(8, None),  # unmeasured: excluded, never zero
73
            total_row(32, 64.0),
74
        ]
75
        curves = fleet_report.throughput_by_tier(rows)
76
        tiers = curves[("git", "merge_train")]
77
        self.assertEqual(tiers, {2: 10.0, 8: 32.0, 32: 64.0})
78
        self.assertEqual(fleet_report.saturation_knee(tiers), 8)
79
80
    def test_skip_rows_are_excluded(self) -> None:
81
        rows = [total_row(2, 10.0, skipped=True, returncode=77, skip_reason="x")]
82
        self.assertEqual(fleet_report.throughput_by_tier(rows), {})
83
84
85
class FairnessTest(unittest.TestCase):
86
    def test_spread_and_starvation_hand_computed(self) -> None:
87
        # elapsed across 4 workers: [100, 100, 100, 1000]
88
        # nearest-rank p50: rank ceil(0.5*4)=2 -> 100; p99: rank ceil(0.99*4)=4 -> 1000
89
        # spread = 1000/100 = 10.0; one worker completed 0 ops -> starvation 1
90
        rows = [
91
            worker_row(4, 0, 100.0, commits_succeeded=5),
92
            worker_row(4, 1, 100.0, commits_succeeded=5),
93
            worker_row(4, 2, 100.0, commits_succeeded=5),
94
            worker_row(4, 3, 1000.0, commits_succeeded=0),
95
        ]
96
        report = fleet_report.fairness_by_scenario(rows)
97
        self.assertEqual(len(report), 1)
98
        entry = report[0]
99
        self.assertEqual(entry["scenario"], "contention_merge_train_w4")
100
        self.assertEqual(entry["workers_observed"], 4)
101
        self.assertEqual(entry["worker_p50_ms"], 100.0)
102
        self.assertEqual(entry["worker_p99_ms"], 1000.0)
103
        self.assertEqual(entry["worker_p99_spread"], 10.0)
104
        self.assertEqual(entry["starvation_count"], 1)
105
        self.assertEqual(entry["starved_workers"], [3])
106
107
    def test_unmeasured_op_counts_are_never_starved(self) -> None:
108
        rows = [worker_row(2, 0, 50.0), worker_row(2, 1, 60.0)]  # no op-count keys
109
        entry = fleet_report.fairness_by_scenario(rows)[0]
110
        self.assertEqual(entry["starvation_count"], 0)
111
112
    def test_clone_push_storm_zero_pushes_is_starved_absent_is_unmeasured(self) -> None:
113
        def storm_row(worker: int, **extra) -> dict:
114
            row = {
115
                "operation": "parallel.worker",
116
                "scenario": "contention_clone_push_storm_w2",
117
                "contention_mode": "clone_push_storm",
118
                "subject": "git",
119
                "workers": 2,
120
                "worker": worker,
121
                "returncode": 1,
122
                "elapsed_ms": 10.0,
123
            }
124
            row.update(extra)
125
            return row
126
127
        # Worker 0 completed 0 pushes (measured zero -> starved); worker 1
128
        # completed its push. A legacy row without the counter stays
129
        # unmeasured, never starved (ADR-0002).
130
        entry = fleet_report.fairness_by_scenario(
131
            [
132
                storm_row(0, pushes_succeeded=0, clones_succeeded=1),
133
                storm_row(1, pushes_succeeded=1, clones_succeeded=1, returncode=0),
134
                storm_row(2, stderr="push failed"),  # absent counter: unmeasured
135
            ]
136
        )[0]
137
        self.assertEqual(entry["starvation_count"], 1)
138
        self.assertEqual(entry["starved_workers"], [0])
139
        self.assertIsNone(fleet_report.completed_ops(storm_row(2)))
140
        self.assertEqual(fleet_report.completed_ops(storm_row(0, pushes_succeeded=0)), 0)
141
142
    def test_reader_rows_count_polls_as_completed_ops(self) -> None:
143
        rows = [
144
            {
145
                "operation": "parallel.reader",
146
                "scenario": "contention_mixed_fleet_w2",
147
                "subject": "git",
148
                "workers": 2,
149
                "worker": 1,
150
                "elapsed_ms": 10.0,
151
                "returncode": 0,
152
                "polls_total": 0,
153
            }
154
        ]
155
        entry = fleet_report.fairness_by_scenario(rows)[0]
156
        self.assertEqual(entry["starvation_count"], 1)
157
158
159
class ServerErrorTest(unittest.TestCase):
160
    def test_pattern_hits_and_nonzero_returncodes_by_tier(self) -> None:
161
        rows = [
162
            # w2: clean
163
            worker_row(2, 0, 10.0),
164
            # w8: HTTP 503 in stderr AND nonzero returncode
165
            worker_row(8, 0, 10.0, stderr="fatal: the remote end hung up: HTTP 503", returncode=1),
166
            # w32: throttle hint in error_examples (rc 0), plus a plain failure,
167
            # plus a skip row that must NOT count as a nonzero returncode
168
            worker_row(32, 0, 10.0, error_examples=["server throttled the request"]),
169
            worker_row(32, 1, 10.0, returncode=1),
170
            total_row(32, None, returncode=77, skipped=True, skip_reason="r"),
171
        ]
172
        summary = fleet_report.server_error_summary(rows)
173
        by_tier = {entry["workers"]: entry for entry in summary}
174
        self.assertEqual(by_tier[2]["server_error_pattern_hits"], 0)
175
        self.assertEqual(by_tier[2]["nonzero_returncodes"], 0)
176
        self.assertEqual(by_tier[8]["server_error_pattern_hits"], 1)
177
        self.assertEqual(by_tier[8]["nonzero_returncodes"], 1)
178
        self.assertEqual(by_tier[32]["rows"], 3)
179
        self.assertEqual(by_tier[32]["server_error_pattern_hits"], 1)
180
        self.assertEqual(by_tier[32]["nonzero_returncodes"], 1)
181
182
183
class RenderAndCliTest(unittest.TestCase):
184
    def _rows(self) -> list[dict]:
185
        return [
186
            total_row(2, 10.0),
187
            total_row(8, 32.0),
188
            total_row(32, 64.0),
189
            worker_row(2, 0, 100.0, commits_succeeded=1),
190
            worker_row(2, 1, 110.0, commits_succeeded=0),
191
        ]
192
193
    def test_render_marks_knee_and_tables(self) -> None:
194
        report = fleet_report.render_markdown(self._rows())
195
        self.assertIn("knee (w8)", report)
196
        self.assertIn("contention_merge_train_w2", report)
197
        self.assertIn("## Server errors / throttling per tier", report)
198
199
    def test_cli_reads_jsonl_and_writes_report(self) -> None:
200
        with tempfile.TemporaryDirectory(prefix="oakbench-fleet-report-") as tmp:
201
            jsonl = Path(tmp) / "rows.jsonl"
202
            with jsonl.open("w") as fh:
203
                for row in self._rows():
204
                    fh.write(json.dumps(row) + "\n")
205
                fh.write("not json\n")  # malformed lines are skipped
206
            out = Path(tmp) / "report.md"
207
            rc = fleet_report.main([str(jsonl), "--out", str(out)])
208
            self.assertEqual(rc, 0)
209
            self.assertIn("# Fleet Report", out.read_text())
210
211
212
if __name__ == "__main__":
213
    unittest.main()