| 1 | """Pin the load-generation instrument (oakbench.loadgen).
|
| 2 |
|
| 3 | stress-ng discipline is non-negotiable: every built command carries --timeout
|
| 4 | (stress-ng's default is 24h), --temp-path (contend on the benchmarked
|
| 5 | filesystem), and a deterministic --seed. Absence of the optional stress-ng
|
| 6 | binary is a reported condition, never a crash. bogo-ops are parsed ONLY to
|
| 7 | verify the load ran.
|
| 8 | """
|
| 9 |
|
| 10 | from __future__ import annotations
|
| 11 |
|
| 12 | import tempfile
|
| 13 | import unittest
|
| 14 | from pathlib import Path
|
| 15 | from unittest import mock
|
| 16 |
|
| 17 | from oakbench import loadgen
|
| 18 | from oakbench.loadgen import (
|
| 19 | LOAD_TIERS,
|
| 20 | LoadGenerator,
|
| 21 | cpu_worker_count,
|
| 22 | environment_boundary_fields,
|
| 23 | load_tier_unavailable_reason,
|
| 24 | stress_ng_command,
|
| 25 | verify_ran,
|
| 26 | )
|
| 27 |
|
| 28 |
|
| 29 | class LoadTierTableTest(unittest.TestCase):
|
| 30 | def test_tiers_are_exactly_the_published_set(self) -> None:
|
| 31 | self.assertEqual(set(LOAD_TIERS), {"none", "cpu50", "io", "mixed"})
|
| 32 | self.assertIsNone(LOAD_TIERS["none"])
|
| 33 |
|
| 34 | def test_cpu_worker_count_is_half_cores_at_least_one(self) -> None:
|
| 35 | self.assertEqual(cpu_worker_count(8), 4)
|
| 36 | self.assertEqual(cpu_worker_count(7), 3)
|
| 37 | self.assertEqual(cpu_worker_count(1), 1)
|
| 38 |
|
| 39 |
|
| 40 | class StressNgCommandTest(unittest.TestCase):
|
| 41 | def test_none_tier_builds_no_command(self) -> None:
|
| 42 | self.assertIsNone(stress_ng_command("none", timeout_s=30, temp_path="/t", seed=7))
|
| 43 |
|
| 44 | def test_unknown_tier_raises(self) -> None:
|
| 45 | with self.assertRaises(ValueError):
|
| 46 | stress_ng_command("cpu100", timeout_s=30, temp_path="/t", seed=7)
|
| 47 |
|
| 48 | def test_nonpositive_timeout_raises(self) -> None:
|
| 49 | # stress-ng's default run length is 24 hours; a missing/zero timeout
|
| 50 | # must never silently fall through to it.
|
| 51 | for bad in (0, -5):
|
| 52 | with self.assertRaises(ValueError):
|
| 53 | stress_ng_command("cpu50", timeout_s=bad, temp_path="/t", seed=7)
|
| 54 |
|
| 55 | def test_cpu50_exact_args(self) -> None:
|
| 56 | command = stress_ng_command("cpu50", timeout_s=30, temp_path="/repo/fs/tmp", seed=7, cpu_count=8)
|
| 57 | self.assertEqual(
|
| 58 | command,
|
| 59 | [
|
| 60 | "stress-ng",
|
| 61 | "--timeout", "30s",
|
| 62 | "--temp-path", "/repo/fs/tmp",
|
| 63 | "--seed", "7",
|
| 64 | "--metrics-brief",
|
| 65 | "--cpu", "4",
|
| 66 | "--cpu-load", "50",
|
| 67 | ],
|
| 68 | )
|
| 69 |
|
| 70 | def test_io_exact_args(self) -> None:
|
| 71 | command = stress_ng_command("io", timeout_s=45, temp_path="/repo/fs/tmp", seed=11, cpu_count=8)
|
| 72 | self.assertEqual(
|
| 73 | command,
|
| 74 | [
|
| 75 | "stress-ng",
|
| 76 | "--timeout", "45s",
|
| 77 | "--temp-path", "/repo/fs/tmp",
|
| 78 | "--seed", "11",
|
| 79 | "--metrics-brief",
|
| 80 | "--hdd", "2",
|
| 81 | "--hdd-bytes", "256M",
|
| 82 | ],
|
| 83 | )
|
| 84 |
|
| 85 | def test_mixed_exact_args(self) -> None:
|
| 86 | command = stress_ng_command("mixed", timeout_s=60, temp_path="/repo/fs/tmp", seed=3, cpu_count=2)
|
| 87 | self.assertEqual(
|
| 88 | command,
|
| 89 | [
|
| 90 | "stress-ng",
|
| 91 | "--timeout", "60s",
|
| 92 | "--temp-path", "/repo/fs/tmp",
|
| 93 | "--seed", "3",
|
| 94 | "--metrics-brief",
|
| 95 | "--cpu", "1",
|
| 96 | "--cpu-load", "50",
|
| 97 | "--hdd", "2",
|
| 98 | "--hdd-bytes", "256M",
|
| 99 | ],
|
| 100 | )
|
| 101 |
|
| 102 | def test_every_non_none_tier_always_has_timeout_temp_path_and_seed(self) -> None:
|
| 103 | for tier in ("cpu50", "io", "mixed"):
|
| 104 | command = stress_ng_command(tier, timeout_s=120, temp_path="/contended/fs", seed=42, cpu_count=4)
|
| 105 | assert command is not None
|
| 106 | self.assertIn("--timeout", command)
|
| 107 | self.assertEqual(command[command.index("--timeout") + 1], "120s")
|
| 108 | self.assertIn("--temp-path", command)
|
| 109 | self.assertEqual(command[command.index("--temp-path") + 1], "/contended/fs")
|
| 110 | self.assertIn("--seed", command)
|
| 111 | self.assertEqual(command[command.index("--seed") + 1], "42")
|
| 112 |
|
| 113 | def test_command_is_deterministic(self) -> None:
|
| 114 | first = stress_ng_command("mixed", timeout_s=30, temp_path="/t", seed=9, cpu_count=8)
|
| 115 | second = stress_ng_command("mixed", timeout_s=30, temp_path="/t", seed=9, cpu_count=8)
|
| 116 | self.assertEqual(first, second)
|
| 117 |
|
| 118 |
|
| 119 | # Golden stderr samples crafted from the stress-ng man page / observed output
|
| 120 | # format: info chatter, a metrics header pair, then per-stressor bogo-ops rows.
|
| 121 | GOLDEN_METRC_OUTPUT = "\n".join(
|
| 122 | [
|
| 123 | "stress-ng: info: [4242] setting to a 10 second run per stressor",
|
| 124 | "stress-ng: info: [4242] dispatching hogs: 4 cpu",
|
| 125 | "stress-ng: metrc: [4242] stressor bogo ops real time usr time sys time bogo ops/s bogo ops/s",
|
| 126 | "stress-ng: metrc: [4242] (secs) (secs) (secs) (real time) (usr+sys time)",
|
| 127 | "stress-ng: metrc: [4242] cpu 20704 10.00 38.69 0.02 2070.21 534.95",
|
| 128 | "stress-ng: info: [4242] successful run completed in 10.01s",
|
| 129 | ]
|
| 130 | )
|
| 131 |
|
| 132 | GOLDEN_OLD_INFO_OUTPUT = "\n".join(
|
| 133 | [
|
| 134 | "stress-ng: info: [991] dispatching hogs: 2 hdd",
|
| 135 | "stress-ng: info: [991] stressor bogo ops real time usr time sys time bogo ops/s bogo ops/s",
|
| 136 | "stress-ng: info: [991] (secs) (secs) (secs) (real time) (usr+sys time)",
|
| 137 | "stress-ng: info: [991] hdd 512 10.00 0.10 1.20 51.20 393.85",
|
| 138 | "stress-ng: info: [991] successful run completed in 10.02s",
|
| 139 | ]
|
| 140 | )
|
| 141 |
|
| 142 | GOLDEN_ZERO_OPS_OUTPUT = "\n".join(
|
| 143 | [
|
| 144 | "stress-ng: info: [77] dispatching hogs: 2 hdd",
|
| 145 | "stress-ng: metrc: [77] stressor bogo ops real time usr time sys time bogo ops/s bogo ops/s",
|
| 146 | "stress-ng: metrc: [77] hdd 0 0.01 0.00 0.00 0.00 0.00",
|
| 147 | ]
|
| 148 | )
|
| 149 |
|
| 150 |
|
| 151 | class VerifyRanTest(unittest.TestCase):
|
| 152 | def test_metrc_format_with_positive_bogo_ops_verifies(self) -> None:
|
| 153 | self.assertTrue(verify_ran(GOLDEN_METRC_OUTPUT))
|
| 154 |
|
| 155 | def test_older_info_format_with_positive_bogo_ops_verifies(self) -> None:
|
| 156 | self.assertTrue(verify_ran(GOLDEN_OLD_INFO_OUTPUT))
|
| 157 |
|
| 158 | def test_zero_bogo_ops_does_not_verify(self) -> None:
|
| 159 | self.assertFalse(verify_ran(GOLDEN_ZERO_OPS_OUTPUT))
|
| 160 |
|
| 161 | def test_chatter_without_metrics_does_not_verify(self) -> None:
|
| 162 | chatter = "\n".join(
|
| 163 | [
|
| 164 | "stress-ng: info: [10] setting to a 10 second run per stressor",
|
| 165 | "stress-ng: info: [10] dispatching hogs: 4 cpu",
|
| 166 | ]
|
| 167 | )
|
| 168 | self.assertFalse(verify_ran(chatter))
|
| 169 | self.assertFalse(verify_ran(""))
|
| 170 | self.assertFalse(verify_ran("stress-ng: error: [9] No such file or directory"))
|
| 171 |
|
| 172 |
|
| 173 | class LoadGeneratorTest(unittest.TestCase):
|
| 174 | def test_unknown_tier_raises_at_construction(self) -> None:
|
| 175 | with self.assertRaises(ValueError):
|
| 176 | LoadGenerator("turbo", "/tmp/x")
|
| 177 |
|
| 178 | def test_none_tier_never_starts(self) -> None:
|
| 179 | generator = LoadGenerator("none", "/tmp/never-used")
|
| 180 | self.assertEqual(generator.start(), {"started": False, "reason": "tier_none"})
|
| 181 |
|
| 182 | def test_missing_stress_ng_reports_started_false(self) -> None:
|
| 183 | with tempfile.TemporaryDirectory() as tmp:
|
| 184 | generator = LoadGenerator("cpu50", Path(tmp) / "load-temp", timeout_s=5)
|
| 185 | with mock.patch.object(loadgen.shutil, "which", return_value=None):
|
| 186 | result = generator.start()
|
| 187 | self.assertEqual(result, {"started": False, "reason": "stress_ng_missing"})
|
| 188 | # stop() after a failed start is safe and reports nothing-was-measured.
|
| 189 | stop = generator.stop()
|
| 190 | self.assertFalse(stop["stopped"])
|
| 191 | self.assertIsNone(stop["verified"])
|
| 192 |
|
| 193 | def test_stop_without_start_reports_unmeasured(self) -> None:
|
| 194 | generator = LoadGenerator("io", "/tmp/never-started")
|
| 195 | stop = generator.stop()
|
| 196 | self.assertEqual(stop["stopped"], False)
|
| 197 | self.assertIsNone(stop["returncode"])
|
| 198 | self.assertIsNone(stop["verified"])
|
| 199 |
|
| 200 |
|
| 201 | class RowFieldHelperTest(unittest.TestCase):
|
| 202 | def test_unavailable_reason_is_canonical(self) -> None:
|
| 203 | self.assertEqual(
|
| 204 | load_tier_unavailable_reason("stress_ng_missing"),
|
| 205 | "load_tier_unavailable:stress_ng_missing",
|
| 206 | )
|
| 207 |
|
| 208 | def test_boundary_fields_with_samples_and_clean_window(self) -> None:
|
| 209 | sample = {"loadavg": [0.1, 0.1, 0.1], "cpu_speed_limit_pct": 100, "cpu_governor": None}
|
| 210 | fields = environment_boundary_fields("cpu50", sample, sample, cpu_count=8)
|
| 211 | self.assertEqual(fields["load_tier"], "cpu50")
|
| 212 | self.assertEqual(fields["environment_suspect"], []) # ran-and-clean (ADR-0002)
|
| 213 | self.assertEqual(fields["environment_before"], sample)
|
| 214 | self.assertEqual(fields["environment_after"], sample)
|
| 215 |
|
| 216 | def test_boundary_fields_without_samples_keep_suspect_null(self) -> None:
|
| 217 | fields = environment_boundary_fields("none", None, None)
|
| 218 | self.assertEqual(fields["load_tier"], "none") # always the explicit string
|
| 219 | self.assertIsNone(fields["environment_suspect"]) # machinery did not run
|
| 220 |
|
| 221 |
|
| 222 | if __name__ == "__main__":
|
| 223 | unittest.main()
|