tests/unit/application/test_materialize.py

# 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,
                },
            ],
        },
    }