File size: 3,667 Bytes
33d23e1 | 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 104 105 106 107 108 109 110 | # SPDX-FileCopyrightText: © 2026 Tenstorrent USA, Inc.
# SPDX-License-Identifier: Apache-2.0
import json
import pytest
from models.tt_dit.utils.vbench import assert_scores
from models.tt_dit.utils.vbench_bundle import METRICS, aggregate, digest
@pytest.fixture
def gate(tmp_path):
bundle = tmp_path / "input"
bundle.mkdir()
manifest = {
"prompt": "test",
"thresholds": {metric: 0.8 for metric in METRICS},
"clips": [{"name": f"seed_{i}.mp4", "shape": [145, 1088, 1920], "sha256": "unused"} for i in range(5)],
}
(bundle / "manifest.json").write_text(json.dumps(manifest))
results = tmp_path / "scores"
results.mkdir()
for i in range(5):
scores = {metric: 0.9 for metric in METRICS}
scores["dynamic_degree"] = int(i != 0)
row = {"index": i, "manifest_sha256": digest(bundle / "manifest.json"), "scores": scores}
(results / f"score_{i}.json").write_text(json.dumps(row))
return bundle, results
def test_gate_averages_all_five_seeds(gate):
assert aggregate(*gate)["dynamic_degree"] == 0.8
@pytest.mark.parametrize("failure", ["missing", "duplicate", "metric", "nan", "wrong_input", "below_floor"])
def test_incomplete_or_bad_quality_cannot_pass(gate, failure, expect_error):
bundle, results = gate
path = results / "score_4.json"
row = json.loads(path.read_text())
if failure == "missing":
path.unlink()
else:
if failure == "duplicate":
row["index"] = 0
elif failure == "metric":
del row["scores"]["motion_smoothness"]
elif failure == "nan":
row["scores"]["motion_smoothness"] = float("nan")
elif failure == "wrong_input":
row["manifest_sha256"] = "different build"
else:
row["scores"]["dynamic_degree"] = 0
path.write_text(json.dumps(row))
with expect_error((ValueError, AssertionError), "VBench"):
aggregate(bundle, results)
@pytest.mark.parametrize("score", [float("nan"), float("inf"), -1])
def test_invalid_scores_fail(score, expect_error):
with expect_error(AssertionError, "VBench quality gate failed"):
assert_scores({"quality": score}, {"quality": 0.8})
def test_parallel_worker_failure_cannot_reach_aggregate(gate, monkeypatch, expect_error):
from types import SimpleNamespace
from models.tt_dit.utils import vbench_bundle
submitted = []
def fail():
raise RuntimeError("scoring worker failed")
class Pool:
def __init__(self, **kwargs):
pass
def __enter__(self):
return self
def __exit__(self, *args):
pass
def submit(self, fn, bundle, index, output):
submitted.append(index)
return SimpleNamespace(result=fail)
def aggregate(*args):
raise AssertionError("must not aggregate after a worker failure")
monkeypatch.setattr(vbench_bundle, "ProcessPoolExecutor", Pool)
monkeypatch.setattr(vbench_bundle, "aggregate", aggregate)
with expect_error(RuntimeError, "scoring worker failed"):
vbench_bundle.evaluate(gate[0])
assert submitted == list(range(5))
@pytest.mark.parametrize("width", [-1, 959, 1922, "960"])
def test_invalid_temporal_policy_cannot_pass(gate, width, expect_error):
from models.tt_dit.utils.vbench_bundle import read_manifest
bundle, _ = gate
path = bundle / "manifest.json"
manifest = json.loads(path.read_text())
manifest["temporal_width"] = width
path.write_text(json.dumps(manifest))
with expect_error(ValueError, "Invalid VBench temporal width"):
read_manifest(bundle)
|