#!/usr/bin/env python3 """Finite runner/schema regressions. Never create a large PDF or launch a GUI.""" import copy import hashlib import json from pathlib import Path import subprocess import tempfile import unittest from unittest.mock import patch import run_stress_pdf as runner def measurement(source_hash): actions = [{"index": i, "series": "page-scroll", "command": "zoom.in" if i % 10 == 8 else "zoom.fit_width" if i % 10 == 9 else "nav.page", "positionChanged": True, "cacheHit": i % 2 == 0, "firstResponseFrameMs": 2, "finalQualityFrameMs": 3} for i in range(100)] return {"schemaVersion": 2, "success": True, "format": "pdf", "documentSha256": source_hash, "requestedOpenTimings": 3, "requestedCloseCycles": 2, "openCount": 3, "openFrameMs": [1, 2, 3], "openRecords": [{}, {}, {}], "operations": actions, "maximumMemoryPressureLevel": 0, "continuousScroll": {"requestedInputs": 0, "dispatchedInputs": 0}, "tileCachePeakChargedBytes": 10, "tileCacheBudgetBytes": 20, "openCloseCycles": [{"workload": "open-first-visible-only" if i < 2 else "navigation-and-scroll", "stateEmpty": True, "cacheBytesAfterClose": 0} for i in range(3)]} class StressRunnerTest(unittest.TestCase): def test_schema_requires_exact_opens_pressure_zero_and_real_navigation(self): valid = measurement("source") checks, summary = runner.summarize_benchmark(valid, "source") self.assertTrue(all(checks.values())) self.assertEqual(summary["cachedNavigation"]["n"], 40) self.assertEqual(summary["uncachedNavigation"]["n"], 40) for field, value in [("schemaVersion", 1), ("openCount", 11), ("maximumMemoryPressureLevel", 3), ("documentSha256", "changed"), ("requestedCloseCycles", 10), ("tileCachePeakChargedBytes", 21)]: with self.subTest(field=field): invalid = copy.deepcopy(valid); invalid[field] = value self.assertFalse(all(runner.summarize_benchmark(invalid, "source")[0].values())) invalid = copy.deepcopy(valid) invalid["operations"][0]["positionChanged"] = False checks, summary = runner.summarize_benchmark(invalid, "source") self.assertFalse(checks["page_navigation_moved"]) self.assertEqual(summary["excludedNoopPageNavigation"], 1) self.assertEqual(summary["cachedNavigation"]["n"], 39) invalid = copy.deepcopy(valid); invalid["operations"][0]["finalQualityFrameMs"] = float("nan") self.assertFalse(runner.summarize_benchmark(invalid, "source")[0]["finite_timings"]) invalid = copy.deepcopy(valid); invalid["openCloseCycles"][0]["cacheBytesAfterClose"] = 1 self.assertFalse(runner.summarize_benchmark(invalid, "source")[0]["close_releases_document"]) def test_missing_executable_removes_prior_aggregate_without_generating(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary); output = root / "output"; output.mkdir() for name in ["report.json", "benchmark.json", "source-manifest.json"]: (output / name).write_text('{"success":true}') unrelated = output / "keep.txt"; unrelated.write_text("keep") with patch.object(runner, "generate") as generate, patch("builtins.print"): code = runner.main(["--binary", str(root / "absent"), "--output", str(output), "--work-directory", str(root / "work")]) self.assertEqual(code, 1); generate.assert_not_called() self.assertFalse(json.loads((output / "report.json").read_text())["success"]) self.assertFalse((output / "benchmark.json").exists()) self.assertFalse((output / "source-manifest.json").exists()) self.assertEqual(unrelated.read_text(), "keep") def test_explicit_workload_and_executable_changes_gate_success(self): for mutate in (False, True): with self.subTest(mutate=mutate), tempfile.TemporaryDirectory() as temporary: root = Path(temporary); output = root / "output" for name in ("docview", "docview-pdf-worker", "docview-archive-worker"): (root / name).write_bytes(b"fixture executable") source_hash = hashlib.sha256(b"synthetic small test").hexdigest() def generate(path): path.write_bytes(b"synthetic small test") return {"sha256": source_hash, "sizeBytes": path.stat().st_size, "images": []} def independent(command, **kwargs): if command[0].endswith("pdfinfo"): return subprocess.CompletedProcess(command, 0, stdout="Pages: 5000\n") page = int(command[command.index("-f") + 1]) return subprocess.CompletedProcess(command, 0, stdout=f"DOCVIEW STRESS PAGE {page:05d} OF 05000 IMAGE {(page - 1) % 1280:04d}") def observed(command, env, directory, name, **kwargs): if name == "benchmark": for flag, value in [("--benchmark-runs", "3"), ("--benchmark-close-cycles", "2"), ("--benchmark-operations", "100"), ("--benchmark-scroll-steps", "0")]: self.assertEqual(command[command.index(flag) + 1], value) (directory / "benchmark.json").write_text(json.dumps(measurement(source_hash))) else: smoke = directory / "last-page"; smoke.mkdir() (smoke / "state.txt").write_text("Ready\npdf\n\n") (smoke / "runtime.json").write_text('{"qtPlatform":"xcb"}') (smoke / "screen.png").write_bytes(b"test fixture image") if mutate: (root / "docview-pdf-worker").write_bytes(b"changed executable") return {"exitCode": 0} with patch.object(runner, "generate", side_effect=generate), patch.object(runner, "validate_source", return_value={}), \ patch.object(runner, "run_observed", side_effect=observed), patch.object(runner.subprocess, "run", side_effect=independent), \ patch.dict(runner.os.environ, {"QTWEBENGINE_DISABLE_SANDBOX": "0", "QTWEBENGINE_CHROMIUM_FLAGS": ""}), patch("builtins.print"): code = runner.main(["--binary", str(root / "docview"), "--output", str(output), "--work-directory", str(root / "work")]) result = json.loads((output / "report.json").read_text()) self.assertEqual(code, 1 if mutate else 0) self.assertEqual(result["success"], not mutate) self.assertEqual(result["executablesUnchanged"], not mutate) self.assertTrue(result["originalUnchanged"]); self.assertTrue(result["sourceRemoved"]) self.assertEqual(set(result["executableSha256"]), {"docview", "docview-pdf-worker", "docview-archive-worker"}) if __name__ == "__main__": unittest.main()