124 lines
5.0 KiB
Python
124 lines
5.0 KiB
Python
"""Audiobook ingest: raw-cache transcoding must terminate and clean up."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from collections import namedtuple
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, PropertyMock, patch
|
|
|
|
import numpy as np
|
|
|
|
SRC_SR = 44100
|
|
SRC_SECONDS = 25 # two full 10 s blocks + one partial block
|
|
TARGET_SR = 22050
|
|
|
|
_DiskUsage = namedtuple("_DiskUsage", "total used free")
|
|
|
|
|
|
class TranscodeToRawTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
try:
|
|
import soundfile as sf
|
|
except ImportError:
|
|
self.skipTest("soundfile not installed")
|
|
from javis.voices import audiobook_ingest
|
|
self.sf = sf
|
|
self.ingest = audiobook_ingest
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self._tmp.name)
|
|
self.source = self.root / "book.wav"
|
|
t = np.arange(SRC_SR * SRC_SECONDS) / SRC_SR
|
|
tone = 0.3 * np.sin(2 * np.pi * 220 * t)
|
|
sf.write(str(self.source), np.stack([tone, tone], axis=1), SRC_SR, subtype="PCM_16")
|
|
self.target = self.root / "cache" / "ingest_audio_cache.i16"
|
|
|
|
def tearDown(self) -> None:
|
|
self._tmp.cleanup()
|
|
|
|
def _bogus_frame_count(self):
|
|
"""Mimic an MP3 with a malformed ID3 tag: INT64_MAX frames, unseekable."""
|
|
frames = patch.object(self.sf.SoundFile, "frames", new_callable=PropertyMock,
|
|
return_value=sys.maxsize)
|
|
seekable = patch.object(self.sf.SoundFile, "seekable", return_value=False)
|
|
return frames, seekable
|
|
|
|
def _run_with_timeout(self, timeout: float = 60.0, **kw) -> BaseException | None:
|
|
errors: list[BaseException] = []
|
|
|
|
def run():
|
|
try:
|
|
self.ingest._transcode_to_raw(self.source, self.target, TARGET_SR, **kw)
|
|
except BaseException as e: # noqa: BLE001
|
|
errors.append(e)
|
|
|
|
th = threading.Thread(target=run, daemon=True)
|
|
th.start()
|
|
th.join(timeout)
|
|
self.assertFalse(th.is_alive(), "transcode did not stop at EOF (runaway)")
|
|
return errors[0] if errors else None
|
|
|
|
def test_unknown_frame_count_stops_at_eof(self) -> None:
|
|
frames, seekable = self._bogus_frame_count()
|
|
with frames, seekable:
|
|
self.assertGreaterEqual(self.sf.info(str(self.source)).frames, sys.maxsize // 2)
|
|
err = self._run_with_timeout()
|
|
self.assertIsNone(err)
|
|
self.assertEqual(self.target.stat().st_size, SRC_SECONDS * TARGET_SR * 2)
|
|
self.assertFalse(self.target.with_suffix(".part").exists())
|
|
|
|
def test_stop_request_aborts_and_removes_partial_file(self) -> None:
|
|
err = self._run_with_timeout(should_stop=lambda: True)
|
|
self.assertIsInstance(err, self.ingest.TranscodeCancelled)
|
|
self.assertFalse(self.target.exists())
|
|
self.assertFalse(self.target.with_suffix(".part").exists())
|
|
|
|
def test_disk_running_full_aborts_and_removes_partial_file(self) -> None:
|
|
frames, seekable = self._bogus_frame_count()
|
|
full = patch.object(self.ingest.shutil, "disk_usage",
|
|
return_value=_DiskUsage(10**12, 10**12, 0))
|
|
every_block = patch.object(self.ingest, "_DISK_CHECK_EVERY_BLOCKS", 1)
|
|
with frames, seekable, full, every_block:
|
|
err = self._run_with_timeout()
|
|
self.assertIsInstance(err, RuntimeError)
|
|
self.assertIn("free space", str(err))
|
|
self.assertFalse(self.target.exists())
|
|
self.assertFalse(self.target.with_suffix(".part").exists())
|
|
|
|
def test_insufficient_space_for_known_length_fails_fast(self) -> None:
|
|
with patch.object(self.ingest.shutil, "disk_usage",
|
|
return_value=_DiskUsage(10**12, 10**12, 0)):
|
|
err = self._run_with_timeout()
|
|
self.assertIsInstance(err, RuntimeError)
|
|
self.assertIn("Not enough disk space", str(err))
|
|
self.assertFalse(self.target.with_suffix(".part").exists())
|
|
|
|
def test_stop_button_during_transcode_ends_ingest_as_stopped(self) -> None:
|
|
events: list[dict] = []
|
|
|
|
class Bus:
|
|
async def publish(self, topic, payload):
|
|
events.append(payload)
|
|
|
|
mgr = MagicMock()
|
|
mgr.root = self.root
|
|
mgr.create.return_value = SimpleNamespace(dir=self.root / "voice")
|
|
stop = asyncio.Event()
|
|
stop.set()
|
|
with patch("javis.voices.manager.VoiceManager", return_value=mgr):
|
|
stats = asyncio.run(self.ingest.ingest_audiobook(
|
|
None, self.source, "voice", bus=Bus(), stop_event=stop, resume=False))
|
|
self.assertEqual(stats.total_segments, 0)
|
|
self.assertEqual(events[-1]["phase"], "stopped")
|
|
self.assertEqual(events[-1]["during"], "transcoding")
|
|
self.assertFalse((self.root / "voice" / "ingest_audio_cache.i16").exists())
|
|
self.assertFalse((self.root / "voice" / "ingest_audio_cache.part").exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|