UmbraAudio/tests/test_controller_bundle.py

"""Tests for the controller bundle builder.

The builder's job is to emit something the *controller's* validator accepts, so
most of these tests build a small synthetic project and run the real validator
against the result. Running the actual validator -- copied into the bundle by the
builder -- is the point: a test that re-implements the checks would only prove the
test agrees with itself.

The synthetic project is built rather than the real one, so the suite does not
depend on 4,000 utterances and 4,500 audio files being present and current. The
real project is exercised by ``scripts/audit_controller_bundle.py``.
"""

from __future__ import annotations

import importlib.util
import json
import sqlite3
import subprocess
import sys
from pathlib import Path
from typing import Any

import pytest

REPO = Path(__file__).resolve().parent.parent
BUILDER_PATH = REPO / "scripts" / "build_controller_bundle.py"
VALIDATOR_SOURCE = Path("/tmp/controller-validator.py")


def _load_builder() -> Any:
    spec = importlib.util.spec_from_file_location("build_controller_bundle", BUILDER_PATH)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


builder = _load_builder()


SCHEMA = """
CREATE TABLE project (
    id INTEGER PRIMARY KEY, name TEXT NOT NULL, source_path TEXT NOT NULL,
    source_format TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    updated_at TEXT NOT NULL DEFAULT (datetime('now')),
    status TEXT NOT NULL DEFAULT 'initialized'
);
CREATE TABLE chapters (
    id INTEGER PRIMARY KEY, project_id INTEGER NOT NULL REFERENCES project(id),
    chapter_index INTEGER NOT NULL, title TEXT, source_text TEXT NOT NULL,
    word_count INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL DEFAULT 'parsed',
    audio_path TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE utterances (
    id INTEGER PRIMARY KEY, chapter_id INTEGER NOT NULL REFERENCES chapters(id),
    character_id INTEGER REFERENCES characters(id),
    utterance_index INTEGER NOT NULL, text TEXT NOT NULL,
    type TEXT NOT NULL DEFAULT 'narration', context_before TEXT, context_after TEXT,
    emotion TEXT, pace TEXT, emphasis_words TEXT, pause_before_ms INTEGER,
    pause_after_ms INTEGER, instruct TEXT, speaker_name TEXT,
    status TEXT NOT NULL DEFAULT 'pending'
);
CREATE TABLE characters (id INTEGER PRIMARY KEY, project_id INTEGER NOT NULL,
    name TEXT NOT NULL, aliases TEXT NOT NULL DEFAULT '[]', gender TEXT,
    is_narrator INTEGER NOT NULL DEFAULT 0);
"""


def make_project(root: Path, *, narrator_spellings: tuple[str, ...] = ("narrator",),
                 null_narrator_rows: int = 2, chapters: int = 3) -> Path:
    """A project shaped like the real one, including its awkward edges."""

    umbra = root / ".umbra"
    (umbra / "chapter_audio").mkdir(parents=True)
    (umbra / "voices").mkdir(parents=True)
    db = umbra / "project.db"
    conn = sqlite3.connect(db)
    conn.executescript(SCHEMA)
    conn.execute(
        "INSERT INTO project (id, name, source_path, source_format, status) "
        "VALUES (1, ?, ?, 'docx', 'analyzed')", (root.name, str(root)))
    for index in range(chapters):
        conn.execute(
            "INSERT INTO chapters (id, project_id, chapter_index, title, source_text, word_count) "
            "VALUES (?, 1, ?, ?, 'body text', 10)",
            (index + 1, index - 1, f"Chapter {index}"))
    utterance_id = 0
    for chapter in range(1, chapters + 1):
        for position in range(2):
            utterance_id += 1
            is_narrator_row = position == 0
            speaker = narrator_spellings[chapter % len(narrator_spellings)] \
                if is_narrator_row else "Kainda"
            conn.execute(
                "INSERT INTO utterances (id, chapter_id, utterance_index, text, "
                "type, speaker_name) VALUES (?, ?, ?, ?, ?, ?)",
                (utterance_id, chapter, position,
                 f"line {utterance_id}", "narration" if is_narrator_row else "dialogue",
                 speaker))
        # A row whose speaker is expressed by being empty, as the real data does.
        for _ in range(null_narrator_rows if chapter == 1 else 0):
            utterance_id += 1
            conn.execute(
                "INSERT INTO utterances (id, chapter_id, utterance_index, text, "
                "type, speaker_name) VALUES (?, ?, ?, ?, 'narration', NULL)",
                (utterance_id, chapter, 99, "implicit narration"))
    conn.commit()
    conn.close()

    (umbra / "speakers.json").write_text(json.dumps({
        "speakers": [
            {"primary_name": "narrator", "is_narrator": True, "aliases": [],
             "chapters_appeared": [1, 2, 3], "has_voice_ref": True},
            {"primary_name": "Kainda", "aliases": ["Princess Kainda"],
             "chapters_appeared": [1], "has_voice_ref": True},
        ],
        "book_title": "Test Book", "total_chapters": chapters, "total_words": 30,
    }))
    (umbra / "voice_status.json").write_text(json.dumps({"narrator": "approved"}))
    (umbra / "voice_engines.json").write_text(json.dumps(
        {"Kainda": {"engine": "qwen", "reason": "ok", "at": "2026-10-10T00:00:00+00:00"}}))
    (umbra / "voice_assignments.json").write_text(json.dumps({"narrator": "en-US-GuyNeural"}))
    return root


