| |
| |
| |
| |
| |
| |
| |
|
|
| from __future__ import annotations |
|
|
| import os |
| from pathlib import Path |
|
|
| import pytest |
|
|
| from speech_text_data_aligners.core.errors import MetadataError |
| from speech_text_data_aligners.core.extraction import BOUNDARY_COLUMNS |
| from speech_text_data_aligners.data.metadata import ( |
| MetadataTable, |
| read_metadata, |
| write_annotated_metadata, |
| ) |
|
|
|
|
| def valid_annotations(count: int) -> tuple[dict[str, str | float | None], ...]: |
| return tuple( |
| { |
| "first_word_text": "Hello", |
| "first_word_start_s": 0.000001, |
| "first_word_end_s": 0.5, |
| "last_word_text": "world", |
| "last_word_start_s": 1.0, |
| "last_word_end_s": None, |
| } |
| for _ in range(count) |
| ) |
|
|
|
|
| def test_read_and_write_preserve_bom_crlf_quoted_values_and_order( |
| tmp_path: Path, |
| ) -> None: |
| source = tmp_path / "source.csv" |
| source.write_bytes( |
| b"\xef\xbb\xbfline_id,audio_id,en_transcript,ru_translation\r\n" |
| b'L1,a1,"Hello, world",Russian\r\n' |
| b'L1,a2,"multi\r\nline",Exact\r\n' |
| ) |
| table = read_metadata(source) |
| assert table.csv_format.encoding == "utf-8-sig" |
| assert table.csv_format.lineterminator == "\r\n" |
| assert table.row_dict(0)["en_transcript"] == "Hello, world" |
| assert table.row_dict(1)["en_transcript"] == "multi\r\nline" |
|
|
| output = tmp_path / "output.csv" |
| write_annotated_metadata(table, output, valid_annotations(2)) |
| reread = read_metadata(output) |
| assert reread.fieldnames == (*table.fieldnames, *BOUNDARY_COLUMNS) |
| assert reread.rows[0][:4] == table.rows[0] |
| assert reread.row_dict(0)["first_word_start_s"] == "0.000001" |
| assert reread.row_dict(0)["first_word_end_s"] == "0.5" |
| assert reread.row_dict(0)["last_word_end_s"] == "" |
|
|
|
|
| @pytest.mark.parametrize( |
| "content", |
| [ |
| b"", |
| b"line_id,audio_id,en_transcript\nL,a,t\n", |
| b"line_id,audio_id,en_transcript,ru_translation,line_id\n", |
| b"line_id,audio_id,en_transcript,ru_translation\nL,a,t\n", |
| b'line_id,audio_id,en_transcript,ru_translation\nL,a,"unterminated,x\n', |
| b"\xff\xfeinvalid", |
| ], |
| ) |
| def test_read_rejects_empty_missing_duplicate_malformed_and_non_utf8( |
| tmp_path: Path, content: bytes |
| ) -> None: |
| path = tmp_path / "bad.csv" |
| path.write_bytes(content) |
| with pytest.raises(MetadataError): |
| read_metadata(path) |
|
|
|
|
| def test_read_rejects_missing_path(tmp_path: Path) -> None: |
| with pytest.raises(MetadataError): |
| read_metadata(tmp_path / "missing.csv") |
|
|
|
|
| def test_write_rejects_unsafe_schema_counts_keys_and_overwrite( |
| metadata_fixture: tuple[Path, Path], tmp_path: Path |
| ) -> None: |
| source, _ = metadata_fixture |
| table = read_metadata(source) |
| output = tmp_path / "out.csv" |
| output.write_text("existing", encoding="utf-8") |
| with pytest.raises(MetadataError, match="source metadata"): |
| write_annotated_metadata(table, source, valid_annotations(2)) |
| with pytest.raises(MetadataError, match="count"): |
| write_annotated_metadata(table, tmp_path / "other.csv", valid_annotations(1)) |
| with pytest.raises(MetadataError, match="six"): |
| write_annotated_metadata(table, tmp_path / "other.csv", ({"bad": "x"},) * 2) |
| with pytest.raises(MetadataError, match="overwrite"): |
| write_annotated_metadata(table, output, valid_annotations(2)) |
|
|
| enriched = MetadataTable( |
| table.source_path, |
| (*table.fieldnames, BOUNDARY_COLUMNS[0]), |
| tuple((*row, "") for row in table.rows), |
| table.csv_format, |
| ) |
| with pytest.raises(MetadataError, match="already"): |
| write_annotated_metadata( |
| enriched, tmp_path / "enriched.csv", valid_annotations(2) |
| ) |
|
|
|
|
| def test_write_overwrites_when_authorized( |
| metadata_fixture: tuple[Path, Path], tmp_path: Path |
| ) -> None: |
| source, _ = metadata_fixture |
| table = read_metadata(source) |
| output = tmp_path / "out.csv" |
| output.write_text("old", encoding="utf-8") |
| write_annotated_metadata(table, output, valid_annotations(2), overwrite=True) |
| assert read_metadata(output).row_dict(0)["first_word_text"] == "Hello" |
|
|
|
|
| def test_write_translates_publication_oserror( |
| metadata_fixture: tuple[Path, Path], tmp_path: Path, monkeypatch: pytest.MonkeyPatch |
| ) -> None: |
| source, _ = metadata_fixture |
| table = read_metadata(source) |
|
|
| def fail_replace(source_path: Path, destination_path: Path) -> None: |
| raise OSError(source_path, destination_path) |
|
|
| monkeypatch.setattr(os, "replace", fail_replace) |
| with pytest.raises(MetadataError, match="publish"): |
| write_annotated_metadata(table, tmp_path / "out.csv", valid_annotations(2)) |
| assert not list(tmp_path.glob(".out.csv.*.tmp")) |
|
|
