| |
| |
| |
| |
| |
| |
| |
|
|
| from __future__ import annotations |
|
|
| import csv |
| from pathlib import Path |
|
|
| import pytest |
|
|
| from speech_text_data_aligners.application.benchmark import ( |
| BenchmarkOptions, |
| benchmark_backend, |
| benchmark_backends, |
| ) |
| from speech_text_data_aligners.core.errors import BenchmarkError, SpeechTextAlignerError |
| from tests.fakes import FakeBackend |
|
|
|
|
| def write_reference(path: Path, rows: list[tuple[str, str, str, str]]) -> Path: |
| with path.open("w", encoding="utf-8", newline="") as stream: |
| writer = csv.writer(stream, lineterminator="\n") |
| writer.writerow( |
| ( |
| "line_id", |
| "audio_id", |
| "en_transcript", |
| "ru_translation", |
| "first_word_text", |
| "first_word_end_s", |
| ) |
| ) |
| for line_id, audio_id, first_word, end in rows: |
| writer.writerow((line_id, audio_id, "ignored", "ignored", first_word, end)) |
| return path |
|
|
|
|
| def test_benchmark_runs_identical_requests_and_builds_gold_and_pairwise_report( |
| metadata_fixture: tuple[Path, Path], tmp_path: Path |
| ) -> None: |
| metadata, audio_dir = metadata_fixture |
| reference = write_reference( |
| tmp_path / "gold.csv", |
| [("L1", "a1", "Hello", "0.2"), ("L1", "a2", "Can't", "0.2")], |
| ) |
| left = FakeBackend(offset=0.0) |
| right = FakeBackend(offset=0.1) |
| report = benchmark_backends( |
| backends={"left": left, "right": right}, |
| metadata_csv=metadata, |
| reference_csv=reference, |
| audio_dir=audio_dir, |
| ) |
| assert left.batches[0] == right.batches[0] |
| assert [request.transcript for request in left.batches[0]] == [ |
| "Hello world.", |
| "Can't stop", |
| ] |
| assert report.backend_reports[0].gold_metrics.exact_text_match_count == 2 |
| assert report.backend_reports[0].gold_metrics.first_end_errors.mae == 0.0 |
| assert report.backend_reports[0].observations[0].native_artifacts == () |
| assert report.pairwise_metrics[0].first_end.mae == pytest.approx(0.1) |
|
|
|
|
| def test_benchmark_single_wrapper_and_sparse_missing_reference( |
| metadata_fixture: tuple[Path, Path], tmp_path: Path |
| ) -> None: |
| metadata, audio_dir = metadata_fixture |
| reference = write_reference(tmp_path / "gold.csv", [("L1", "a1", "", "")]) |
| backend = FakeBackend() |
| report = benchmark_backend( |
| backend=backend, |
| metadata_csv=metadata, |
| reference_csv=reference, |
| audio_dir=audio_dir, |
| ) |
| metrics = report.backend_reports[0].gold_metrics |
| assert metrics.referenced_rows == 1 |
| assert metrics.missing_reference_rows == 1 |
| assert metrics.exact_text_match_rate is None |
| assert report.pairwise_metrics == () |
|
|
|
|
| @pytest.mark.parametrize("bad_end", ["bad", "nan", "-0.1", "inf"]) |
| def test_benchmark_rejects_invalid_reference_timestamp( |
| metadata_fixture: tuple[Path, Path], tmp_path: Path, bad_end: str |
| ) -> None: |
| metadata, audio_dir = metadata_fixture |
| reference = write_reference(tmp_path / "gold.csv", [("L1", "a1", "Hello", bad_end)]) |
| with pytest.raises(BenchmarkError, match="invalid"): |
| benchmark_backend( |
| backend=FakeBackend(), |
| metadata_csv=metadata, |
| reference_csv=reference, |
| audio_dir=audio_dir, |
| ) |
|
|
|
|
| def test_benchmark_rejects_duplicate_and_extra_reference_rows( |
| metadata_fixture: tuple[Path, Path], tmp_path: Path |
| ) -> None: |
| metadata, audio_dir = metadata_fixture |
| duplicate = write_reference( |
| tmp_path / "duplicate.csv", |
| [("L1", "a1", "Hello", "0.2"), ("L1", "a1", "Hello", "0.2")], |
| ) |
| with pytest.raises(BenchmarkError, match="duplicate"): |
| benchmark_backend( |
| backend=FakeBackend(), |
| metadata_csv=metadata, |
| reference_csv=duplicate, |
| audio_dir=audio_dir, |
| ) |
| extra = write_reference(tmp_path / "extra.csv", [("X", "other", "A", "0.2")]) |
| with pytest.raises(BenchmarkError, match="absent"): |
| benchmark_backend( |
| backend=FakeBackend(), |
| metadata_csv=metadata, |
| reference_csv=extra, |
| audio_dir=audio_dir, |
| ) |
|
|
|
|
| def test_benchmark_rejects_missing_reference_fields_and_backend_mapping( |
| metadata_fixture: tuple[Path, Path], tmp_path: Path |
| ) -> None: |
| metadata, audio_dir = metadata_fixture |
| bad_reference = tmp_path / "bad.csv" |
| bad_reference.write_text( |
| "line_id,audio_id,en_transcript,ru_translation\nL1,a1,x,y\n", |
| encoding="utf-8", |
| ) |
| with pytest.raises(BenchmarkError, match="lacks"): |
| benchmark_backend( |
| backend=FakeBackend(), |
| metadata_csv=metadata, |
| reference_csv=bad_reference, |
| audio_dir=audio_dir, |
| ) |
| valid = write_reference(tmp_path / "valid.csv", []) |
| with pytest.raises(BenchmarkError, match="at least"): |
| benchmark_backends( |
| backends={}, metadata_csv=metadata, reference_csv=valid, audio_dir=audio_dir |
| ) |
| with pytest.raises(BenchmarkError, match="cannot be empty"): |
| benchmark_backends( |
| backends={"": FakeBackend()}, |
| metadata_csv=metadata, |
| reference_csv=valid, |
| audio_dir=audio_dir, |
| ) |
|
|
|
|
| def test_benchmark_continue_mode_records_failures_and_strict_mode_aborts( |
| metadata_fixture: tuple[Path, Path], tmp_path: Path |
| ) -> None: |
| metadata, audio_dir = metadata_fixture |
| reference = write_reference(tmp_path / "gold.csv", [("L1", "a1", "Hello", "0.2")]) |
| report = benchmark_backend( |
| backend=FakeBackend(fail_ids={"000000"}), |
| metadata_csv=metadata, |
| reference_csv=reference, |
| audio_dir=audio_dir, |
| ) |
| assert report.backend_reports[0].gold_metrics.failure_count == 1 |
| assert report.backend_reports[0].observations[0].failure_code == "fake_failure" |
| with pytest.raises(SpeechTextAlignerError): |
| benchmark_backend( |
| backend=FakeBackend(fail_ids={"000000"}), |
| metadata_csv=metadata, |
| reference_csv=reference, |
| audio_dir=audio_dir, |
| options=BenchmarkOptions(strict=True), |
| ) |
|
|
