| 1 | #!/usr/bin/env python3 |
| 2 | """Hermetic contract tests for the persistence backlog ratchet.""" |
| 3 | |
| 4 | from __future__ import annotations |
| 5 | |
| 6 | import copy |
| 7 | import importlib.util |
| 8 | import io |
| 9 | import json |
| 10 | import sys |
| 11 | import tempfile |
| 12 | import unittest |
| 13 | from contextlib import redirect_stderr, redirect_stdout |
| 14 | from pathlib import Path |
| 15 | from unittest import mock |
| 16 | |
| 17 | |
| 18 | ROOT = Path(__file__).resolve().parents[1] |
| 19 | SCRIPT = ROOT / "scripts" / "check-persistence-backlog-budget.py" |
| 20 | SPEC = importlib.util.spec_from_file_location("check_persistence_backlog_budget", SCRIPT) |
| 21 | assert SPEC and SPEC.loader |
| 22 | mod = importlib.util.module_from_spec(SPEC) |
| 23 | sys.modules[SPEC.name] = mod |
| 24 | SPEC.loader.exec_module(mod) |
| 25 | |
| 26 | |
| 27 | def receipt_fixture(*, rss_supported: bool = True) -> dict: |
| 28 | receipt = { |
| 29 | "document_kind": mod.RECEIPT_KIND, |
| 30 | "schema_version": mod.SCHEMA_VERSION, |
| 31 | **mod.FIXTURE, |
| 32 | "source_sha": "0123456789abcdef0123456789abcdef01234567", |
| 33 | "source_dirty": False, |
| 34 | "rustc_version": "rustc test", |
| 35 | "cargo_version": "cargo test", |
| 36 | "build_profile": "test", |
| 37 | "sample_count": 1, |
| 38 | "platform": "macos" if rss_supported else "linux", |
| 39 | "accepted_requests": 128, |
| 40 | "retained_queued_requests": 128, |
| 41 | "estimated_retained_payload_bytes": 8_500_000, |
| 42 | "applied_version": 127, |
| 43 | "final_version_applied": True, |
| 44 | "enqueue_elapsed_ns": 500_000, |
| 45 | "rss_supported": rss_supported, |
| 46 | "rss_before_bytes": 100_000_000 if rss_supported else None, |
| 47 | "rss_during_bytes": 112_000_000 if rss_supported else None, |
| 48 | "rss_after_bytes": 103_000_000 if rss_supported else None, |
| 49 | "rss_during_delta_bytes": 12_000_000 if rss_supported else None, |
| 50 | "rss_after_delta_bytes": 3_000_000 if rss_supported else None, |
| 51 | "limitations": ["macOS RSS only"], |
| 52 | } |
| 53 | return receipt |
| 54 | |
| 55 | |
| 56 | def budget_fixture(receipt: dict | None = None) -> dict: |
| 57 | receipt = receipt or receipt_fixture() |
| 58 | metrics = { |
| 59 | field: receipt[field] if receipt[field] is not None else 0 |
| 60 | for field in mod.CEILING_FIELDS |
| 61 | } |
| 62 | return { |
| 63 | "document_kind": mod.BUDGET_KIND, |
| 64 | "schema_version": mod.SCHEMA_VERSION, |
| 65 | "baseline_receipt": mod.BASELINE_RECEIPT_REFERENCE, |
| 66 | "fixture": copy.deepcopy(mod.FIXTURE), |
| 67 | "baseline_observation": { |
| 68 | "accepted_requests": receipt["accepted_requests"], |
| 69 | "applied_version": receipt["applied_version"], |
| 70 | "provenance": { |
| 71 | "platform": "macos", |
| 72 | "source_sha": receipt["source_sha"], |
| 73 | "source_dirty": False, |
| 74 | "rustc_version": receipt["rustc_version"], |
| 75 | "cargo_version": receipt["cargo_version"], |
| 76 | "build_profile": "test", |
| 77 | "sample_count": 1, |
| 78 | }, |
| 79 | **copy.deepcopy(metrics), |
| 80 | }, |
| 81 | "ceilings": copy.deepcopy(metrics), |
| 82 | } |
| 83 | |
| 84 | |
| 85 | class PersistenceBacklogBudgetTests(unittest.TestCase): |
| 86 | def test_equal_baseline_passes(self) -> None: |
| 87 | self.assertEqual( |
| 88 | mod.compare(receipt_fixture(), budget_fixture()), |
| 89 | ([], []), |
| 90 | ) |
| 91 | |
| 92 | def test_every_receipt_field_is_required(self) -> None: |
| 93 | budget = budget_fixture() |
| 94 | for field in mod.REQUIRED_RECEIPT_FIELDS: |
| 95 | with self.subTest(field=field): |
| 96 | receipt = receipt_fixture() |
| 97 | del receipt[field] |
| 98 | with self.assertRaisesRegex( |
| 99 | mod.PersistenceBacklogError, "missing required field" |
| 100 | ): |
| 101 | mod.compare(receipt, budget) |
| 102 | |
| 103 | def test_frozen_workload_cannot_be_weakened_to_fake_an_improvement(self) -> None: |
| 104 | budget = budget_fixture() |
| 105 | for field, replacement in [ |
| 106 | ("paused_consumer", False), |
| 107 | ("requests_attempted", 64), |
| 108 | ("content_bytes_per_request", 32 * 1024), |
| 109 | ("single_session_id", False), |
| 110 | ("expected_applied_version", 63), |
| 111 | ("request_variant", "clear_checkpoint"), |
| 112 | ("payload_estimator", "shallow-size"), |
| 113 | ]: |
| 114 | with self.subTest(field=field): |
| 115 | receipt = receipt_fixture() |
| 116 | receipt[field] = replacement |
| 117 | with self.assertRaisesRegex(mod.PersistenceBacklogError, field): |
| 118 | mod.compare(receipt, budget) |
| 119 | |
| 120 | def test_boolean_fixture_fields_reject_integer_aliases(self) -> None: |
| 121 | budget = budget_fixture() |
| 122 | for field in ("paused_consumer", "single_session_id"): |
| 123 | with self.subTest(field=field): |
| 124 | receipt = receipt_fixture() |
| 125 | receipt[field] = 1 |
| 126 | with self.assertRaisesRegex(mod.PersistenceBacklogError, field): |
| 127 | mod.compare(receipt, budget) |
| 128 | |
| 129 | def test_budget_boolean_and_sample_count_aliases_are_rejected(self) -> None: |
| 130 | receipt = receipt_fixture() |
| 131 | for field in ("paused_consumer", "single_session_id"): |
| 132 | with self.subTest(field=field): |
| 133 | budget = budget_fixture() |
| 134 | budget["fixture"][field] = 1 |
| 135 | with self.assertRaisesRegex(mod.PersistenceBacklogError, field): |
| 136 | mod.compare(receipt, budget) |
| 137 | |
| 138 | budget = budget_fixture() |
| 139 | budget["baseline_observation"]["provenance"]["sample_count"] = True |
| 140 | with self.assertRaisesRegex(mod.PersistenceBacklogError, "sample count"): |
| 141 | mod.compare(receipt, budget) |
| 142 | |
| 143 | def test_every_ceiling_rejects_growth_and_accepts_tightening(self) -> None: |
| 144 | baseline = receipt_fixture() |
| 145 | baseline["retained_queued_requests"] = 64 |
| 146 | # Leave enough valid payload headroom for the retained-count subtest |
| 147 | # to change that one metric without making the receipt impossible. |
| 148 | baseline["estimated_retained_payload_bytes"] = 8_500_000 |
| 149 | budget = budget_fixture(baseline) |
| 150 | for field in mod.CEILING_FIELDS: |
| 151 | with self.subTest(field=field): |
| 152 | grown = copy.deepcopy(baseline) |
| 153 | grown[field] += 1 |
| 154 | if field == "rss_during_delta_bytes": |
| 155 | grown["rss_during_bytes"] += 1 |
| 156 | elif field == "rss_after_delta_bytes": |
| 157 | grown["rss_after_bytes"] += 1 |
| 158 | increases, _ = mod.compare(grown, budget) |
| 159 | self.assertEqual([item[0] for item in increases], [field]) |
| 160 | |
| 161 | reduced = copy.deepcopy(baseline) |
| 162 | reduced[field] -= 1 |
| 163 | if field == "rss_during_delta_bytes": |
| 164 | reduced["rss_during_bytes"] -= 1 |
| 165 | elif field == "rss_after_delta_bytes": |
| 166 | reduced["rss_after_bytes"] -= 1 |
| 167 | increases, decreases = mod.compare(reduced, budget) |
| 168 | self.assertEqual(increases, []) |
| 169 | self.assertIn(field, [item[0] for item in decreases]) |
| 170 | |
| 171 | def test_non_macos_receipt_keeps_rss_shape_but_skips_rss_ceilings(self) -> None: |
| 172 | receipt = receipt_fixture(rss_supported=False) |
| 173 | budget = budget_fixture() |
| 174 | increases, decreases = mod.compare(receipt, budget) |
| 175 | self.assertEqual(increases, []) |
| 176 | self.assertNotIn( |
| 177 | "rss_during_delta_bytes", [item[0] for item in decreases] |
| 178 | ) |
| 179 | self.assertNotIn("rss_after_delta_bytes", [item[0] for item in decreases]) |
| 180 | |
| 181 | def test_rss_delta_must_match_samples(self) -> None: |
| 182 | receipt = receipt_fixture() |
| 183 | receipt["rss_during_delta_bytes"] += 1 |
| 184 | with self.assertRaisesRegex(mod.PersistenceBacklogError, "inconsistent"): |
| 185 | mod.compare(receipt, budget_fixture()) |
| 186 | |
| 187 | def test_sender_rejection_cannot_masquerade_as_backlog_improvement(self) -> None: |
| 188 | receipt = receipt_fixture() |
| 189 | receipt["accepted_requests"] = 1 |
| 190 | receipt["retained_queued_requests"] = 1 |
| 191 | receipt["estimated_retained_payload_bytes"] = 66_000 |
| 192 | with self.assertRaisesRegex( |
| 193 | mod.PersistenceBacklogError, "sender rejection is not backlog improvement" |
| 194 | ): |
| 195 | mod.compare(receipt, budget_fixture()) |
| 196 | |
| 197 | def test_stale_applied_version_cannot_pass_as_coalescing(self) -> None: |
| 198 | receipt = receipt_fixture() |
| 199 | receipt["applied_version"] = 126 |
| 200 | receipt["final_version_applied"] = False |
| 201 | with self.assertRaisesRegex(mod.PersistenceBacklogError, "final sent version"): |
| 202 | mod.compare(receipt, budget_fixture()) |
| 203 | |
| 204 | def test_impossible_one_byte_retained_payload_is_rejected(self) -> None: |
| 205 | receipt = receipt_fixture() |
| 206 | receipt["retained_queued_requests"] = 1 |
| 207 | receipt["estimated_retained_payload_bytes"] = 1 |
| 208 | with self.assertRaisesRegex(mod.PersistenceBacklogError, "frozen retained content"): |
| 209 | mod.compare(receipt, budget_fixture()) |
| 210 | |
| 211 | def test_rss_support_is_required_exactly_on_macos(self) -> None: |
| 212 | macos_without_rss = receipt_fixture(rss_supported=False) |
| 213 | macos_without_rss["platform"] = "macos" |
| 214 | with self.assertRaisesRegex(mod.PersistenceBacklogError, "exactly on the macOS"): |
| 215 | mod.compare(macos_without_rss, budget_fixture()) |
| 216 | |
| 217 | linux_with_rss = receipt_fixture() |
| 218 | linux_with_rss["platform"] = "linux" |
| 219 | with self.assertRaisesRegex(mod.PersistenceBacklogError, "exactly on the macOS"): |
| 220 | mod.compare(linux_with_rss, budget_fixture()) |
| 221 | |
| 222 | unknown = receipt_fixture(rss_supported=False) |
| 223 | unknown["platform"] = "unknown" |
| 224 | with self.assertRaisesRegex(mod.PersistenceBacklogError, "unsupported"): |
| 225 | mod.compare(unknown, budget_fixture()) |
| 226 | |
| 227 | malformed = receipt_fixture(rss_supported=False) |
| 228 | malformed["platform"] = [] |
| 229 | with self.assertRaisesRegex(mod.PersistenceBacklogError, "unsupported"): |
| 230 | mod.compare(malformed, budget_fixture()) |
| 231 | |
| 232 | def test_cli_source_identity_rejects_historical_or_dirty_receipts(self) -> None: |
| 233 | receipt = receipt_fixture() |
| 234 | expected = { |
| 235 | field: receipt[field] |
| 236 | for field in ( |
| 237 | "source_sha", |
| 238 | "source_dirty", |
| 239 | "rustc_version", |
| 240 | "cargo_version", |
| 241 | "build_profile", |
| 242 | "sample_count", |
| 243 | ) |
| 244 | } |
| 245 | historical = copy.deepcopy(receipt) |
| 246 | historical["source_sha"] = "f" * 40 |
| 247 | with self.assertRaisesRegex(mod.PersistenceBacklogError, "checked source"): |
| 248 | mod.compare(historical, budget_fixture(), expected_source=expected) |
| 249 | |
| 250 | dirty = copy.deepcopy(receipt) |
| 251 | dirty["source_dirty"] = True |
| 252 | expected_dirty = copy.deepcopy(expected) |
| 253 | expected_dirty["source_dirty"] = True |
| 254 | with self.assertRaisesRegex(mod.PersistenceBacklogError, "source tree is dirty"): |
| 255 | mod.compare( |
| 256 | dirty, |
| 257 | budget_fixture(), |
| 258 | expected_source=expected_dirty, |
| 259 | require_clean_source=True, |
| 260 | ) |
| 261 | |
| 262 | def test_budget_cannot_hide_a_baseline_above_its_ceiling(self) -> None: |
| 263 | budget = budget_fixture() |
| 264 | budget["baseline_observation"]["retained_queued_requests"] += 1 |
| 265 | with self.assertRaisesRegex(mod.PersistenceBacklogError, "exceeds its ceiling"): |
| 266 | mod.compare(receipt_fixture(), budget) |
| 267 | |
| 268 | def test_budget_cannot_claim_an_empty_retained_baseline(self) -> None: |
| 269 | budget = budget_fixture() |
| 270 | budget["baseline_observation"]["retained_queued_requests"] = 0 |
| 271 | budget["baseline_observation"]["estimated_retained_payload_bytes"] = 0 |
| 272 | with self.assertRaisesRegex(mod.PersistenceBacklogError, "retain the final"): |
| 273 | mod.compare(receipt_fixture(), budget) |
| 274 | |
| 275 | def test_raw_baseline_receipt_must_match_budget_metrics_and_provenance(self) -> None: |
| 276 | receipt = receipt_fixture() |
| 277 | budget = budget_fixture(receipt) |
| 278 | mod.validate_baseline_receipt(budget, receipt) |
| 279 | |
| 280 | stale_metric = copy.deepcopy(receipt) |
| 281 | stale_metric["enqueue_elapsed_ns"] += 1 |
| 282 | with self.assertRaisesRegex(mod.PersistenceBacklogError, "does not match"): |
| 283 | mod.validate_baseline_receipt(budget, stale_metric) |
| 284 | |
| 285 | stale_source = copy.deepcopy(receipt) |
| 286 | stale_source["source_sha"] = "f" * 40 |
| 287 | with self.assertRaisesRegex(mod.PersistenceBacklogError, "does not match"): |
| 288 | mod.validate_baseline_receipt(budget, stale_source) |
| 289 | |
| 290 | def _run_cli(self, receipt: dict, budget: dict, *extra: str) -> tuple[int, str, dict]: |
| 291 | """Run main() against temp files with the source identity pinned to ``receipt``.""" |
| 292 | source = { |
| 293 | field: receipt[field] |
| 294 | for field in ( |
| 295 | "source_sha", |
| 296 | "source_dirty", |
| 297 | "rustc_version", |
| 298 | "cargo_version", |
| 299 | "build_profile", |
| 300 | "sample_count", |
| 301 | ) |
| 302 | } |
| 303 | with tempfile.TemporaryDirectory() as tmp: |
| 304 | receipt_path = Path(tmp) / "receipt.json" |
| 305 | budget_path = Path(tmp) / "budget.json" |
| 306 | baseline_path = Path(tmp) / "baseline.json" |
| 307 | receipt_path.write_text(json.dumps(receipt), encoding="utf-8") |
| 308 | budget_path.write_text(json.dumps(budget, indent=2) + "\n", encoding="utf-8") |
| 309 | baseline_path.write_text(json.dumps(receipt_fixture()), encoding="utf-8") |
| 310 | output = io.StringIO() |
| 311 | argv = [ |
| 312 | "check", |
| 313 | "--receipt", |
| 314 | str(receipt_path), |
| 315 | "--budget", |
| 316 | str(budget_path), |
| 317 | *extra, |
| 318 | ] |
| 319 | with ( |
| 320 | mock.patch.object(mod, "current_source_identity", return_value=source), |
| 321 | mock.patch.object(mod, "BASELINE_RECEIPT_PATH", baseline_path), |
| 322 | mock.patch.object(sys, "argv", argv), |
| 323 | redirect_stdout(output), |
| 324 | redirect_stderr(output), |
| 325 | ): |
| 326 | result = mod.main() |
| 327 | written = json.loads(budget_path.read_text(encoding="utf-8")) |
| 328 | return result, output.getvalue(), written |
| 329 | |
| 330 | def test_failure_prints_update_receipt_and_update_raises_only_exceeded(self) -> None: |
| 331 | budget = budget_fixture() |
| 332 | receipt = receipt_fixture() |
| 333 | receipt["enqueue_elapsed_ns"] = budget["ceilings"]["enqueue_elapsed_ns"] + 7 |
| 334 | receipt["retained_queued_requests"] -= 1 |
| 335 | |
| 336 | result, output, unchanged = self._run_cli(receipt, budget) |
| 337 | self.assertEqual(result, 1) |
| 338 | self.assertIn("check-persistence-backlog-budget.py", output) |
| 339 | self.assertIn("--update", output) |
| 340 | self.assertEqual(unchanged, budget) |
| 341 | |
| 342 | result, output, updated = self._run_cli(receipt, budget, "--update") |
| 343 | self.assertEqual(result, 0, output) |
| 344 | self.assertEqual( |
| 345 | updated["ceilings"]["enqueue_elapsed_ns"], receipt["enqueue_elapsed_ns"] |
| 346 | ) |
| 347 | # Decreases keep their noise headroom: --update never lowers a ceiling. |
| 348 | self.assertEqual( |
| 349 | updated["ceilings"]["retained_queued_requests"], |
| 350 | budget["ceilings"]["retained_queued_requests"], |
| 351 | ) |
| 352 | self.assertEqual(updated["baseline_observation"], budget["baseline_observation"]) |
| 353 | |
| 354 | def test_update_without_growth_leaves_budget_untouched(self) -> None: |
| 355 | budget = budget_fixture() |
| 356 | result, output, written = self._run_cli(receipt_fixture(), budget, "--update") |
| 357 | self.assertEqual(result, 0, output) |
| 358 | self.assertEqual(written, budget) |
| 359 | |
| 360 | |
| 361 | if __name__ == "__main__": |
| 362 | unittest.main() |
| 363 |