Skip to content

test_evaluator_paths

Path-input variants of the public Evaluator API.

The worker boundary accepts video / reference as either a pre-loaded (T, C, H, W) tensor or a path-like (str / Path). These tests pin the path-form so future refactors don't accidentally re-require pre-loaded tensors.

Classes

Functions

fastvideo.tests.eval.test_evaluator_paths.test_dispatcher_holds_paths_not_tensors_in_queue 

test_dispatcher_holds_paths_not_tensors_in_queue(tmp_path)

Memory invariant: when many paths are passed, the queued samples are tiny strings, not full tensors. Verify by checking the length of the per-sample reference set the dispatcher materializes.

Source code in fastvideo/tests/eval/test_evaluator_paths.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def test_dispatcher_holds_paths_not_tensors_in_queue(tmp_path):
    """Memory invariant: when many paths are passed, the queued samples
    are tiny strings, not full tensors. Verify by checking the length
    of the per-sample reference set the dispatcher materializes."""
    torch.manual_seed(1)
    n = 8
    paths: list[Path] = []
    for i in range(n):
        t = torch.rand(_T, _C, _H, _W)
        p = tmp_path / f"clip_{i}.mp4"
        _write_tensor_as_mp4(t, p)
        paths.append(p)

    samples = [{"video": str(p), "reference": str(p)} for p in paths]
    # Each sample dict is just two strings — no tensor allocations until
    # the worker's _resolve_video_input runs.
    for s in samples:
        assert isinstance(s["video"], str)
        assert isinstance(s["reference"], str)

    ev = create_evaluator(metrics=["common.psnr"], device="cpu")
    try:
        out = ev.evaluate(samples=samples)
    finally:
        ev.shutdown()
    assert len(out) == n
    # Self-paired ⇒ all PSNRs should be very high.
    for row in out:
        assert row["common.psnr"].score > 50.0

fastvideo.tests.eval.test_evaluator_paths.test_missing_path_surfaces_as_exception 

test_missing_path_surfaces_as_exception(evaluator, tmp_path)

Decode failures must propagate, not silently produce a None score.

Source code in fastvideo/tests/eval/test_evaluator_paths.py 
123
124
125
126
127
def test_missing_path_surfaces_as_exception(evaluator, tmp_path):
    """Decode failures must propagate, not silently produce a None score."""
    bogus = tmp_path / "does_not_exist.mp4"
    with pytest.raises(Exception):
        evaluator.evaluate(video=str(bogus), reference=str(bogus))

fastvideo.tests.eval.test_evaluator_paths.test_one_shot_evaluate_accepts_paths 

test_one_shot_evaluate_accepts_paths(video_paths)

The top-level fastvideo.eval.evaluate helper also flows paths.

Source code in fastvideo/tests/eval/test_evaluator_paths.py 
115
116
117
118
119
120
def test_one_shot_evaluate_accepts_paths(video_paths):
    """The top-level ``fastvideo.eval.evaluate`` helper also flows paths."""
    paths, _ = video_paths
    out = evaluate(generated=str(paths[0]), reference=str(paths[0]),
                   metrics=["common.psnr"], device="cpu")
    assert out["common.psnr"].score > 50.0

fastvideo.tests.eval.test_evaluator_paths.test_path_form_score_matches_tensor_form 

test_path_form_score_matches_tensor_form(evaluator, video_paths)

Loading via path must produce the same score as loading via the public load_video helper and passing the tensor in directly.

Source code in fastvideo/tests/eval/test_evaluator_paths.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
def test_path_form_score_matches_tensor_form(evaluator, video_paths):
    """Loading via path must produce the same score as loading via the
    public ``load_video`` helper and passing the tensor in directly."""
    from fastvideo.eval.io import load_video

    paths, _ = video_paths
    via_path = evaluator.evaluate(video=str(paths[0]), reference=str(paths[0]))
    tensor = load_video(str(paths[0]))
    via_tensor = evaluator.evaluate(video=tensor, reference=tensor)
    assert via_path["common.psnr"].score == pytest.approx(
        via_tensor["common.psnr"].score, abs=1e-4)
    assert via_path["common.ssim"].score == pytest.approx(
        via_tensor["common.ssim"].score, abs=1e-4)

fastvideo.tests.eval.test_evaluator_paths.test_samples_list_can_mix_paths_and_tensors 

test_samples_list_can_mix_paths_and_tensors(evaluator, video_paths)

A single samples call can mix path and tensor entries.

Source code in fastvideo/tests/eval/test_evaluator_paths.py 
87
88
89
90
91
92
93
94
95
96
97
def test_samples_list_can_mix_paths_and_tensors(evaluator, video_paths):
    """A single ``samples`` call can mix path and tensor entries."""
    paths, tensors = video_paths
    samples = [
        {"video": str(paths[0]), "reference": tensors[0]},   # path + tensor
        {"video": tensors[1], "reference": str(paths[1])},   # tensor + path
    ]
    out = evaluator.evaluate(samples=samples)
    assert len(out) == 2
    for row in out:
        assert row["common.psnr"].score > 0.0

fastvideo.tests.eval.test_evaluator_paths.video_paths 

video_paths(tmp_path)

Two reproducible mp4s on disk + their pre-loaded tensors for parity.

Source code in fastvideo/tests/eval/test_evaluator_paths.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
@pytest.fixture
def video_paths(tmp_path):
    """Two reproducible mp4s on disk + their pre-loaded tensors for parity."""
    torch.manual_seed(0)
    paths: list[Path] = []
    tensors: list[torch.Tensor] = []
    for i in range(2):
        t = torch.rand(_T, _C, _H, _W)
        p = tmp_path / f"clip_{i}.mp4"
        _write_tensor_as_mp4(t, p)
        paths.append(p)
        tensors.append(t)
    return paths, tensors