50 lines
2.5 KiB
Python
50 lines
2.5 KiB
Python
"""Execute VoiceStudio's real assembly generator against a cache over its cap."""
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import types
|
|
import unittest
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from api.routers import audiobook as router
|
|
from services.audiobook import parse_audiobook_script
|
|
from services.longform_render import prune_cache_dir
|
|
|
|
|
|
class PreparedCacheTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_cmp_assembly_keeps_warmed_audio_even_over_cache_limit(self):
|
|
await self.check_assembly(0)
|
|
|
|
async def test_failed_encoder_does_not_report_done(self):
|
|
await self.check_assembly(1)
|
|
|
|
async def check_assembly(self, returncode):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
wav = Path(root) / 'longform_cache/fixture.wav'
|
|
wav.parent.mkdir()
|
|
wav.write_bytes(b'prepared chapter larger than configured cap')
|
|
async def cached_chapter(*args, **kwargs):
|
|
self.assertTrue(wav.is_file(), 'assembly evicted an already rendered chapter')
|
|
return str(wav), 79213.0, True, None
|
|
job_store = types.SimpleNamespace(**{n: lambda *a, **k: None for n in
|
|
['create', 'mark_running', 'append_event', 'mark_done', 'mark_failed']})
|
|
with patch.dict(os.environ, {'CMP_AUDIOBOOK_GPUS': 'gpu0,gpu1,gpu2'}), \
|
|
patch('core.config.OUTPUTS_DIR', root), \
|
|
patch('core.job_store', job_store, create=True), \
|
|
patch('services.ffmpeg_utils.find_ffmpeg', return_value='ffmpeg'), \
|
|
patch('services.ffmpeg_utils.run_ffmpeg', new=AsyncMock(return_value=(returncode, b'', b'encode failed'))) as encode, \
|
|
patch('services.gpu_gateway.decide'), \
|
|
patch.object(router, '_resolve_default_language', return_value='ru'), \
|
|
patch.object(router, '_run_chapter', side_effect=cached_chapter), \
|
|
patch.object(router, 'prune_cache_dir', side_effect=lambda p: prune_cache_dir(p, max_bytes=1)):
|
|
events = [e async for e in router._render_longform_sse(
|
|
parse_audiobook_script('# Chapter\nText'), default_voice=None)]
|
|
self.assertTrue(wav.exists())
|
|
self.assertEqual(encode.call_args.kwargs['timeout'], 21600.0)
|
|
self.assertEqual(any('"type": "done"' in e for e in events), returncode == 0, events)
|
|
self.assertEqual(any('"type": "error"' in e for e in events), returncode != 0, events)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|