@pytest.fixture()
def built(tmp_path: Path) -> Path:
    make_project(tmp_path / "TestProject")
    out = tmp_path / "bundle"
    builder.build("TestProject", out, "2026-10-14T00:00:00Z",
                  VALIDATOR_SOURCE if VALIDATOR_SOURCE.exists() else None,
                  source_root=tmp_path)
    return out


def read_manifest(bundle: Path) -> list[dict[str, Any]]:
    return [json.loads(line) for line in
            (bundle / "manifest.jsonl").read_text().splitlines() if line.strip()]


# --------------------------------------------------------------------------
# the controller's own validator
# --------------------------------------------------------------------------


@pytest.mark.skipif(not VALIDATOR_SOURCE.exists(),
                    reason="controller validator not fetched to /tmp")
def test_controller_validator_accepts_the_bundle(built: Path) -> None:
    validator = built / "validate.py"
    if not validator.exists():
        pytest.skip("validator was not copied into the bundle")
    result = subprocess.run([sys.executable, str(validator), str(built)],
                            capture_output=True, text=True)
    assert result.returncode == 0, result.stdout + result.stderr


@pytest.mark.skipif(not VALIDATOR_SOURCE.exists(),
                    reason="controller validator not fetched to /tmp")
@pytest.mark.parametrize("mutation,expect", [
    ("dangling_id", "not in the manifest"),
    ("required_file_removed", "missing required file"),
    ("unitless_duration", "no unit"),
    ("credential", "credential"),
    # The index declares relative_to required, so that check fires before the
    # convention check does; either message would be correct.
    ("path_without_base", "relative_to"),
])
def test_controller_validator_rejects_a_broken_bundle(built: Path, mutation: str,
                                                       expect: str) -> None:
    """A validator nobody has seen fail is not evidence of anything.

    Each case breaks one thing the request asks for and asserts the validator
    notices, so a passing bundle means something.
    """

    validator = built / "validate.py"
    if not validator.exists():
        pytest.skip("validator was not copied into the bundle")
    lines = (built / "manifest.jsonl").read_text().splitlines()

    if mutation == "dangling_id":
        record = json.loads(lines[0])
        record["chapter_id"] = "ch-does-not-exist"
        lines[0] = json.dumps(record, sort_keys=True, separators=(",", ":"))
    elif mutation == "unitless_duration":
        record = json.loads(lines[0])
        record["duration"] = 1.5
        lines[0] = json.dumps(record, sort_keys=True, separators=(",", ":"))
    elif mutation == "credential":
        record = json.loads(lines[0])
        record["token"] = "sk-" + "A" * 24
        lines[0] = json.dumps(record, sort_keys=True, separators=(",", ":"))
    elif mutation == "path_without_base":
        for index, line in enumerate(lines):
            record = json.loads(line)
            if record.get("kind") == "render":
                record.pop("relative_to", None)
                record["file"] = "x.wav"
                lines[index] = json.dumps(record, sort_keys=True, separators=(",", ":"))
                break
        else:
            pytest.skip("no render record to break")

    if mutation == "required_file_removed":
        (built / "VOCABULARY.md").unlink()
    else:
        (built / "manifest.jsonl").write_text("\n".join(lines) + "\n")

    result = subprocess.run([sys.executable, str(validator), str(built)],
                            capture_output=True, text=True)
    assert result.returncode == 1
    assert expect in result.stdout


