342 lines
15 KiB
Python
342 lines
15 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Tests for the release/publish mechanism (spec 22).
|
|
|
|
The mechanism exists because Deal Hunter shipped audio that differed from what was
|
|
approved. These tests pin that lesson so it cannot regress silently.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
import pytest
|
|
import soundfile as sf
|
|
|
|
from umbraaudio.release import (
|
|
DURATION_EPSILON_S,
|
|
DeliveryProfile,
|
|
build_publication,
|
|
file_sha256,
|
|
list_publications,
|
|
publication_path,
|
|
verify_publication,
|
|
write_publication,
|
|
)
|
|
|
|
|
|
def _wav(path: Path, seconds: float, rate: int = 24000) -> Path:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tone = np.sin(np.linspace(0, 440, int(rate * seconds), dtype="float32")) * 0.1
|
|
sf.write(str(path), tone, rate)
|
|
return path
|
|
|
|
|
|
def _project(
|
|
tmp_path: Path,
|
|
tracks: list[dict[str, Any]],
|
|
fmt: dict | None = None,
|
|
speakers: list[str] | None = None,
|
|
ledger: dict[str, str] | None = None,
|
|
dialogue: list[tuple[str, str]] | None = None,
|
|
) -> Path:
|
|
"""Build a minimal project tree with an exporter manifest."""
|
|
data_dir = tmp_path / ".umbra"
|
|
package_dir = data_dir / "export" / "ar_package"
|
|
package_dir.mkdir(parents=True, exist_ok=True)
|
|
for track in tracks:
|
|
source = track["source"]
|
|
chapter_id = int("".join(c for c in Path(source).stem if c.isdigit()))
|
|
_wav(data_dir / "chapter_audio" / source, track["chapter_seconds"])
|
|
mp3 = package_dir / track["file"]
|
|
size = int(track.get("size_bytes", 4096))
|
|
mp3.write_bytes(b"ID3" + b"x" * max(size - 3, 0))
|
|
track.setdefault("size_bytes", mp3.stat().st_size)
|
|
track.setdefault("chapter_id", chapter_id)
|
|
manifest = {
|
|
"base_name": "Book",
|
|
"built_at": "2026-09-09T11:21:11+00:00",
|
|
"format": fmt or {"codec": "mp3", "sample_rate": 44100, "channels": 1,
|
|
"bit_rate_kbps": 192, "cbr": True, "silence_s": 2.0},
|
|
"track_count": len(tracks),
|
|
"tracks": tracks,
|
|
}
|
|
(data_dir / "export" / "ar_package.json").write_text(json.dumps(manifest))
|
|
names = speakers if speakers is not None else ["Hallr", "Xylia", "Unreviewed"]
|
|
(data_dir / "speakers.json").write_text(
|
|
json.dumps({"speakers": [{"primary_name": n} for n in names]})
|
|
)
|
|
entries = ledger if ledger is not None else {"Hallr": "approved"}
|
|
(data_dir / "voice_status.json").write_text(json.dumps(entries))
|
|
|
|
# A real project DB so voice usage can be measured, not assumed.
|
|
if dialogue is not None:
|
|
chapter_id = tracks[0].get("chapter_id", 1)
|
|
import sqlite3
|
|
|
|
con = sqlite3.connect(str(data_dir / "project.db"))
|
|
con.executescript(
|
|
"CREATE TABLE chapters (id INTEGER PRIMARY KEY, source_text TEXT);"
|
|
"CREATE TABLE utterances (id INTEGER PRIMARY KEY, chapter_id INTEGER, "
|
|
"utterance_index INTEGER, text TEXT, type TEXT DEFAULT 'narration', "
|
|
"speaker_name TEXT);"
|
|
)
|
|
con.execute("INSERT INTO chapters (id, source_text) VALUES (?,?)", (chapter_id, "Body."))
|
|
for i, (speaker, text) in enumerate(dialogue):
|
|
con.execute(
|
|
"INSERT INTO utterances (id, chapter_id, utterance_index, text, type, "
|
|
"speaker_name) VALUES (?,?,?,?,'dialogue',?)",
|
|
(i + 1, chapter_id, i, text, speaker),
|
|
)
|
|
con.commit()
|
|
con.close()
|
|
return data_dir
|
|
|
|
|
|
class TestDeliveryProfile:
|
|
def test_reads_the_profile_actually_used(self) -> None:
|
|
profile = DeliveryProfile.from_ar_package(
|
|
{"format": {"sample_rate": 44100, "channels": 1, "silence_s": 2.0}}
|
|
)
|
|
assert profile.sample_rate == 44100
|
|
assert profile.lead_silence_s == 2.0
|
|
assert profile.trail_silence_s == 2.0
|
|
|
|
def test_silence_is_symmetric_by_default(self) -> None:
|
|
"""The Deal Hunter fix was lead AND trail. Asymmetric padding must still be visible."""
|
|
profile = DeliveryProfile.from_ar_package(
|
|
{"format": {"sample_rate": 44100, "channels": 1, "silence_s": 1.5}}
|
|
)
|
|
assert profile.lead_silence_s == profile.trail_silence_s == 1.5
|
|
|
|
def test_missing_format_is_not_fatal(self) -> None:
|
|
profile = DeliveryProfile.from_ar_package({})
|
|
assert profile.sample_rate == 0
|
|
|
|
|
|
class TestBuildPublication:
|
|
def test_pins_chapters_and_package(self, tmp_path: Path) -> None:
|
|
data_dir = _project(tmp_path, [
|
|
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
|
|
"duration_s": 6.0, "chapter_seconds": 2.0, "rms_db": -20.5, "peak_db": -3.0,
|
|
"loudness_ok": True, "peak_ok": True},
|
|
])
|
|
record = build_publication(data_dir, "Book", now="2026-10-03T00:00:00+00:00")
|
|
assert record["publishable"] is True
|
|
assert len(record["chapters"]) == 1
|
|
chapter = record["chapters"][0]
|
|
assert chapter["chapter_wav"]["sha256"] == file_sha256(
|
|
data_dir / "chapter_audio" / "ch1_full.wav"
|
|
)
|
|
assert chapter["mp3"]["sha256"] == file_sha256(
|
|
data_dir / "export" / "ar_package" / "01.mp3"
|
|
)
|
|
assert record["package"]["track_count"] == 1
|
|
|
|
def test_approval_is_split_explicit_implicit_and_unused(self, tmp_path: Path) -> None:
|
|
"""The three categories must never be conflated.
|
|
|
|
Releasing approves every voice a listener HEARS. It does not approve a profile
|
|
entry the release never uses.
|
|
"""
|
|
data_dir = _project(
|
|
tmp_path,
|
|
[{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
|
|
"duration_s": 2.0, "chapter_seconds": 2.0}],
|
|
speakers=["Hallr", "Xylia", "NeverSpeaks"],
|
|
ledger={"Hallr": "approved"},
|
|
dialogue=[("Hallr", "Hello."), ("Xylia", "Goodbye.")],
|
|
)
|
|
approvals = build_publication(data_dir, "Book")["approvals"]
|
|
assert approvals["voices_in_profile"] == 3
|
|
assert approvals["voices_explicitly_approved"] == 1
|
|
# Xylia speaks but was never explicitly reviewed -> implicit
|
|
assert approvals["voices_implicitly_approved"] == 1
|
|
assert approvals["voices_implicitly_approved_names"] == ["Xylia"]
|
|
# NeverSpeaks is in the profile but has no lines -> not covered at all
|
|
assert approvals["voices_in_profile_without_lines"] == 1
|
|
assert approvals["voices_without_lines_names"] == ["NeverSpeaks"]
|
|
assert approvals["implicit_approval_basis"] == "the act of releasing this audiobook"
|
|
|
|
def test_publication_record_is_the_chapter_and_audiobook_approval(
|
|
self, tmp_path: Path
|
|
) -> None:
|
|
data_dir = _project(tmp_path, [
|
|
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
|
|
"duration_s": 2.0, "chapter_seconds": 2.0},
|
|
])
|
|
approvals = build_publication(data_dir, "Book")["approvals"]
|
|
assert approvals["chapter_approved"] is True
|
|
assert approvals["audiobook_approved"] is True
|
|
assert approvals["approved_by"] == "this publication record"
|
|
|
|
|
|
class TestPostApprovalDelta:
|
|
"""The Deal Hunter defect, made mechanical."""
|
|
|
|
def test_padding_after_approval_is_flagged(self, tmp_path: Path) -> None:
|
|
data_dir = _project(tmp_path, [
|
|
{"seq": 6, "title": "Salvage", "file": "06.mp3", "source": "ch6_full.wav",
|
|
# shipped is 4s longer: 2s lead + 2s trail
|
|
"duration_s": 6.0, "chapter_seconds": 2.0},
|
|
])
|
|
chapter = build_publication(data_dir, "Book")["chapters"][0]
|
|
assert any("4.0s longer" in w for w in chapter["warnings"])
|
|
|
|
def test_resample_after_approval_is_flagged(self, tmp_path: Path) -> None:
|
|
data_dir = _project(tmp_path, [
|
|
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
|
|
"duration_s": 2.0, "chapter_seconds": 2.0},
|
|
]) # chapter wav is 24000 Hz, profile says 44100
|
|
chapter = build_publication(data_dir, "Book")["chapters"][0]
|
|
assert any("24000 Hz" in w and "resampled" in w for w in chapter["warnings"])
|
|
|
|
def test_no_delta_means_no_warning(self, tmp_path: Path) -> None:
|
|
data_dir = _project(tmp_path, [
|
|
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
|
|
"duration_s": 2.0, "chapter_seconds": 2.0},
|
|
], fmt={"sample_rate": 24000, "channels": 1, "silence_s": 0.0})
|
|
chapter = build_publication(data_dir, "Book")["chapters"][0]
|
|
assert chapter["warnings"] == []
|
|
|
|
def test_epsilon_tolerates_rounding(self, tmp_path: Path) -> None:
|
|
data_dir = _project(tmp_path, [
|
|
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
|
|
"duration_s": 2.0 + DURATION_EPSILON_S / 2, "chapter_seconds": 2.0},
|
|
])
|
|
chapter = build_publication(data_dir, "Book")["chapters"][0]
|
|
assert not any("longer" in w for w in chapter["warnings"])
|
|
|
|
def test_failed_quality_gate_surfaces(self, tmp_path: Path) -> None:
|
|
data_dir = _project(tmp_path, [
|
|
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
|
|
"duration_s": 2.0, "chapter_seconds": 2.0, "loudness_ok": False},
|
|
])
|
|
chapter = build_publication(data_dir, "Book")["chapters"][0]
|
|
assert any("quality gate" in w for w in chapter["warnings"])
|
|
|
|
|
|
class TestBlockers:
|
|
def test_missing_artifact_blocks_publication(self, tmp_path: Path) -> None:
|
|
data_dir = _project(tmp_path, [
|
|
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
|
|
"duration_s": 2.0, "chapter_seconds": 2.0},
|
|
])
|
|
(data_dir / "export" / "ar_package" / "01.mp3").unlink()
|
|
record = build_publication(data_dir, "Book")
|
|
assert record["publishable"] is False
|
|
assert any(b["id"].startswith("missing-artifact") for b in record["blockers"])
|
|
|
|
def test_empty_package_blocks_publication(self, tmp_path: Path) -> None:
|
|
data_dir = tmp_path / ".umbra"
|
|
(data_dir / "export").mkdir(parents=True)
|
|
(data_dir / "export" / "ar_package.json").write_text(json.dumps({"tracks": []}))
|
|
record = build_publication(data_dir, "Book")
|
|
assert record["publishable"] is False
|
|
assert any(b["id"] == "no-package" for b in record["blockers"])
|
|
|
|
def test_no_package_file_at_all_blocks(self, tmp_path: Path) -> None:
|
|
data_dir = tmp_path / ".umbra"
|
|
data_dir.mkdir(parents=True)
|
|
record = build_publication(data_dir, "Book")
|
|
assert record["publishable"] is False
|
|
|
|
|
|
class TestImmutability:
|
|
def test_publication_cannot_be_overwritten(self, tmp_path: Path) -> None:
|
|
data_dir = _project(tmp_path, [
|
|
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
|
|
"duration_s": 2.0, "chapter_seconds": 2.0},
|
|
])
|
|
record = build_publication(data_dir, "Book", now="2026-10-03T00:00:00+00:00")
|
|
write_publication(data_dir, record)
|
|
with pytest.raises(FileExistsError, match="immutable"):
|
|
write_publication(data_dir, record)
|
|
|
|
def test_successor_records_what_it_supersedes(self, tmp_path: Path) -> None:
|
|
data_dir = _project(tmp_path, [
|
|
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
|
|
"duration_s": 2.0, "chapter_seconds": 2.0},
|
|
])
|
|
first = build_publication(data_dir, "Book", edition_number=1,
|
|
now="2026-10-03T00:00:00+00:00")
|
|
write_publication(data_dir, first)
|
|
second = build_publication(data_dir, "Book", edition_number=2,
|
|
supersedes=first["publication_id"],
|
|
now="2026-10-03T01:00:00+00:00")
|
|
assert second["edition_number"] == 2
|
|
assert second["supersedes"] == first["publication_id"]
|
|
write_publication(data_dir, second)
|
|
ids = [r["publication_id"] for r in list_publications(data_dir)]
|
|
assert len(ids) == 2
|
|
# the first record is untouched and still on disk
|
|
stored = json.loads(publication_path(data_dir, first["publication_id"]).read_text())
|
|
assert stored["edition_number"] == 1
|
|
|
|
def test_write_is_atomic_no_tmp_left_behind(self, tmp_path: Path) -> None:
|
|
data_dir = _project(tmp_path, [
|
|
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
|
|
"duration_s": 2.0, "chapter_seconds": 2.0},
|
|
])
|
|
record = build_publication(data_dir, "Book")
|
|
path = write_publication(data_dir, record)
|
|
assert path.exists()
|
|
assert list(path.parent.glob("*.tmp")) == []
|
|
|
|
|
|
class TestVerification:
|
|
def _record(self, tmp_path: Path) -> tuple[Path, dict[str, Any]]:
|
|
data_dir = _project(tmp_path, [
|
|
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
|
|
"duration_s": 2.0, "chapter_seconds": 2.0},
|
|
])
|
|
return data_dir, build_publication(data_dir, "Book")
|
|
|
|
def test_unchanged_tree_verifies(self, tmp_path: Path) -> None:
|
|
data_dir, record = self._record(tmp_path)
|
|
result = verify_publication(data_dir, record)
|
|
assert result["ok"] is True
|
|
assert result["drift"] == []
|
|
|
|
def test_chapter_changed_after_publication_is_drift(self, tmp_path: Path) -> None:
|
|
data_dir, record = self._record(tmp_path)
|
|
_wav(data_dir / "chapter_audio" / "ch1_full.wav", 3.0)
|
|
result = verify_publication(data_dir, record)
|
|
assert result["ok"] is False
|
|
assert any(d["id"] == "chapter-changed-ch1" for d in result["drift"])
|
|
|
|
def test_missing_chapter_is_drift(self, tmp_path: Path) -> None:
|
|
data_dir, record = self._record(tmp_path)
|
|
(data_dir / "chapter_audio" / "ch1_full.wav").unlink()
|
|
result = verify_publication(data_dir, record)
|
|
assert result["ok"] is False
|
|
assert any("missing-chapter-wav" in d["id"] for d in result["drift"])
|
|
|
|
def test_tampered_mp3_is_drift(self, tmp_path: Path) -> None:
|
|
data_dir, record = self._record(tmp_path)
|
|
(data_dir / "export" / "ar_package" / "01.mp3").write_bytes(b"tampered!!")
|
|
result = verify_publication(data_dir, record)
|
|
assert result["ok"] is False
|
|
assert any(d["id"] == "mp3-changed-ch1" for d in result["drift"])
|
|
|
|
def test_silence_is_surfaced_as_a_standing_caveat(self, tmp_path: Path) -> None:
|
|
"""Even a clean tree carries the Deal Hunter caveat into every verification."""
|
|
data_dir, record = self._record(tmp_path)
|
|
result = verify_publication(data_dir, record)
|
|
assert any("Approving a chapter does not approve" in w for w in result["warnings"])
|
|
|
|
|
|
def test_file_sha256_matches_hashlib(tmp_path: Path) -> None:
|
|
import hashlib
|
|
|
|
target = tmp_path / "big.bin"
|
|
target.write_bytes(b"x" * (3 * 1024 * 1024)) # spans several chunks
|
|
assert file_sha256(target) == hashlib.sha256(b"x" * (3 * 1024 * 1024)).hexdigest()
|
|
|
|
|
|
def test_file_sha256_on_missing_file() -> None:
|
|
assert file_sha256(Path("/nonexistent/nope.bin")) is None
|