UmbraAudio/tests/test_generation/test_language_wiring.py

"""The per-utterance language tag must reach the TTS engine and the marker.

Spec 31's source axis: an utterance tagged with the language it is *written in*
must be voiced in that language, and the marker must record it so a later
retag invalidates the clip instead of leaving English phonology behind an
unchanged text.
"""

from __future__ import annotations

import json
import sqlite3
from pathlib import Path
from typing import Any

import pytest

from umbraaudio.generation import chapter_builder as cb


class _FakeConfig:
    def __init__(self, data_dir: Path) -> None:
        self.data_dir = data_dir
        self.project_dir = data_dir.parent
        self.tts = type(
            "T",
            (),
            {"qwen_tts_url": "http://127.0.0.1:1", "qwen_tts_helper_url": ""},
        )()


@pytest.fixture
def project(tmp_path: Path) -> Path:
    """A minimal project DB with one English line and one Spanish line."""
    data_dir = tmp_path / ".umbra"
    data_dir.mkdir()
    con = sqlite3.connect(str(data_dir / "project.db"))
    con.executescript(
        """
        CREATE TABLE project (id INTEGER PRIMARY KEY, name TEXT, source_language TEXT DEFAULT 'en');
        CREATE TABLE chapters (
            id INTEGER PRIMARY KEY,
            chapter_index INTEGER,
            title TEXT,
            source_text TEXT
        );
        CREATE TABLE utterances (
            id INTEGER PRIMARY KEY,
            chapter_id INTEGER,
            utterance_index INTEGER,
            text TEXT,
            speaker_name TEXT,
            emotion TEXT,
            pace TEXT,
            instruct TEXT,
            type TEXT,
            source_unit_key TEXT,
            language_code TEXT,
            language_origin TEXT,
            context_before TEXT,
            context_after TEXT
        );
        INSERT INTO project (id, name, source_language) VALUES (1, 'Test', 'en');
        INSERT INTO chapters (id, chapter_index, title, source_text)
            VALUES (1, 0, 'One', 'The ship landed. Hola, amigo.');
        INSERT INTO utterances
            (id, chapter_id, utterance_index, text, speaker_name, type, language_code)
        VALUES
            (1, 1, 0, 'The ship landed.', 'Narrator', 'narration', NULL),
            (2, 1, 1, 'Hola, amigo.', 'Kabir', 'narration', 'es');
        """
    )
    con.commit()
    con.close()
    return tmp_path


def _builder(project: Path) -> Any:
    return cb.ChapterBuilder(_FakeConfig(project / ".umbra"), chapter_id=1)


def test_untranslated_line_uses_the_project_source_language(project: Path) -> None:
    builder = _builder(project)
    missing = builder._validate_utterances()
    # Both lines need generating (no audio on disk yet); the point is that
    # the validation pass did not crash and found the right chapter.
    assert len(missing) == 2


def test_retagged_line_goes_stale_even_when_the_text_is_unchanged(project: Path) -> None:
    """The marker records the language, so a retag invalidates the clip."""
    from umbraaudio.generation.staleness import write_marker

    audio_dir = project / ".umbra" / "chapter_audio" / "ch1"
    audio_dir.mkdir(parents=True)
    wav = audio_dir / "utt_0001.wav"
    wav.write_bytes(b"RIFF----WAVEfmt ")

    # The clip was rendered as English.
    write_marker("Hola, amigo.", wav, speaker="Kabir", language="en")

    builder = _builder(project)
    missing = builder._validate_utterances()
    indexes = {u["utterance_index"] for u in missing}
    assert 1 in indexes, "a Spanish line rendered in English must be regenerated"


def test_a_line_already_rendered_in_its_tagged_language_is_current(project: Path) -> None:
    from umbraaudio.generation.staleness import write_marker

    audio_dir = project / ".umbra" / "chapter_audio" / "ch1"
    audio_dir.mkdir(parents=True)

    english = audio_dir / "utt_0000.wav"
    english.write_bytes(b"RIFF----WAVEfmt ")
    write_marker("The ship landed.", english, speaker="Narrator", language="en")

    spanish = audio_dir / "utt_0001.wav"
    spanish.write_bytes(b"RIFF----WAVEfmt ")
    write_marker("Hola, amigo.", spanish, speaker="Kabir", language="es")

    builder = _builder(project)
    missing = builder._validate_utterances()
    assert missing == [], f"unexpected regeneration: {missing}"


def test_legacy_marker_without_a_language_is_still_current_for_english(
    project: Path,
) -> None:
    """Existing clips must not all be invalidated by adding the tag.

    A marker written before the language tag existed records no language;
    that audio was English, so absence means English (spec 31 section 10).
    """
    audio_dir = project / ".umbra" / "chapter_audio" / "ch1"
    audio_dir.mkdir(parents=True)
    wav = audio_dir / "utt_0000.wav"
    wav.write_bytes(b"RIFF----WAVEfmt ")

    # Simulate a legacy marker: no language key at all.
    marker = wav.with_suffix(".wav.txt")
    marker.write_text(json.dumps({"speaker": "Narrator", "text": "The ship landed."}))

    builder = _builder(project)
    missing = builder._validate_utterances()
    indexes = {u["utterance_index"] for u in missing}
    assert 0 not in indexes


def test_engine_receives_the_language_the_line_is_written_in(
    project: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The language must reach the TTS engine, not just the marker."""
    import numpy as np
    import soundfile as sf

    import umbraaudio.voices.qwen_tts as qwen_tts

    calls: list[dict[str, Any]] = []

    class _FakeEngine:
        def __init__(self, *args: Any, **kwargs: Any) -> None:
            pass

        def build_voice_prompt(self, *args: Any, **kwargs: Any) -> str:
            return "prompt-slug"

        def generate(self, **kwargs: Any) -> Path:
            calls.append(kwargs)
            out = kwargs["output_path"]
            out.parent.mkdir(parents=True, exist_ok=True)
            sf.write(str(out), np.zeros(2400, dtype=np.float32), 24000)
            return out

    monkeypatch.setattr(qwen_tts, "QwenTTSEngine", _FakeEngine)

    builder = _builder(project)
    # The health monitor would race the generation loop against a TTS server
    # that does not exist; the builder's own flag is what matters here.
    monkeypatch.setattr(builder, "_start_health_monitor", lambda: None)
    builder._tts_healthy = True

    missing = builder._validate_utterances()
    builder._generate_utterances(missing)

    languages = {c["text"]: c["language"] for c in calls}
    assert languages["The ship landed."] == "en"
    assert languages["Hola, amigo."] == "es"

    # The marker records the same language, so a later retag invalidates it.
    marker = json.loads(
        (project / ".umbra" / "chapter_audio" / "ch1" / "utt_0001.wav.txt").read_text()
    )
    assert marker["language"] == "es"