# --------------------------------------------------------------------------
# structure and self-consistency
# --------------------------------------------------------------------------


def test_bundle_has_every_required_piece(built: Path) -> None:
    for name in ("README.md", "bundle.json", "VOCABULARY.md", "manifest.jsonl"):
        assert (built / name).exists(), name
    assert (built / "schemas").is_dir()
    assert (built / "examples").is_dir()
    examples = [p for p in (built / "examples").iterdir() if p.is_dir()]
    assert examples
    for book in examples:
        assert (book / "manifest.json").exists()
        assert (book / "lineage.json").exists()


def test_content_hash_describes_the_file_it_ships_with(built: Path) -> None:
    import hashlib
    index = json.loads((built / "bundle.json").read_text())
    actual = hashlib.sha256((built / "manifest.jsonl").read_bytes()).hexdigest()
    assert index["content_hash"]["value"] == actual
    assert index["counts"]["records_total"] == len(read_manifest(built))


def test_build_is_deterministic(tmp_path: Path) -> None:
    """Regenerating an unchanged project must produce the same bytes.

    Without this, the content hash is decoration: it changes on every run for
    reasons unrelated to the data, and comparing two builds tells you nothing.
    """

    make_project(tmp_path / "TestProject")
    first = tmp_path / "one"
    second = tmp_path / "two"
    builder.build("TestProject", first, "2026-10-14T00:00:00Z", None, source_root=tmp_path)
    builder.build("TestProject", second, "2026-10-14T00:00:00Z", None, source_root=tmp_path)
    assert (first / "manifest.jsonl").read_bytes() == (second / "manifest.jsonl").read_bytes()


def test_every_reference_resolves_including_unvalidated_fields(built: Path) -> None:
    """The controller checks some reference fields; these are the rest.

    `speaker_id` and `render_id` are not on its list, but a dangling speaker
    reference is just as wrong, so it is checked here rather than assumed.
    """

    records = read_manifest(built)
    known = {r["id"] for r in records}
    for record in records:
        for field in ("book_id", "chapter_id", "speaker_id", "render_id", "voice_id"):
            value = record.get(field)
            if isinstance(value, str):
                assert value in known, f"{record['id']}.{field} -> {value}"
        for field in ("chapters", "renders", "inputs", "published_artifact_ids"):
            for value in record.get(field, []) or []:
                assert value in known, f"{record['id']}.{field} contains {value}"


def test_no_field_states_a_duration_without_a_unit(built: Path) -> None:
    forbidden = {"duration", "length", "elapsed", "runtime", "time"}
    for record in read_manifest(built):
        for field in record:
            assert field.lower() not in forbidden, f"{record['id']} has {field}"


def test_every_path_declares_what_it_is_relative_to(built: Path) -> None:
    for record in read_manifest(built):
        if "path" in record:
            assert record.get("relative_to") == "book_directory", record["id"]


def test_no_absolute_host_paths_leak_into_the_bundle(built: Path) -> None:
    """The bundle is committable and must not describe the machine layout."""

    blob = "\n".join((built / "manifest.jsonl").read_text().splitlines())
    for leak in ("/Volumes/", "/Users/", str(REPO)):
        assert leak not in blob, f"absolute path leaked: {leak}"


# --------------------------------------------------------------------------
# the findings this build exists to surface
# --------------------------------------------------------------------------


