# License: Apache-2.0 License # Created by: Patrick Lumbantobing, VertoX-AI # Copyright (c) 2026 VertoX-AI. All rights reserved. # # This work is licensed under the Apache-2.0 License. # To view a copy of this license, visit # https://www.apache.org/licenses/LICENSE-2.0 """Focused tests for deterministic full-word timing materialization.""" from __future__ import annotations import csv import hashlib import io import json import zipfile from decimal import Decimal from pathlib import Path from typing import Any import pytest from speech_text_data_aligners.application.materialize import ( MaterializationError, TokenSpan, _decimal, _find_member, _json_bytes, _read_cases, _read_json_bytes, _read_json_file, _read_jsonl, _validate_capsule, compute_producer_implementation_sha256, confidence_label, decimal_consensus, materialize_full_word_timing, normalize_readiness_text, publish_materialized_outputs, reconcile_token_word_indices, tokenize_readiness_text, ) _CASE_IDS = ( "mim_02484_01416200252", "mim_02484_01200690643", "mim_03397_00876280561", "mim_02484_01233863243", "mim_04310_01682036526", "mim_04310_01670539014", "mim_03397_00399564759", "mim_04310_01403424076", "mim_02484_01977040283", "mim_02484_00309012605", "mim_02484_00496390374", "mim_02484_00622999980", "mim_02484_00741161044", "mim_02484_00042684773", "mim_04310_00642611087", "mim_02484_01089412305", "mim_02484_01070019332", "mim_03397_00363796383", "mim_02484_00076618054", "mim_03397_01686943957", "mim_02484_01464080210", "mim_02484_01574758769", "mim_02484_02070436300", "mim_03397_01156264928", "mim_03397_00492380451", "mif_03397_01795413856", "mif_03397_01151916618", "mif_03397_00066725045", "mif_03397_00936175599", "mif_03397_00076176932", "mif_03397_01656162863", "mif_03397_01739171182", "mif_03397_00445744175", "mif_03397_01441900102", "mif_02484_00381972222", "mif_03397_00992306079", ) def _words(*spans: tuple[str, int, int]) -> list[dict[str, Any]]: """Build backend-neutral lexical word records for mapping tests.""" return [ {"lexical_word": {"text": text, "start_char": start, "end_char": end}} for text, start, end in spans ] def test_normalization_and_tokenizer_retain_source_spans() -> None: """Normalize NFC/whitespace and split ordinal-like tokens generically.""" normalized = normalize_readiness_text(" Cafe\u0301\u00a0 17th ") assert normalized.text == "Café 17th" tokens = tokenize_readiness_text(" Cafe\u0301\u00a0 17th ") assert [token.text for token in tokens] == ["Caf", "17", "th"] assert tokens[1].raw_start < tokens[2].raw_start with pytest.raises(MaterializationError, match="lexical content"): normalize_readiness_text(" \t\n") with pytest.raises(MaterializationError, match="empty"): normalize_readiness_text("") with pytest.raises(MaterializationError, match="lexical tokens"): tokenize_readiness_text("!!!") def test_reconcile_source_character_mapping_variants() -> None: """Cover one-to-one, one-to-many, many-to-one, and repeated words.""" source = "go go alpha" tokens = tokenize_readiness_text(source) assert reconcile_token_word_indices(source, tokens[0], _words(("go", 0, 2))) == (0,) assert reconcile_token_word_indices(source, tokens[1], _words(("go", 3, 5))) == (0,) alpha = tokens[2] assert reconcile_token_word_indices( source, alpha, _words(("al", 6, 8), ("pha", 8, 11)) ) == (0, 1) ordinal_tokens = tokenize_readiness_text("17th") ordinal_words = _words(("17th", 0, 4)) assert reconcile_token_word_indices("17th", ordinal_tokens[0], ordinal_words) == ( 0, ) assert reconcile_token_word_indices("17th", ordinal_tokens[1], ordinal_words) == ( 0, ) def test_reconcile_rejects_ambiguous_and_incomplete_coverage() -> None: """Fail closed when source characters are not uniquely covered.""" token = tokenize_readiness_text("abc")[0] with pytest.raises(MaterializationError, match="ambiguously"): reconcile_token_word_indices("abc", token, _words(("ab", 0, 2), ("bc", 1, 3))) with pytest.raises(MaterializationError, match="ambiguously"): reconcile_token_word_indices("abc", token, _words(("ab", 0, 2))) with pytest.raises(MaterializationError, match="ASCII"): reconcile_token_word_indices("!", TokenSpan("!", 0, 1, 0, 1), ()) with pytest.raises(MaterializationError, match="lacks lexical"): reconcile_token_word_indices("abc", token, [{"lexical_word": None}]) with pytest.raises(MaterializationError, match="invalid"): reconcile_token_word_indices( "abc", token, [{"lexical_word": {"start_char": 2, "end_char": 1}}] ) def test_decimal_consensus_and_confidence_thresholds() -> None: """Use Decimal medians and exact 100/250 millisecond boundaries.""" assert decimal_consensus([Decimal("1.0"), Decimal("1.2")]) == Decimal("1.1") assert decimal_consensus( [Decimal("1.0"), Decimal("1.3"), Decimal("1.2")] ) == Decimal("1.2") assert confidence_label([]) == "UNRESOLVED" assert confidence_label([Decimal("1")]) == "SINGLE_BACKEND" assert confidence_label([Decimal("0"), Decimal("0.1"), Decimal("0.1")]) == "HIGH" assert confidence_label([Decimal("0"), Decimal("0.25")]) == "MEDIUM" assert confidence_label([Decimal("0"), Decimal("0.251")]) == "LOW" with pytest.raises(MaterializationError, match="nonnegative"): decimal_consensus([Decimal("-1")]) with pytest.raises(MaterializationError, match="without"): decimal_consensus([]) with pytest.raises(MaterializationError, match="nonnegative"): confidence_label([Decimal("-1")]) def test_materialize_and_publish_are_deterministic(tmp_path: Path) -> None: """Materialize a complete synthetic frozen cohort and enforce publication safety.""" source_zip, readiness_zip, source_manifest = _make_fixture(tmp_path) first = materialize_full_word_timing( source_zip, readiness_zip, source_manifest, "implementation-sha", "config-sha", tmp_path / "run_a", ) second = materialize_full_word_timing( source_zip, readiness_zip, source_manifest, "implementation-sha", "config-sha", tmp_path / "run_b", ) assert first.case_count == 36 assert first.lexical_token_count == 72 assert first.accepted_backend_observations == (("mfa", 72), ("whisperx", 72)) assert first.jsonl_path.read_bytes() == second.jsonl_path.read_bytes() assert first.manifest_path.read_bytes() == second.manifest_path.read_bytes() target = tmp_path / "readiness" assert publish_materialized_outputs(first.jsonl_path.parent, target) == ( ("W1_T_FULL_WORD_TIMING.jsonl", "PUBLISHED"), ("W1_T_FULL_WORD_TIMING_MANIFEST.json", "PUBLISHED"), ) assert publish_materialized_outputs(first.jsonl_path.parent, target) == ( ("W1_T_FULL_WORD_TIMING.jsonl", "ALREADY_PRESENT_IDENTICAL"), ("W1_T_FULL_WORD_TIMING_MANIFEST.json", "ALREADY_PRESENT_IDENTICAL"), ) (target / "W1_T_FULL_WORD_TIMING.jsonl").write_bytes(b"different") with pytest.raises(MaterializationError, match="conflict"): publish_materialized_outputs(first.jsonl_path.parent, target) def test_materialize_rejects_frozen_w1_timing_mismatch(tmp_path: Path) -> None: """Fail closed when generated Decimal timing differs from frozen W1.""" _assert_frozen_reference_mismatch( tmp_path, {"reference_semantic_ready_time_s": 0.4}, "timing mismatch", ) def test_materialize_rejects_frozen_w1_backend_count_mismatch(tmp_path: Path) -> None: """Fail closed when generated backend count differs from frozen W1.""" _assert_frozen_reference_mismatch( tmp_path, {"valid_backend_count": 3}, "backend-count mismatch", ) def test_materialize_rejects_frozen_w1_confidence_mismatch(tmp_path: Path) -> None: """Fail closed when generated confidence differs from frozen W1.""" _assert_frozen_reference_mismatch( tmp_path, {"confidence": "HIGH"}, "confidence mismatch", ) def test_materialize_rejects_frozen_w1_lexical_index_mismatch(tmp_path: Path) -> None: """Fail closed when mapped Stage-2 index differs from frozen W1.""" _assert_frozen_reference_mismatch( tmp_path, {"reference_semantic_ready_word_count": 1}, "lexical-index mismatch", ) def test_producer_hash_rejects_invalid_paths(tmp_path: Path) -> None: """Hash producer bytes canonically and reject duplicate/traversal paths.""" implementation = tmp_path / "implementation.py" implementation.write_bytes(b"pass\n") assert compute_producer_implementation_sha256(tmp_path, ("implementation.py",)) with pytest.raises(MaterializationError, match="unique"): compute_producer_implementation_sha256( tmp_path, ("implementation.py", "implementation.py") ) with pytest.raises(MaterializationError, match="relative"): compute_producer_implementation_sha256(tmp_path, ("../implementation.py",)) with pytest.raises(MaterializationError, match="missing"): compute_producer_implementation_sha256(tmp_path, ("missing.py",)) def test_serialization_and_source_readers_fail_closed(tmp_path: Path) -> None: """Reject malformed JSON, unsupported values, and ambiguous archive members.""" with pytest.raises(MaterializationError, match="malformed"): _read_json_bytes(b"not-json") tmp_path.joinpath("array.json").write_bytes(b"[]") with pytest.raises(MaterializationError, match="not an object"): _read_json_file(tmp_path.joinpath("array.json")) with pytest.raises(MaterializationError, match="not an object"): _read_json_file(tmp_path / "array.json") with pytest.raises(MaterializationError, match="not an object"): _read_jsonl(b"[]\n") with pytest.raises(MaterializationError, match="timestamp"): _decimal("not-a-number") with pytest.raises(MaterializationError, match="JSON number"): _decimal(1.0) with pytest.raises(MaterializationError, match="finite"): _decimal(Decimal("NaN")) with pytest.raises(MaterializationError, match="non-finite"): _json_bytes(Decimal("NaN")) with pytest.raises(MaterializationError, match="non-finite"): _json_bytes(float("inf")) with pytest.raises(MaterializationError, match="serializable"): _json_bytes(object()) archive_path = tmp_path / "members.zip" with zipfile.ZipFile(archive_path, "w") as archive: archive.writestr("a/item.json", b"{}") archive.writestr("b/item.json", b"{}") with ( zipfile.ZipFile(archive_path) as archive, pytest.raises(MaterializationError, match="expected one"), ): _find_member(archive, "*/item.json", "item") def test_empty_or_nonempty_output_directories_are_rejected(tmp_path: Path) -> None: """Keep materialization runs isolated in empty output directories.""" source_zip, readiness_zip, source_manifest = _make_fixture(tmp_path) output = tmp_path / "output" output.mkdir() output.joinpath("existing").write_text("x", encoding="utf-8") with pytest.raises(MaterializationError, match="empty"): materialize_full_word_timing( source_zip, readiness_zip, source_manifest, "implementation-sha", "config-sha", output, ) with pytest.raises(MaterializationError, match="required"): publish_materialized_outputs(tmp_path / "missing", tmp_path / "target") def test_capsule_and_case_validation_fail_closed(tmp_path: Path) -> None: """Reject invalid frozen case shape, identity, provenance, and word timing.""" invalid_archive = tmp_path / "invalid_readiness.zip" with zipfile.ZipFile(invalid_archive, "w") as archive: archive.writestr( "inputs/STAGE2_36_ENGLISH_ONLY.csv", "audio_id,en_transcript\nonly,text\n" ) with ( zipfile.ZipFile(invalid_archive) as archive, pytest.raises(MaterializationError, match="exactly 36"), ): _read_cases(archive) base = _capsule("mfa", "audio", "Alpha beta", "sha") with pytest.raises(MaterializationError, match="identity"): _validate_capsule({}, "mfa", "audio", "sha", "Alpha beta") not_authoritative = dict(base, reference_text_authoritative=False) with pytest.raises(MaterializationError, match="authoritative"): _validate_capsule(not_authoritative, "mfa", "audio", "sha", "Alpha beta") no_words = dict(base, result={"provenance": {"backend_name": "mfa"}}) with pytest.raises(MaterializationError, match="backend-neutral"): _validate_capsule(no_words, "mfa", "audio", "sha", "Alpha beta") bad_provenance = dict( base, result=dict(base["result"], provenance={"backend_name": "other"}) ) with pytest.raises(MaterializationError, match="provenance"): _validate_capsule(bad_provenance, "mfa", "audio", "sha", "Alpha beta") bad_word = dict(base, result=dict(base["result"], words=["not-a-word"])) with pytest.raises(MaterializationError, match="backend word"): _validate_capsule(bad_word, "mfa", "audio", "sha", "Alpha beta") bad_lexical = dict( base, result=dict(base["result"], words=[{"lexical_word": None}]) ) with pytest.raises(MaterializationError, match="lexical span"): _validate_capsule(bad_lexical, "mfa", "audio", "sha", "Alpha beta") bad_span = dict( base, result=dict( base["result"], words=[{"lexical_word": {"start_char": 0, "end_char": 99}}] ), ) with pytest.raises(MaterializationError, match="source bounds"): _validate_capsule(bad_span, "mfa", "audio", "sha", "Alpha beta") bad_text = dict( base, result=dict( base["result"], words=[ { "lexical_word": {"start_char": 0, "end_char": 5, "text": "Wrong"}, "start_s": 0, "end_s": 1, } ], ), ) with pytest.raises(MaterializationError, match="differs"): _validate_capsule(bad_text, "mfa", "audio", "sha", "Alpha beta") bad_time = dict( base, result=dict( base["result"], words=[ { "lexical_word": {"start_char": 0, "end_char": 5, "text": "Alpha"}, "start_s": 2, "end_s": 1, } ], ), ) with pytest.raises(MaterializationError, match="non-monotonic"): _validate_capsule(bad_time, "mfa", "audio", "sha", "Alpha beta") empty_words = dict(base, result=dict(base["result"], words=[])) with pytest.raises(MaterializationError, match="no words"): _validate_capsule(empty_words, "mfa", "audio", "sha", "Alpha beta") def _assert_frozen_reference_mismatch( tmp_path: Path, overrides: dict[str, Any], message: str ) -> None: """Run the production materializer and require one frozen W1 mismatch.""" source_zip, readiness_zip, source_manifest = _make_fixture(tmp_path, overrides) output = tmp_path / "rejected_output" with pytest.raises(MaterializationError, match=message): materialize_full_word_timing( source_zip, readiness_zip, source_manifest, "implementation-sha", "config-sha", output, ) assert list(output.iterdir()) == [] def _make_fixture( tmp_path: Path, frozen_overrides: dict[str, Any] | None = None ) -> tuple[Path, Path, Path]: """Create immutable synthetic archives with the required 36-case shape.""" transcript = "Alpha beta" transcript_sha = hashlib.sha256(transcript.encode()).hexdigest() frozen_values = { "reference_semantic_ready_time_s": 0.3, "reference_semantic_ready_word_count": 2, "valid_backend_count": 2, "confidence": "MEDIUM", } if frozen_overrides is not None: frozen_values.update(frozen_overrides) timing_rows = [ { "audio_id": audio_id, "original_en_transcript_sha256": transcript_sha, **frozen_values, "mapping": {"start_char": 6, "end_char": 10}, "w1_normalized_transcript_sha256": transcript_sha, } for audio_id in _CASE_IDS ] timing_bytes = b"".join( (json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n").encode() for row in timing_rows ) source_zip = tmp_path / "source.zip" accepted: dict[str, dict[str, str]] = {"mfa": {}, "whisperx": {}} with zipfile.ZipFile(source_zip, "w") as archive: archive.writestr("review/timing.jsonl", timing_bytes) for backend in accepted: pattern = f"review/{backend}/{{audio_id}}.json" for audio_id in _CASE_IDS: member = pattern.format(audio_id=audio_id) capsule = _capsule(backend, audio_id, transcript, transcript_sha) raw = ( json.dumps(capsule, sort_keys=True, separators=(",", ":")) + "\n" ).encode() archive.writestr(member, raw) accepted[backend][audio_id] = hashlib.sha256(raw).hexdigest() source_hash = hashlib.sha256(source_zip.read_bytes()).hexdigest() manifest = { "accepted_full_word_evidence": { "mfa": { "count": 36, "member_pattern": "review/mfa/*.json", "member_sha256_by_audio_id": accepted["mfa"], }, "whisperx": { "count": 36, "member_pattern": "review/whisperx/*.json", "member_sha256_by_audio_id": accepted["whisperx"], }, }, "frozen_w1t_timing_member": "review/timing.jsonl", "frozen_w1t_timing_sha256": hashlib.sha256(timing_bytes).hexdigest(), "source_review_zip_sha256": source_hash, } source_manifest = tmp_path / "source_manifest.json" source_manifest.write_text(json.dumps(manifest), encoding="utf-8") readiness_zip = tmp_path / "readiness.zip" csv_buffer = io.StringIO() writer = csv.DictWriter(csv_buffer, fieldnames=["audio_id", "en_transcript"]) writer.writeheader() writer.writerows( {"audio_id": audio_id, "en_transcript": transcript} for audio_id in _CASE_IDS ) with zipfile.ZipFile(readiness_zip, "w") as archive: archive.writestr("inputs/STAGE2_36_ENGLISH_ONLY.csv", csv_buffer.getvalue()) archive.writestr("inputs/W1_T_SEMANTIC_READINESS_TIMING.jsonl", timing_bytes) return source_zip, readiness_zip, source_manifest def _capsule( backend: str, audio_id: str, transcript: str, transcript_sha: str ) -> dict[str, Any]: """Build one backend-neutral synthetic capsule.""" return { "audio_id": audio_id, "original_en_transcript_sha256": transcript_sha, "reference_text_authoritative": True, "result": { "provenance": { "backend_name": backend, "backend_version": "test", "config_sha256": "config", "model_id": "model", "model_revision": None, }, "words": [ { "end_s": 0.1, "lexical_word": {"end_char": 5, "start_char": 0, "text": "Alpha"}, "start_s": 0.0, }, { "end_s": 0.3, "lexical_word": {"end_char": 10, "start_char": 6, "text": "beta"}, "start_s": 0.1, }, ], }, }