nicweyand
557ba7cd69
All checks were successful
Standalone registry checks / check (push) Successful in 5m58s
82 lines
3.4 KiB
Python
82 lines
3.4 KiB
Python
"""Exercise the bounded benchmark evidence runner."""
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
class BenchmarkTests(unittest.TestCase):
|
|
def test_report_pins_source_and_refuses_overwrite(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
manifest = root / "source.json"
|
|
manifest.write_text(json.dumps({
|
|
"schema": "argand.site-source/v3",
|
|
"source": "ror",
|
|
"snapshot": "fixture",
|
|
"sha256": "1" * 64,
|
|
"bytes": 42,
|
|
}))
|
|
output = root / "benchmark"
|
|
command = [
|
|
sys.executable,
|
|
"scripts/benchmark.py",
|
|
"--profile", "small",
|
|
"--output", str(output),
|
|
"--source-manifest", str(manifest),
|
|
"--", sys.executable, "-c", "print('{\"passed\": 1}')",
|
|
]
|
|
subprocess.run(command, check=True)
|
|
report = json.loads((output / "REPORT.json").read_text())
|
|
self.assertEqual(report["schema"], "argand.site-benchmark/v1")
|
|
self.assertEqual(report["sources"][0]["source"], "ror")
|
|
self.assertEqual(report["sources"][0]["source_object_sha256"], "1" * 64)
|
|
self.assertEqual(report["operation"]["result"], {"passed": 1})
|
|
refused = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
self.assertEqual(refused.returncode, 2)
|
|
self.assertIn(b"already exists", refused.stderr)
|
|
|
|
def test_output_cap_does_not_limit_measured_files(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
output = root / "benchmark"
|
|
large_file = root / "database-like-file"
|
|
subprocess.run([
|
|
sys.executable,
|
|
"scripts/benchmark.py",
|
|
"--profile", "small",
|
|
"--output", str(output),
|
|
"--measured-path", f"large={large_file}",
|
|
"--", sys.executable, "-c",
|
|
(
|
|
"import json,sys; "
|
|
"open(sys.argv[1],'wb').write(b'x'*(17*1024*1024)); "
|
|
"print(json.dumps({'complete':True}))"
|
|
),
|
|
str(large_file),
|
|
], check=True)
|
|
report = json.loads((output / "REPORT.json").read_text())
|
|
self.assertFalse(report["operation"]["output_limit_exceeded"])
|
|
self.assertEqual(report["operation"]["result"], {"complete": True})
|
|
self.assertEqual(report["artifacts"][0]["bytes"], 17 * 1024 * 1024)
|
|
|
|
capped = root / "capped"
|
|
result = subprocess.run([
|
|
sys.executable,
|
|
"scripts/benchmark.py",
|
|
"--profile", "small",
|
|
"--output", str(capped),
|
|
"--", sys.executable, "-c",
|
|
"import sys; sys.stdout.buffer.write(b'x'*(17*1024*1024))",
|
|
])
|
|
self.assertNotEqual(result.returncode, 0)
|
|
capped_report = json.loads((capped / "REPORT.json").read_text())
|
|
self.assertTrue(capped_report["operation"]["output_limit_exceeded"])
|
|
self.assertEqual((capped / "stdout").stat().st_size, 16 * 1024 * 1024)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|