def test_narrator_written_three_ways_is_resolved_and_marked(built: Path) -> None:
    """An empty speaker name means the narrator; it must be labelled, not dropped."""

    records = read_manifest(built)
    utterances = [r for r in records if r["kind"] == "utterance"]
    implicit = [u for u in utterances if u["speaker_resolution"] == "implicit_narrator"]
    assert implicit, "the fixture is supposed to contain null-speaker rows"
    for utterance in implicit:
        assert utterance["speaker_name_raw"] in (None, "")
        assert utterance["speaker_id"] is not None, "an implicit narrator still has to resolve"
    for utterance in utterances:
        if utterance["speaker_resolution"] == "explicit":
            assert utterance["speaker_name_raw"] not in (None, "")


def test_case_insensitive_spellings_are_not_silently_merged(tmp_path: Path) -> None:
    make_project(tmp_path / "TestProject", narrator_spellings=("Narrator", "narrator"))
    out = tmp_path / "bundle"
    builder.build("TestProject", out, "2026-10-14T00:00:00Z", None, source_root=tmp_path)
    speakers = [r for r in read_manifest(out) if r["kind"] == "speaker"
                and r["casefold_collision_group"] == "narrator"]
    assert len(speakers) == 2, "the two spellings must stay separate records"
    for speaker in speakers:
        assert speaker["utterance_count"] < speaker["utterance_count_casefold_grouped"]


def test_chapter_numbering_hazard_is_declared(built: Path) -> None:
    book = next(r for r in read_manifest(built) if r["kind"] == "book")
    numbering = book["numbering"]["schemes"]
    assert numbering["database_row_id"]["is_authoritative"] is True
    assert numbering["chapter_index"]["is_authoritative"] is False
    assert numbering["distribution_track_number"]["is_authoritative"] is False


def test_chapters_record_which_number_their_filenames_use(built: Path) -> None:
    for chapter in (r for r in read_manifest(built) if r["kind"] == "chapter"):
        basis = chapter["numbering_basis"]
        assert basis["file_names_use"] == "database_row_id"
        assert basis["chapter_index_is_authoritative"] is False


def test_mismatched_extension_is_reported_not_hidden(built: Path) -> None:
    """A `.wav` that is not a WAV must be visible on the book, not just per file."""

    book = next(r for r in read_manifest(built) if r["kind"] == "book")
    for entry in book["extension_container_mismatches"]:
        assert entry["actual_container"] != entry["named_extension"]


def test_speakers_without_a_profile_are_named_on_the_book(built: Path) -> None:
    book = next(r for r in read_manifest(built) if r["kind"] == "book")
    assert isinstance(book["speaker_names_without_a_profile"], list)
    for name in book["speaker_names_without_a_profile"]:
        assert not any(r["kind"] == "speaker" and r["primary_name"] == name
                       and r["has_profile"] for r in read_manifest(built))


def test_declared_and_measured_chapter_counts_are_both_reported(built: Path) -> None:
    book = next(r for r in read_manifest(built) if r["kind"] == "book")
    assert book["chapter_count"] == len(book["chapters"])
    assert book["declared_total_chapters"] is not None


def test_language_designations_are_empty_and_explained(built: Path) -> None:
    """Empty is a real answer here; an invented list would be a guess in a schema."""

    book = next(r for r in read_manifest(built) if r["kind"] == "book")
    assert book["language_designations"] == []
    assert "do not record" in book["language_designation_note"] or \
           "records no per-utterance language" in book["language_designation_note"]


def test_schemas_describe_every_kind_that_appears(built: Path) -> None:
    kinds = {r["kind"] for r in read_manifest(built)}
    for kind in kinds:
        assert (built / "schemas" / f"{kind}.schema.json").exists(), kind
    assert (built / "schemas" / "manifest.schema.json").exists()
    assert (built / "schemas" / "lineage.schema.json").exists()


def test_required_fields_declared_in_the_index_are_actually_present(built: Path) -> None:
    index = json.loads((built / "bundle.json").read_text())
    records = read_manifest(built)
    for record in records:
        for field in index["required_fields"].get(record["kind"], []):
            assert field in record, f"{record['kind']} {record['id']} missing {field}"


