108 lines
7.0 KiB
Python
108 lines
7.0 KiB
Python
#!/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()
|