| import contextlib |
| import io |
| import os |
| import sys |
| import tempfile |
| import unittest |
| import wave |
| from pathlib import Path |
| from unittest import mock |
|
|
|
|
| REQUIRED_MODEL_FILES = [ |
| "config.yaml", |
| "bpe.model", |
| "gpt.pth", |
| "s2mel.pth", |
| "wav2vec2bert_stats.pt", |
| "feat1.pt", |
| "feat2.pt", |
| ] |
| REQUIRED_MODEL_DIRS = [ |
| "qwen0.6bemo4-merge", |
| ] |
| AUX_MODEL_FILES = [ |
| "hf_cache/semantic_codec_model.safetensors", |
| "hf_cache/campplus_cn_common.bin", |
| "hf_cache/bigvgan/config.json", |
| "hf_cache/bigvgan/bigvgan_generator.pt", |
| ] |
| AUX_MODEL_DIRS = [ |
| "hf_cache/w2v-bert-2.0", |
| ] |
|
|
|
|
| def make_model_dir(base_dir): |
| model_dir = base_dir / "checkpoints" |
| model_dir.mkdir() |
| for filename in REQUIRED_MODEL_FILES: |
| (model_dir / filename).write_text("placeholder", encoding="utf-8") |
| for dirname in REQUIRED_MODEL_DIRS: |
| (model_dir / dirname).mkdir() |
| for filename in AUX_MODEL_FILES: |
| target = model_dir / filename |
| target.parent.mkdir(parents=True, exist_ok=True) |
| target.write_text("placeholder", encoding="utf-8") |
| for dirname in AUX_MODEL_DIRS: |
| target = model_dir / dirname |
| target.mkdir(parents=True, exist_ok=True) |
| (target / "config.json").write_text("placeholder", encoding="utf-8") |
| return model_dir |
|
|
|
|
| def assert_model_resource_help(test_case, stderr, model_dir): |
| test_case.assertIn(f"Model directory: {model_dir}", stderr) |
| test_case.assertIn("Missing resources:", stderr) |
| test_case.assertIn("huggingface-cli download IndexTeam/IndexTTS-2", stderr) |
| test_case.assertIn("modelscope download --model IndexTeam/IndexTTS-2", stderr) |
| test_case.assertIn(f"indextts2 config set model_dir {model_dir}", stderr) |
|
|
|
|
| def user_state_env(temp_path): |
| if sys.platform == "win32": |
| return { |
| "APPDATA": str(temp_path / "roaming"), |
| "LOCALAPPDATA": str(temp_path / "local"), |
| } |
| if sys.platform == "darwin": |
| return {"HOME": str(temp_path)} |
| return { |
| "XDG_CONFIG_HOME": str(temp_path / "config"), |
| "XDG_DATA_HOME": str(temp_path / "data"), |
| } |
|
|
|
|
| def write_wav_frames(path, frames, channels=1, sample_width=1, frame_rate=1000): |
| with wave.open(str(path), "wb") as wav_file: |
| wav_file.setnchannels(channels) |
| wav_file.setsampwidth(sample_width) |
| wav_file.setframerate(frame_rate) |
| wav_file.writeframes(frames) |
|
|
|
|
| def read_wav_frames(path): |
| with wave.open(str(path), "rb") as wav_file: |
| return wav_file.readframes(wav_file.getnframes()) |
|
|
|
|
| class working_directory: |
| def __init__(self, path): |
| self.path = path |
| self.previous = None |
|
|
| def __enter__(self): |
| self.previous = Path.cwd() |
| os.chdir(self.path) |
|
|
| def __exit__(self, _exc_type, _exc, _tb): |
| os.chdir(self.previous) |
|
|
|
|
| class BatchCommandDryRunTests(unittest.TestCase): |
| def setUp(self): |
| self.user_state = tempfile.TemporaryDirectory() |
| self.env_patch = mock.patch.dict(os.environ, user_state_env(Path(self.user_state.name)), clear=False) |
| self.env_patch.start() |
|
|
| def tearDown(self): |
| self.env_patch.stop() |
| self.user_state.cleanup() |
|
|
| def run_batch(self, args, tts_factory=None): |
| from indextts.cli_v2 import main |
|
|
| stdout = io.StringIO() |
| stderr = io.StringIO() |
| with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): |
| exit_code = main(args, tts_factory=tts_factory) |
| return exit_code, stdout.getvalue(), stderr.getvalue() |
|
|
| def test_batch_dry_run_validates_manifest_without_loading_model(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| batch_dir = temp_path / "batch" |
| batch_dir.mkdir() |
| voice_path = batch_dir / "voice.wav" |
| batch_file = batch_dir / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '\n{"text": "hello", "voice": "voice.wav", "output": "out.wav"}\n\n', |
| encoding="utf-8", |
| ) |
|
|
| def fail_if_called(**_kwargs): |
| raise AssertionError("tts factory must not be called during dry-run") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| ], |
| tts_factory=fail_if_called, |
| ) |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual(stdout, "Batch file OK: 1 tasks\n") |
| self.assertEqual(stderr, "") |
|
|
| def test_batch_dry_run_rejects_non_object_json_with_1_based_line_number(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| batch_file = temp_path / "batch.jsonl" |
| batch_file.write_text('\n["not", "an", "object"]\n', encoding="utf-8") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 2", stderr) |
| self.assertIn("JSON object", stderr) |
|
|
| def test_batch_dry_run_rejects_unknown_fields(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "output": "out.wav", "bogus": true}\n', |
| encoding="utf-8", |
| ) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 1", stderr) |
| self.assertIn("unknown fields", stderr) |
| self.assertIn("bogus", stderr) |
|
|
| def test_batch_dry_run_rejects_conflicting_text_sources(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| text_path = temp_path / "input.txt" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| text_path.write_text("hello from file", encoding="utf-8") |
| batch_file.write_text( |
| '{"text": "hello", "text_file": "input.txt", "voice": "voice.wav", "output": "out.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 1", stderr) |
| self.assertIn("exactly one text source", stderr) |
|
|
| def test_batch_dry_run_rejects_missing_output(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text('{"text": "hello", "voice": "voice.wav"}\n', encoding="utf-8") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 1", stderr) |
| self.assertIn("missing required field: output", stderr) |
|
|
| def test_batch_dry_run_rejects_duplicate_output_paths_with_line_number(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| "\n".join( |
| [ |
| '{"text": "hello", "voice": "voice.wav", "output": "out.wav"}', |
| '{"text": "world", "voice": "voice.wav", "output": "out.wav"}', |
| ] |
| ) |
| + "\n", |
| encoding="utf-8", |
| ) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 2", stderr) |
| self.assertIn("duplicate output", stderr) |
|
|
| def test_batch_dry_run_resolves_text_file_and_voice_relative_to_batch_file_directory(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| batch_dir = temp_path / "batch" |
| assets_dir = batch_dir / "assets" |
| batch_dir.mkdir() |
| assets_dir.mkdir() |
| voice_path = assets_dir / "voice.wav" |
| text_path = assets_dir / "input.txt" |
| batch_file = batch_dir / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| text_path.write_text("hello from file", encoding="utf-8") |
| batch_file.write_text( |
| '{"text_file": "assets/input.txt", "voice": "assets/voice.wav", "output": "out.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| def fail_if_called(**_kwargs): |
| raise AssertionError("tts factory must not be called during dry-run") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| ], |
| tts_factory=fail_if_called, |
| ) |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual(stdout, "Batch file OK: 1 tasks\n") |
| self.assertEqual(stderr, "") |
|
|
| def test_batch_dry_run_checks_model_files_without_importing_runtime_packages(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| batch_file = temp_path / "batch.jsonl" |
| voice_path = temp_path / "voice.wav" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "output": "out.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| with mock.patch("indextts.cli_v2._import_required_packages", side_effect=AssertionError("must not import")): |
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual(stdout, "Batch file OK: 1 tasks\n") |
| self.assertEqual(stderr, "") |
|
|
| def test_batch_dry_run_with_force_still_rejects_duplicate_output_paths(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| "\n".join( |
| [ |
| '{"text": "hello", "voice": "voice.wav", "output": "out.wav"}', |
| '{"text": "world", "voice": "voice.wav", "output": "out.wav"}', |
| ] |
| ) |
| + "\n", |
| encoding="utf-8", |
| ) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| "--force", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 2", stderr) |
| self.assertIn("duplicate output", stderr) |
|
|
| def test_batch_concat_dry_run_validates_manifest_without_loading_model_or_creating_output_parent(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| batch_dir = temp_path / "batch" |
| batch_dir.mkdir() |
| voice_path = batch_dir / "voice.wav" |
| batch_file = batch_dir / "batch.jsonl" |
| output_path = temp_path / "new-parent" / "final.wav" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "first", "voice": "voice.wav", "silence_after_ms": 125}\n', |
| encoding="utf-8", |
| ) |
|
|
| def fail_if_called(**_kwargs): |
| raise AssertionError("tts factory must not be called during concat dry-run") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(output_path), |
| "--dry-run", |
| ], |
| tts_factory=fail_if_called, |
| ) |
| output_parent_exists = output_path.parent.exists() |
| output_exists = output_path.exists() |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual(stdout, "Batch concat OK: 1 tasks\n") |
| self.assertEqual(stderr, "") |
| self.assertFalse(output_parent_exists) |
| self.assertFalse(output_exists) |
|
|
| def test_batch_concat_rejects_invalid_command_output_contracts(self): |
| cases = [ |
| (["--concat"], "--output is required with --concat"), |
| (["--concat", "--output", "final.mp3"], "--output must be a .wav file"), |
| (["--output", "final.wav"], "--output is only valid with --concat"), |
| (["--keep-temp"], "--keep-temp requires --concat"), |
| (["--concat", "--output", "final.wav", "--output-dir", "auto"], "--concat cannot be used with --output-dir"), |
| ( |
| ["--concat", "--output", "final.wav", "--output-prefix", "chapter"], |
| "--concat cannot be used with --output-prefix", |
| ), |
| ] |
| for extra_args, expected_message in cases: |
| with self.subTest(expected_message=expected_message): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text('{"text": "hello", "voice": "voice.wav"}\n', encoding="utf-8") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| *extra_args, |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn(expected_message, stderr) |
|
|
| def test_batch_concat_enforces_row_output_and_silence_after_ms_contracts(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "output": "row.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| concat_exit_code, concat_stdout, concat_stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(temp_path / "final.wav"), |
| "--dry-run", |
| ] |
| ) |
|
|
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "silence_after_ms": 125, "output": "row.wav"}\n', |
| encoding="utf-8", |
| ) |
| normal_exit_code, normal_stdout, normal_stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| ] |
| ) |
|
|
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "silence_after_ms": 125}\n', |
| encoding="utf-8", |
| ) |
| keep_temp_exit_code, keep_temp_stdout, keep_temp_stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(temp_path / "final.wav"), |
| "--keep-temp", |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(concat_exit_code, 1) |
| self.assertEqual(concat_stdout, "") |
| self.assertIn("line 1", concat_stderr) |
| self.assertIn("field 'output' is not allowed with --concat", concat_stderr) |
| self.assertEqual(normal_exit_code, 1) |
| self.assertEqual(normal_stdout, "") |
| self.assertIn("line 1", normal_stderr) |
| self.assertIn("silence_after_ms", normal_stderr) |
| self.assertIn("only valid with --concat", normal_stderr) |
| self.assertEqual(keep_temp_exit_code, 0) |
| self.assertEqual(keep_temp_stdout, "Batch concat OK: 1 tasks\n") |
| self.assertEqual(keep_temp_stderr, "") |
|
|
| def test_batch_concat_generates_final_wav_and_cleans_temp_dir_by_default(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| output_path = temp_path / "final.wav" |
| calls = [] |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "first", "voice": "voice.wav", "silence_after_ms": 2}\n' |
| '{"text": "second", "voice": "voice.wav", "silence_after_ms": 1}\n', |
| encoding="utf-8", |
| ) |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **kwargs): |
| calls.append(("init", kwargs)) |
|
|
| def infer(self, **kwargs): |
| calls.append(("infer", kwargs)) |
| frames = b"\x01\x02" if kwargs["text"] == "first" else b"\x03" |
| write_wav_frames(Path(kwargs["output_path"]), frames) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(output_path), |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
| output_frames = read_wav_frames(output_path) |
| temp_dirs = [path for path in temp_path.iterdir() if path.is_dir() and path.name.startswith(".final.wav.")] |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual(stdout, f"Generated: {output_path}\n") |
| self.assertEqual(stderr, "") |
| self.assertEqual([call[0] for call in calls], ["init", "infer", "infer"]) |
| self.assertEqual(output_frames, b"\x01\x02\x00\x00\x03\x00") |
| self.assertEqual(temp_dirs, []) |
|
|
| def test_batch_concat_keep_temp_preserves_temp_dir_after_success(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| output_path = temp_path / "final.wav" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text('{"text": "hello", "voice": "voice.wav"}\n', encoding="utf-8") |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **_kwargs): |
| pass |
|
|
| def infer(self, **kwargs): |
| write_wav_frames(Path(kwargs["output_path"]), b"\x04") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(output_path), |
| "--keep-temp", |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
| temp_dirs = [path for path in temp_path.iterdir() if path.is_dir() and path.name.startswith(".final.wav.")] |
| temp_segment_exists = (temp_dirs[0] / "0001.wav").exists() if temp_dirs else False |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual(stdout, f"Generated: {output_path}\nTemp dir: {temp_dirs[0]}\n") |
| self.assertEqual(stderr, "") |
| self.assertEqual(len(temp_dirs), 1) |
| self.assertTrue(temp_segment_exists) |
|
|
| def test_batch_concat_stops_on_inference_failure_and_cleans_temp_dir_by_default(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| output_path = temp_path / "final.wav" |
| calls = [] |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "first", "voice": "voice.wav"}\n' |
| '{"text": "second", "voice": "voice.wav"}\n' |
| '{"text": "third", "voice": "voice.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **kwargs): |
| calls.append(("init", kwargs)) |
|
|
| def infer(self, **kwargs): |
| calls.append(("infer", kwargs)) |
| if kwargs["text"] == "second": |
| raise RuntimeError("boom") |
| write_wav_frames(Path(kwargs["output_path"]), kwargs["text"].encode("utf-8")) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(output_path), |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
| temp_dirs = [path for path in temp_path.iterdir() if path.is_dir() and path.name.startswith(".final.wav.")] |
|
|
| self.assertEqual(exit_code, 4) |
| self.assertEqual(stdout, "") |
| self.assertIn("ERROR: batch file line 2 inference failed: boom", stderr) |
| self.assertEqual([call[0] for call in calls], ["init", "infer", "infer"]) |
| self.assertFalse(output_path.exists()) |
| self.assertEqual(temp_dirs, []) |
|
|
| def test_batch_concat_keep_temp_preserves_temp_dir_after_inference_failure(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| output_path = temp_path / "final.wav" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "first", "voice": "voice.wav"}\n' |
| '{"text": "second", "voice": "voice.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **_kwargs): |
| pass |
|
|
| def infer(self, **kwargs): |
| if kwargs["text"] == "second": |
| raise RuntimeError("boom") |
| write_wav_frames(Path(kwargs["output_path"]), b"\x05") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(output_path), |
| "--keep-temp", |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
| temp_dirs = [path for path in temp_path.iterdir() if path.is_dir() and path.name.startswith(".final.wav.")] |
| temp_segment_exists = (temp_dirs[0] / "0001.wav").exists() if temp_dirs else False |
|
|
| self.assertEqual(exit_code, 4) |
| self.assertEqual(stdout, "") |
| self.assertIn("ERROR: batch file line 2 inference failed: boom", stderr) |
| self.assertEqual(len(temp_dirs), 1) |
| self.assertIn(f"Temp dir: {temp_dirs[0]}", stderr) |
| self.assertTrue(temp_segment_exists) |
| self.assertFalse(output_path.exists()) |
|
|
| def test_batch_concat_rejects_mismatched_generated_segment_format_and_cleans_temp_dir(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| output_path = temp_path / "final.wav" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "first", "voice": "voice.wav"}\n' |
| '{"text": "second", "voice": "voice.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **_kwargs): |
| pass |
|
|
| def infer(self, **kwargs): |
| frame_rate = 1000 if kwargs["text"] == "first" else 2000 |
| write_wav_frames(Path(kwargs["output_path"]), b"\x06", frame_rate=frame_rate) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(output_path), |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
| temp_dirs = [path for path in temp_path.iterdir() if path.is_dir() and path.name.startswith(".final.wav.")] |
|
|
| self.assertEqual(exit_code, 4) |
| self.assertEqual(stdout, "") |
| self.assertIn("ERROR: batch file line 2 inference failed", stderr) |
| self.assertIn("generated WAV format does not match baseline line 1", stderr) |
| self.assertFalse(output_path.exists()) |
| self.assertEqual(temp_dirs, []) |
|
|
| def test_batch_concat_temp_cleanup_failure_does_not_override_inference_failure(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| output_path = temp_path / "final.wav" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text('{"text": "hello", "voice": "voice.wav"}\n', encoding="utf-8") |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **_kwargs): |
| pass |
|
|
| def infer(self, **_kwargs): |
| raise RuntimeError("boom") |
|
|
| import indextts.cli_v2 as cli_v2 |
|
|
| original_cleanup = cli_v2._cleanup_batch_concat_temp_dir |
|
|
| def fail_cleanup(_temp_dir): |
| return OSError("cannot remove temp dir") |
|
|
| cli_v2._cleanup_batch_concat_temp_dir = fail_cleanup |
| try: |
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(output_path), |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
| finally: |
| cli_v2._cleanup_batch_concat_temp_dir = original_cleanup |
|
|
| self.assertEqual(exit_code, 4) |
| self.assertEqual(stdout, "") |
| self.assertIn("ERROR: batch file line 1 inference failed: boom", stderr) |
| self.assertIn("WARNING: cleanup failed: cannot remove temp dir", stderr) |
| self.assertLess( |
| stderr.index("ERROR: batch file line 1 inference failed: boom"), |
| stderr.index("WARNING: cleanup failed: cannot remove temp dir"), |
| ) |
| self.assertFalse(output_path.exists()) |
|
|
| def test_batch_concat_temp_cleanup_failure_after_success_returns_inference_error(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| output_path = temp_path / "final.wav" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text('{"text": "hello", "voice": "voice.wav"}\n', encoding="utf-8") |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **_kwargs): |
| pass |
|
|
| def infer(self, **kwargs): |
| write_wav_frames(Path(kwargs["output_path"]), b"\x07") |
|
|
| import indextts.cli_v2 as cli_v2 |
|
|
| original_cleanup = cli_v2._cleanup_batch_concat_temp_dir |
|
|
| def fail_cleanup(_temp_dir): |
| return OSError("cannot remove temp dir") |
|
|
| cli_v2._cleanup_batch_concat_temp_dir = fail_cleanup |
| try: |
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(output_path), |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
| finally: |
| cli_v2._cleanup_batch_concat_temp_dir = original_cleanup |
| output_exists = output_path.exists() |
|
|
| self.assertEqual(exit_code, 4) |
| self.assertEqual(stdout, "") |
| self.assertIn("ERROR: cleanup failed: cannot remove temp dir", stderr) |
| self.assertTrue(output_exists) |
|
|
| def test_batch_concat_dry_run_rejects_final_output_path_conflicts_without_side_effects(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| output_path = temp_path / "final.wav" |
| voice_path.write_bytes(b"voice") |
| output_path.write_bytes(b"existing") |
| batch_file.write_text('{"text": "hello", "voice": "voice.wav"}\n', encoding="utf-8") |
|
|
| voice_exit_code, voice_stdout, voice_stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(voice_path), |
| "--dry-run", |
| "--force", |
| ] |
| ) |
| existing_exit_code, existing_stdout, existing_stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(output_path), |
| "--dry-run", |
| ] |
| ) |
| force_exit_code, force_stdout, force_stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(output_path), |
| "--dry-run", |
| "--force", |
| ] |
| ) |
| output_bytes = output_path.read_bytes() |
|
|
| self.assertEqual(voice_exit_code, 1) |
| self.assertEqual(voice_stdout, "") |
| self.assertIn("line 1", voice_stderr) |
| self.assertIn("conflicts with protected input path", voice_stderr) |
| self.assertIn(str(voice_path), voice_stderr) |
| self.assertEqual(existing_exit_code, 1) |
| self.assertEqual(existing_stdout, "") |
| self.assertIn("output file already exists", existing_stderr) |
| self.assertIn(str(output_path), existing_stderr) |
| self.assertEqual(force_exit_code, 0) |
| self.assertEqual(force_stdout, "Batch concat OK: 1 tasks\n") |
| self.assertEqual(force_stderr, "") |
| self.assertEqual(output_bytes, b"existing") |
|
|
| def test_batch_concat_dry_run_rejects_final_output_that_matches_batch_file(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.wav" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text('{"text": "hello", "voice": "voice.wav"}\n', encoding="utf-8") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(batch_file), |
| "--dry-run", |
| "--force", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 1", stderr) |
| self.assertIn("conflicts with protected input path", stderr) |
| self.assertIn(str(batch_file), stderr) |
|
|
| def test_batch_concat_dry_run_rejects_final_output_that_matches_empty_batch_file(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| batch_file = temp_path / "batch.wav" |
| batch_file.write_text("", encoding="utf-8") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(batch_file), |
| "--dry-run", |
| "--force", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("conflicts with protected input path", stderr) |
| self.assertIn(str(batch_file), stderr) |
|
|
| def test_batch_concat_dry_run_rejects_invalid_silence_after_ms_values(self): |
| cases = [ |
| ('{"text": "hello", "voice": "voice.wav", "silence_after_ms": -1}\n', "non-negative integer"), |
| ('{"text": "hello", "voice": "voice.wav", "silence_after_ms": 1.5}\n', "non-negative integer"), |
| ('{"text": "hello", "voice": "voice.wav", "silence_after_ms": true}\n', "non-negative integer"), |
| ] |
| for manifest, expected_message in cases: |
| with self.subTest(expected_message=expected_message): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text(manifest, encoding="utf-8") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--concat", |
| "--output", |
| str(temp_path / "final.wav"), |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 1", stderr) |
| self.assertIn("silence_after_ms", stderr) |
| self.assertIn(expected_message, stderr) |
|
|
|
|
| class BatchCommandExecutionTests(unittest.TestCase): |
| def setUp(self): |
| self.user_state = tempfile.TemporaryDirectory() |
| self.env_patch = mock.patch.dict(os.environ, user_state_env(Path(self.user_state.name)), clear=False) |
| self.env_patch.start() |
|
|
| def tearDown(self): |
| self.env_patch.stop() |
| self.user_state.cleanup() |
|
|
| def run_batch(self, args, tts_factory=None): |
| from indextts.cli_v2 import main |
|
|
| stdout = io.StringIO() |
| stderr = io.StringIO() |
| with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): |
| exit_code = main(args, tts_factory=tts_factory) |
| return exit_code, stdout.getvalue(), stderr.getvalue() |
|
|
| def test_batch_executes_tasks_in_order_with_one_model_initialization_and_summary(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| batch_dir = temp_path / "batch" |
| batch_dir.mkdir() |
| voice_path = batch_dir / "voice.wav" |
| first_output = batch_dir / "first.wav" |
| second_output = batch_dir / "second.wav" |
| batch_file = batch_dir / "batch.jsonl" |
| calls = [] |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| "\n".join( |
| [ |
| '{"text": "first", "voice": "voice.wav", "output": "first.wav"}', |
| '{"text": "second", "voice": "voice.wav", "output": "second.wav"}', |
| ] |
| ) |
| + "\n", |
| encoding="utf-8", |
| ) |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **kwargs): |
| calls.append(("init", kwargs)) |
|
|
| def infer(self, **kwargs): |
| calls.append(("infer", kwargs)) |
| Path(kwargs["output_path"]).write_bytes(kwargs["text"].encode("utf-8")) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
| first_output_bytes = first_output.read_bytes() |
| second_output_bytes = second_output.read_bytes() |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual( |
| stdout, |
| f"Generated: {first_output}\nGenerated: {second_output}\nBatch complete: 2 tasks generated\n", |
| ) |
| self.assertEqual(stderr, "") |
| self.assertEqual([call[0] for call in calls], ["init", "infer", "infer"]) |
| self.assertEqual(calls[1][1]["text"], "first") |
| self.assertEqual(calls[2][1]["text"], "second") |
| self.assertEqual(calls[1][1]["spk_audio_prompt"], str(voice_path)) |
| self.assertEqual(calls[2][1]["spk_audio_prompt"], str(voice_path)) |
| self.assertEqual(first_output_bytes, b"first") |
| self.assertEqual(second_output_bytes, b"second") |
|
|
| def test_batch_auto_output_dir_generates_numbered_outputs(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| batch_dir = temp_path / "batch" |
| output_dir = temp_path / "auto" |
| batch_dir.mkdir() |
| voice_path = batch_dir / "voice.wav" |
| first_output = output_dir / "0001.wav" |
| second_output = output_dir / "0002.wav" |
| batch_file = batch_dir / "batch.jsonl" |
| calls = [] |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| "\n".join( |
| [ |
| '{"text": "first", "voice": "voice.wav"}', |
| "", |
| '{"text": "second", "voice": "voice.wav"}', |
| ] |
| ) |
| + "\n", |
| encoding="utf-8", |
| ) |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **kwargs): |
| calls.append(("init", kwargs)) |
|
|
| def infer(self, **kwargs): |
| calls.append(("infer", kwargs)) |
| Path(kwargs["output_path"]).write_bytes(kwargs["text"].encode("utf-8")) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--output-dir", |
| str(output_dir), |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
| first_output_bytes = first_output.read_bytes() |
| second_output_bytes = second_output.read_bytes() |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual( |
| stdout, |
| f"Generated: {first_output}\nGenerated: {second_output}\nBatch complete: 2 tasks generated\n", |
| ) |
| self.assertEqual(stderr, "") |
| self.assertEqual([call[0] for call in calls], ["init", "infer", "infer"]) |
| self.assertEqual(calls[1][1]["output_path"], str(first_output)) |
| self.assertEqual(calls[2][1]["output_path"], str(second_output)) |
| self.assertEqual(first_output_bytes, b"first") |
| self.assertEqual(second_output_bytes, b"second") |
|
|
| def test_batch_auto_output_dir_rejects_generated_output_that_conflicts_with_inputs_even_with_force(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| output_dir = temp_path / "auto" |
| output_dir.mkdir() |
| voice_path = output_dir / "0001.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text('{"text": "hello", "voice": "auto/0001.wav"}\n', encoding="utf-8") |
|
|
| def fail_if_called(**_kwargs): |
| raise AssertionError("tts factory must not be called when output precheck fails") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--output-dir", |
| str(output_dir), |
| "--force", |
| ], |
| tts_factory=fail_if_called, |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 1", stderr) |
| self.assertIn("conflicts with protected input path", stderr) |
| self.assertIn(str(voice_path), stderr) |
|
|
| def test_batch_auto_output_dir_rejects_generated_output_that_conflicts_with_batch_file_even_with_force(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| output_dir = temp_path / "auto" |
| output_dir.mkdir() |
| voice_path = temp_path / "voice.wav" |
| batch_file = output_dir / "0001.wav" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text('{"text": "hello", "voice": "../voice.wav"}\n', encoding="utf-8") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--output-dir", |
| str(output_dir), |
| "--force", |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 1", stderr) |
| self.assertIn("conflicts with protected input path", stderr) |
| self.assertIn(str(batch_file), stderr) |
|
|
| def test_batch_auto_output_dir_uses_output_prefix(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| output_dir = temp_path / "auto" |
| voice_path = temp_path / "voice.wav" |
| output_path = output_dir / "chapter-0001.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text('{"text": "hello", "voice": "voice.wav"}\n', encoding="utf-8") |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **_kwargs): |
| pass |
|
|
| def infer(self, **kwargs): |
| Path(kwargs["output_path"]).write_bytes(b"audio") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--output-dir", |
| str(output_dir), |
| "--output-prefix", |
| "chapter", |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual(stdout, f"Generated: {output_path}\nBatch complete: 1 tasks generated\n") |
| self.assertEqual(stderr, "") |
|
|
| def test_batch_auto_output_dir_rejects_invalid_output_configuration(self): |
| cases = [ |
| (["--output-prefix", "chapter"], "--output-prefix requires --output-dir"), |
| (["--output-dir", "auto", "--output-prefix", "chapter.wav"], "file extension"), |
| (["--output-dir", "auto", "--output-prefix", "nested/chapter"], "path separators"), |
| (["--output-dir", "auto", "--output-prefix", "nested\\chapter"], "path separators"), |
| ] |
| for extra_args, expected_message in cases: |
| with self.subTest(expected_message=expected_message): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text('{"text": "hello", "voice": "voice.wav"}\n', encoding="utf-8") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| *extra_args, |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn(expected_message, stderr) |
|
|
| def test_batch_auto_output_dir_rejects_row_output(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "output": "row.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--output-dir", |
| str(temp_path / "auto"), |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 1", stderr) |
| self.assertIn("not allowed with --output-dir", stderr) |
|
|
| def test_batch_auto_output_dir_rejects_concat_output_configuration(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text('{"text": "hello", "voice": "voice.wav"}\n', encoding="utf-8") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--output-dir", |
| str(temp_path / "auto"), |
| "--concat", |
| "--output", |
| str(temp_path / "final.wav"), |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("--concat", stderr) |
| self.assertIn("--output-dir", stderr) |
|
|
| def test_batch_auto_output_dir_dry_run_does_not_create_output_dir(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| output_dir = temp_path / "auto" |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text('{"text": "hello", "voice": "voice.wav"}\n', encoding="utf-8") |
|
|
| def fail_if_called(**_kwargs): |
| raise AssertionError("tts factory must not be called during dry-run") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--output-dir", |
| str(output_dir), |
| "--dry-run", |
| ], |
| tts_factory=fail_if_called, |
| ) |
| output_dir_exists = output_dir.exists() |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual(stdout, "Batch file OK: 1 tasks\n") |
| self.assertEqual(stderr, "") |
| self.assertFalse(output_dir_exists) |
|
|
| def test_batch_auto_output_dir_respects_force_for_existing_external_outputs(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| output_dir = temp_path / "auto" |
| output_dir.mkdir() |
| voice_path = temp_path / "voice.wav" |
| output_path = output_dir / "0001.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| output_path.write_bytes(b"existing") |
| batch_file.write_text('{"text": "hello", "voice": "voice.wav"}\n', encoding="utf-8") |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **_kwargs): |
| pass |
|
|
| def infer(self, **kwargs): |
| Path(kwargs["output_path"]).write_bytes(b"new audio") |
|
|
| reject_exit_code, reject_stdout, reject_stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--output-dir", |
| str(output_dir), |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
| force_exit_code, force_stdout, force_stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--output-dir", |
| str(output_dir), |
| "--force", |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
| output_bytes = output_path.read_bytes() |
|
|
| self.assertEqual(reject_exit_code, 1) |
| self.assertEqual(reject_stdout, "") |
| self.assertIn("output file already exists", reject_stderr) |
| self.assertEqual(force_exit_code, 0) |
| self.assertEqual(force_stdout, f"Generated: {output_path}\nBatch complete: 1 tasks generated\n") |
| self.assertEqual(force_stderr, "") |
| self.assertEqual(output_bytes, b"new audio") |
|
|
| def test_batch_auto_output_dir_resolves_relative_to_current_working_directory(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| cwd_path = temp_path / "cwd" |
| batch_dir = temp_path / "batch" |
| cwd_path.mkdir() |
| batch_dir.mkdir() |
| voice_path = batch_dir / "voice.wav" |
| batch_file = batch_dir / "batch.jsonl" |
| expected_output = cwd_path / "auto" / "0001.wav" |
| batch_relative_to_cwd = Path("..") / "batch" / "batch.jsonl" |
| model_relative_to_cwd = Path("..") / "checkpoints" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text('{"text": "hello", "voice": "voice.wav"}\n', encoding="utf-8") |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **_kwargs): |
| pass |
|
|
| def infer(self, **kwargs): |
| Path(kwargs["output_path"]).write_bytes(b"audio") |
|
|
| with working_directory(cwd_path): |
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_relative_to_cwd), |
| "--model-dir", |
| str(model_relative_to_cwd), |
| "--output-dir", |
| "auto", |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
| output_exists = expected_output.exists() |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual(stdout, f"Generated: {expected_output}\nBatch complete: 1 tasks generated\n") |
| self.assertEqual(stderr, "") |
| self.assertTrue(output_exists) |
|
|
| def test_batch_auto_output_dir_rejects_output_parent_that_is_a_file_during_dry_run(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| blocked_output_dir = temp_path / "blocked" |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| blocked_output_dir.write_text("file blocks output directory", encoding="utf-8") |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text('{"text": "hello", "voice": "voice.wav"}\n', encoding="utf-8") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--output-dir", |
| str(blocked_output_dir), |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("output parent path cannot be created", stderr) |
| self.assertIn(str(blocked_output_dir), stderr) |
|
|
| def test_batch_maps_command_runtime_options_to_indextts2_once(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| batch_file = temp_path / "batch.jsonl" |
| voice_path = temp_path / "voice.wav" |
| output_path = temp_path / "out.wav" |
| calls = [] |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "output": "out.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **kwargs): |
| calls.append(("init", kwargs)) |
|
|
| def infer(self, **kwargs): |
| calls.append(("infer", kwargs)) |
| Path(kwargs["output_path"]).write_bytes(b"audio") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--device", |
| "cuda:0", |
| "--fp16", |
| "--deepspeed", |
| "--cuda-kernel", |
| "--verbose", |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual(stderr, "") |
| self.assertIn(f"Generated: {output_path}\n", stdout) |
| self.assertEqual( |
| calls[0][1], |
| { |
| "cfg_path": str(model_dir / "config.yaml"), |
| "model_dir": str(model_dir), |
| "use_fp16": True, |
| "device": "cuda:0", |
| "use_cuda_kernel": True, |
| "use_deepspeed": True, |
| }, |
| ) |
| self.assertTrue(calls[1][1]["verbose"]) |
|
|
| def test_batch_applies_command_defaults_and_row_emotion_overrides(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| batch_dir = temp_path / "batch" |
| batch_dir.mkdir() |
| default_voice = temp_path / "default_voice.wav" |
| row_voice = batch_dir / "row_voice.wav" |
| default_emotion = temp_path / "default_emotion.wav" |
| row_emotion = batch_dir / "row_emotion.wav" |
| first_output = batch_dir / "first.wav" |
| second_output = batch_dir / "second.wav" |
| third_output = batch_dir / "third.wav" |
| batch_file = batch_dir / "batch.jsonl" |
| calls = [] |
| default_voice.write_bytes(b"default voice") |
| row_voice.write_bytes(b"row voice") |
| default_emotion.write_bytes(b"default emotion") |
| row_emotion.write_bytes(b"row emotion") |
| batch_file.write_text( |
| "\n".join( |
| [ |
| '{"text": "first", "output": "first.wav"}', |
| '{"text": "second", "voice": "row_voice.wav", "emotion_audio": "row_emotion.wav", "emotion_weight": 0.25, "output": "second.wav"}', |
| '{"text": "third", "emotion_vector": [0, 0, 0.5, 0, 0, 0, 0, 0], "emotion_weight": "0.4", "output": "third.wav"}', |
| ] |
| ) |
| + "\n", |
| encoding="utf-8", |
| ) |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **kwargs): |
| calls.append(("init", kwargs)) |
|
|
| def infer(self, **kwargs): |
| calls.append(("infer", kwargs)) |
| Path(kwargs["output_path"]).write_bytes(b"audio") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--voice", |
| str(default_voice), |
| "--emotion-audio", |
| str(default_emotion), |
| "--emotion-weight", |
| "0.75", |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual(stderr, "") |
| self.assertEqual( |
| stdout, |
| f"Generated: {first_output}\nGenerated: {second_output}\nGenerated: {third_output}\nBatch complete: 3 tasks generated\n", |
| ) |
| self.assertEqual(calls[1][1]["spk_audio_prompt"], str(default_voice)) |
| self.assertEqual(calls[1][1]["emo_audio_prompt"], str(default_emotion)) |
| self.assertEqual(calls[1][1]["emo_alpha"], 0.75) |
| self.assertEqual(calls[2][1]["spk_audio_prompt"], str(row_voice)) |
| self.assertEqual(calls[2][1]["emo_audio_prompt"], str(row_emotion)) |
| self.assertEqual(calls[2][1]["emo_alpha"], 0.25) |
| self.assertEqual(calls[3][1]["spk_audio_prompt"], str(default_voice)) |
| self.assertEqual(calls[3][1]["emo_vector"], [0.0, 0.0, 0.5, 0.0, 0.0, 0.0, 0.0, 0.0]) |
| self.assertEqual(calls[3][1]["emo_alpha"], 0.4) |
| self.assertNotIn("emo_audio_prompt", calls[3][1]) |
|
|
| def test_batch_row_emotion_weight_inherits_command_emotion_source(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| output_path = temp_path / "out.wav" |
| batch_file = temp_path / "batch.jsonl" |
| calls = [] |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "hello", "emotion_weight": 0.3, "output": "out.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **kwargs): |
| calls.append(("init", kwargs)) |
|
|
| def infer(self, **kwargs): |
| calls.append(("infer", kwargs)) |
| Path(kwargs["output_path"]).write_bytes(b"audio") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--voice", |
| str(voice_path), |
| "--emotion-text", |
| "warm and calm", |
| "--emotion-weight", |
| "0.8", |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual(stdout, f"Generated: {output_path}\nBatch complete: 1 tasks generated\n") |
| self.assertEqual(stderr, "") |
| self.assertEqual(calls[1][1]["use_emo_text"], True) |
| self.assertEqual(calls[1][1]["emo_text"], "warm and calm") |
| self.assertEqual(calls[1][1]["emo_alpha"], 0.3) |
|
|
| def test_batch_inherits_command_emotion_vector(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| output_path = temp_path / "out.wav" |
| batch_file = temp_path / "batch.jsonl" |
| calls = [] |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text('{"text": "hello", "output": "out.wav"}\n', encoding="utf-8") |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **kwargs): |
| calls.append(("init", kwargs)) |
|
|
| def infer(self, **kwargs): |
| calls.append(("infer", kwargs)) |
| Path(kwargs["output_path"]).write_bytes(b"audio") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--voice", |
| str(voice_path), |
| "--emotion-vector", |
| "0,0,0.8,0,0,0,0,0", |
| "--emotion-weight", |
| "0.6", |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual(stdout, f"Generated: {output_path}\nBatch complete: 1 tasks generated\n") |
| self.assertEqual(stderr, "") |
| self.assertEqual(calls[1][1]["emo_vector"], [0.0, 0.0, 0.8, 0.0, 0.0, 0.0, 0.0, 0.0]) |
| self.assertEqual(calls[1][1]["emo_alpha"], 0.6) |
|
|
| def test_batch_accepts_row_emotion_vector_cli_style_string(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| output_path = temp_path / "out.wav" |
| batch_file = temp_path / "batch.jsonl" |
| calls = [] |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "emotion_vector": "0,0,0.8,0,0,0,0,0", "emotion_weight": 0.45, "output": "out.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **kwargs): |
| calls.append(("init", kwargs)) |
|
|
| def infer(self, **kwargs): |
| calls.append(("infer", kwargs)) |
| Path(kwargs["output_path"]).write_bytes(b"audio") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual(stdout, f"Generated: {output_path}\nBatch complete: 1 tasks generated\n") |
| self.assertEqual(stderr, "") |
| self.assertEqual(calls[1][1]["emo_vector"], [0.0, 0.0, 0.8, 0.0, 0.0, 0.0, 0.0, 0.0]) |
| self.assertEqual(calls[1][1]["emo_alpha"], 0.45) |
|
|
| def test_batch_rejects_row_emotion_weight_without_emotion_source(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "emotion_weight": 0.3, "output": "out.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 1", stderr) |
| self.assertIn("emotion_weight", stderr) |
| self.assertIn("requires an emotion source", stderr) |
|
|
| def test_batch_rejects_conflicting_row_emotion_sources(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| emotion_path = temp_path / "emotion.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| emotion_path.write_bytes(b"emotion") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "emotion_audio": "emotion.wav", "emotion_text": "calm", "output": "out.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 1", stderr) |
| self.assertIn("mutually exclusive", stderr) |
|
|
| def test_batch_reuses_synth_emotion_vector_validation_for_rows(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "emotion_vector": "0.5,0.5,0,0,0,0,0,0", "output": "out.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 1", stderr) |
| self.assertIn("emotion_vector", stderr) |
| self.assertIn("sum must be <= 0.8", stderr) |
|
|
| def test_batch_rejects_boolean_entries_in_json_emotion_vector(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| voice_path = temp_path / "voice.wav" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "emotion_vector": [true, 0, 0, 0, 0, 0, 0, 0], "output": "out.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 1", stderr) |
| self.assertIn("emotion_vector", stderr) |
| self.assertIn("entries must be numeric", stderr) |
|
|
| def test_batch_stops_on_first_inference_failure_and_keeps_prior_outputs(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| batch_file = temp_path / "batch.jsonl" |
| voice_path = temp_path / "voice.wav" |
| first_output = temp_path / "first.wav" |
| second_output = temp_path / "second.wav" |
| third_output = temp_path / "third.wav" |
| calls = [] |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| "\n".join( |
| [ |
| '{"text": "first", "voice": "voice.wav", "output": "first.wav"}', |
| '{"text": "second", "voice": "voice.wav", "output": "second.wav"}', |
| '{"text": "third", "voice": "voice.wav", "output": "third.wav"}', |
| ] |
| ) |
| + "\n", |
| encoding="utf-8", |
| ) |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **kwargs): |
| calls.append(("init", kwargs)) |
|
|
| def infer(self, **kwargs): |
| calls.append(("infer", kwargs)) |
| if kwargs["text"] == "second": |
| raise RuntimeError("boom") |
| Path(kwargs["output_path"]).write_bytes(kwargs["text"].encode("utf-8")) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
| first_output_bytes = first_output.read_bytes() |
| second_output_exists = second_output.exists() |
| third_output_exists = third_output.exists() |
|
|
| self.assertEqual(exit_code, 4) |
| self.assertEqual(stdout, f"Generated: {first_output}\n") |
| self.assertIn("ERROR: batch file line 2 inference failed: boom", stderr) |
| self.assertEqual([call[0] for call in calls], ["init", "infer", "infer"]) |
| self.assertEqual(first_output_bytes, b"first") |
| self.assertFalse(second_output_exists) |
| self.assertFalse(third_output_exists) |
| self.assertNotIn("Batch complete", stdout) |
|
|
| def test_batch_rejects_existing_external_output_without_force_before_model_initialization(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| batch_file = temp_path / "batch.jsonl" |
| voice_path = temp_path / "voice.wav" |
| output_path = temp_path / "out.wav" |
| voice_path.write_bytes(b"voice") |
| output_path.write_bytes(b"existing") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "output": "out.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| def fail_if_called(**_kwargs): |
| raise AssertionError("tts factory must not be called when output precheck fails") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| ], |
| tts_factory=fail_if_called, |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("ERROR: batch file line 1 output file already exists", stderr) |
| self.assertIn(str(output_path), stderr) |
|
|
| def test_batch_force_allows_existing_external_output(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| batch_file = temp_path / "batch.jsonl" |
| voice_path = temp_path / "voice.wav" |
| output_path = temp_path / "out.wav" |
| voice_path.write_bytes(b"voice") |
| output_path.write_bytes(b"existing") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "output": "out.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| class FakeIndexTTS2: |
| def __init__(self, **_kwargs): |
| pass |
|
|
| def infer(self, **kwargs): |
| Path(kwargs["output_path"]).write_bytes(b"new audio") |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--force", |
| ], |
| tts_factory=FakeIndexTTS2, |
| ) |
| output_bytes = output_path.read_bytes() |
|
|
| self.assertEqual(exit_code, 0) |
| self.assertEqual(stdout, f"Generated: {output_path}\nBatch complete: 1 tasks generated\n") |
| self.assertEqual(stderr, "") |
| self.assertEqual(output_bytes, b"new audio") |
|
|
| def test_batch_rejects_runtime_options_inside_batch_rows(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| batch_file = temp_path / "batch.jsonl" |
| voice_path = temp_path / "voice.wav" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "output": "out.wav", "device": "cpu"}\n', |
| encoding="utf-8", |
| ) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| "--dry-run", |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 1) |
| self.assertEqual(stdout, "") |
| self.assertIn("line 1", stderr) |
| self.assertIn("unknown fields", stderr) |
| self.assertIn("device", stderr) |
|
|
| def test_batch_returns_resource_error_when_model_directory_is_missing(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| batch_file = temp_path / "batch.jsonl" |
| voice_path = temp_path / "voice.wav" |
| missing_model_dir = temp_path / "missing-models" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "output": "out.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(missing_model_dir), |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 2) |
| self.assertEqual(stdout, "") |
| self.assertIn("ERROR: model directory does not exist", stderr) |
| self.assertIn(str(missing_model_dir), stderr) |
| assert_model_resource_help(self, stderr, missing_model_dir) |
|
|
| def test_batch_returns_resource_error_with_download_help_when_model_file_is_missing(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = temp_path / "models" |
| batch_file = temp_path / "batch.jsonl" |
| voice_path = temp_path / "voice.wav" |
| model_dir.mkdir() |
| (model_dir / "config.yaml").write_text("placeholder", encoding="utf-8") |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "output": "out.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 2) |
| self.assertEqual(stdout, "") |
| self.assertIn("ERROR: missing required model files", stderr) |
| self.assertIn("bpe.model", stderr) |
| assert_model_resource_help(self, stderr, model_dir) |
|
|
| def test_batch_returns_runtime_error_when_indextts2_import_fails(self): |
| with tempfile.TemporaryDirectory() as temp_dir: |
| temp_path = Path(temp_dir) |
| model_dir = make_model_dir(temp_path) |
| batch_file = temp_path / "batch.jsonl" |
| voice_path = temp_path / "voice.wav" |
| voice_path.write_bytes(b"voice") |
| batch_file.write_text( |
| '{"text": "hello", "voice": "voice.wav", "output": "out.wav"}\n', |
| encoding="utf-8", |
| ) |
|
|
| with mock.patch("indextts.cli_v2._load_indextts2", side_effect=ImportError("torch")): |
| exit_code, stdout, stderr = self.run_batch( |
| [ |
| "batch", |
| "--batch-file", |
| str(batch_file), |
| "--model-dir", |
| str(model_dir), |
| ] |
| ) |
|
|
| self.assertEqual(exit_code, 3) |
| self.assertEqual(stdout, "") |
| self.assertIn("ERROR: runtime unavailable: torch", stderr) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