def test_utterance_text_hash_matches_its_own_text(built: Path) -> None:
    import hashlib
    for record in read_manifest(built):
        if record["kind"] == "utterance":
            assert hashlib.sha256(record["text"].encode()).hexdigest() == record["text_sha256"]
            assert len(record["text"]) == record["text_char_count"]


def test_chapter_utterance_counts_are_true(built: Path) -> None:
    records = read_manifest(built)
    per_chapter: dict[str, int] = {}
    for record in records:
        if record["kind"] == "utterance":
            per_chapter[record["chapter_id"]] = per_chapter.get(record["chapter_id"], 0) + 1
    for chapter in (r for r in records if r["kind"] == "chapter"):
        assert chapter["utterance_count"] == per_chapter.get(chapter["id"], 0)


def test_speaker_utterance_counts_are_true_and_separate_from_the_inferred_one(built: Path) -> None:
    records = read_manifest(built)
    per_speaker: dict[str, int] = {}
    for record in records:
        if record["kind"] == "utterance" and record["speaker_resolution"] == "explicit":
            per_speaker[record["speaker_name_raw"]] = \
                per_speaker.get(record["speaker_name_raw"], 0) + 1
    for speaker in (r for r in records if r["kind"] == "speaker"):
        assert speaker["utterance_count"] == per_speaker.get(speaker["primary_name"], 0)
        assert speaker["utterance_count_including_implicit_narrator"] >= speaker["utterance_count"]


def test_example_is_a_readable_subset_not_a_copy(built: Path) -> None:
    """The example exists to teach the shape; a second copy of the data cannot."""

    example = next(p for p in (built / "examples").iterdir() if p.is_dir())
    example_records = json.loads((example / "manifest.json").read_text())
    assert 0 < len(example_records) < len(read_manifest(built))
    lineage = json.loads((example / "lineage.json").read_text())
    assert lineage["edges"], "lineage with no edges teaches nothing"
    assert all(edge.get("transform") for edge in lineage["edges"])
    assert all(edge.get("derivation") in {"direct", "derived", "assembled", "transcoded"}
               for edge in lineage["edges"])


def test_generator_never_writes_to_the_project(tmp_path: Path) -> None:
    """The audio system keeps its data and keeps running; the build only reads."""

    project = make_project(tmp_path / "TestProject")
    before = {p: p.stat().st_mtime_ns for p in sorted(project.rglob("*")) if p.is_file()}
    builder.build("TestProject", tmp_path / "bundle", "2026-10-14T00:00:00Z", None,
                  source_root=tmp_path)
    after = {p: p.stat().st_mtime_ns for p in sorted(project.rglob("*")) if p.is_file()}
    assert before == after


def test_generated_at_is_the_only_difference_between_two_timestamps(tmp_path: Path) -> None:
    make_project(tmp_path / "TestProject")
    first = tmp_path / "a"
    second = tmp_path / "b"
    builder.build("TestProject", first, "2026-10-14T00:00:00Z", None, source_root=tmp_path)
    builder.build("TestProject", second, "2026-10-15T00:00:00Z", None, source_root=tmp_path)

    def strip(path: Path) -> list[str]:
        # Everything that is a build-time stamp, not a fact about the data.
        volatile = {"generated_at", "file_modified_at"}
        return [
            json.dumps({k: v for k, v in json.loads(line).items() if k not in volatile},
                       sort_keys=True)
            for line in (path / "manifest.jsonl").read_text().splitlines()
        ]

    assert strip(first) == strip(second)


def test_production_time_is_not_claimed_as_build_time(tmp_path: Path) -> None:
    """`produced_at: <build time>` would be a claim about the audio, not the export.

    A render record must carry when the *file* was written, or nothing at all.
    """

    make_project(tmp_path / "TestProject")
    out = tmp_path / "bundle"
    builder.build("TestProject", out, "2026-10-14T00:00:00Z", None, source_root=tmp_path)
    for record in read_manifest(out):
        assert "produced_at" not in record, record["id"]
    # And the build timestamp never leaks into a file-time field.
    for record in read_manifest(out):
        assert record.get("file_modified_at") != "2026-10-14T00:00:00Z", record["id"]