File size: 3,983 Bytes
49ad2ef | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | import importlib.util
import json
import sys
import tempfile
import unittest
from pathlib import Path
def _load_tool(name: str, file_name: str):
path = Path(__file__).parents[1] / "tools" / file_name
spec = importlib.util.spec_from_file_location(name, path)
if spec is None or spec.loader is None:
raise RuntimeError(f"could not load {file_name}")
module = importlib.util.module_from_spec(spec)
sys.modules[name] = module
spec.loader.exec_module(module)
return module
benchmark_tool = _load_tool("verify_browser_benchmark", "verify_browser_benchmark.py")
merge_tool = _load_tool("vons_merge_browser_benchmarks", "merge_browser_benchmarks.py")
def _sample(repeat: int) -> dict[str, object]:
return {
"questionId": f"case-{repeat + 1}",
"live_candidates": 4,
"allocated_candidates": 4,
"live_tokens": 12,
"sequence_length": 512,
"tokenizeMs": 1.0,
"prepareFeedMs": 2.0,
"inferenceAndReadbackMs": 3.0 + repeat,
"postprocessMs": 4.0,
"totalRequestMs": 10.0 + repeat,
"run_id": f"run-{repeat}",
"session_start": 1,
"repeat": repeat,
"case_id": f"case-{repeat + 1}",
"backend": "direct",
"requested_provider": "wasm",
"observed_provider": None,
"observed_provider_status": "session_created; execution_partition_not_exposed",
"bundle_hash": "c" * 64,
"tokenizer_hash": "a" * 64,
"input_hash": "b" * 64,
"noise_hash": None,
"js_heap_used_bytes_before": None,
"js_heap_used_bytes_after": None,
"status": "ok",
}
def _report(run_id: str, offset: int) -> dict[str, object]:
return {
"schema": "vons.browser-benchmark/v1",
"run_id": run_id,
"backend": "direct",
"requested_provider": "wasm",
"sessions": 1,
"warmup_excluded_per_session": 1,
"repeats_per_session": 2,
"cases": 2,
"session_start_semantics": "new ORT session on one page; module and browser caches may persist",
"execution_provider_evidence": "requested provider and session creation only; graph partition status is not exposed by this harness",
"environment": {"user_agent": "test", "cross_origin_isolated": False},
"runtime_info": None,
"session_records": [{"session_start": 1, "load_ms": 5.0, "status": "ok"}],
"samples": [_sample(offset), _sample(offset + 1)],
}
class MergeBrowserBenchmarkTests(unittest.TestCase):
def test_merges_complete_one_session_reports(self) -> None:
with tempfile.TemporaryDirectory() as directory:
first = Path(directory) / "first.json"
second = Path(directory) / "second.json"
first.write_text(json.dumps(_report("first", 0)), encoding="utf-8")
second.write_text(json.dumps(_report("second", 0)), encoding="utf-8")
merged = merge_tool.merge_reports([first, second])
self.assertEqual(merged["sessions"], 2)
self.assertEqual(len(merged["samples"]), 4)
self.assertEqual({sample["session_start"] for sample in merged["samples"]}, {1, 2})
self.assertEqual(merged["merge"]["source_count"], 2)
self.assertEqual(merged["summary"]["successful_samples"], 4)
def test_rejects_mismatched_provider(self) -> None:
first_report = _report("first", 0)
second_report = _report("second", 0)
second_report["requested_provider"] = "webgpu"
with tempfile.TemporaryDirectory() as directory:
first = Path(directory) / "first.json"
second = Path(directory) / "second.json"
first.write_text(json.dumps(first_report), encoding="utf-8")
second.write_text(json.dumps(second_report), encoding="utf-8")
with self.assertRaises(ValueError):
merge_tool.merge_reports([first, second])
if __name__ == "__main__":
unittest.main()
|