import contextlib import io import os import tempfile import unittest import wave from pathlib import Path from unittest import mock def write_wav(path, channels=1, sample_width=2, frame_rate=24000, frames=8): 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(b"\0" * channels * sample_width * frames) def write_empty_wav(path, channels=1, sample_width=2, frame_rate=24000): with wave.open(str(path), "wb") as wav_file: wav_file.setnchannels(channels) wav_file.setsampwidth(sample_width) wav_file.setframerate(frame_rate) 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(path): with wave.open(str(path), "rb") as wav_file: return { "channels": wav_file.getnchannels(), "sample_width": wav_file.getsampwidth(), "frame_rate": wav_file.getframerate(), "frames": 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 ConcatCommandDryRunTests(unittest.TestCase): def run_concat(self, args): 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) return exit_code, stdout.getvalue(), stderr.getvalue() def test_concat_dry_run_validates_manifest_without_creating_output_parent(self): with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) concat_dir = temp_path / "concat" concat_dir.mkdir() audio_path = concat_dir / "clip.wav" concat_file = concat_dir / "manifest.jsonl" output_path = temp_path / "new-parent" / "out.wav" write_wav(audio_path) concat_file.write_text( '\n{"audio": "clip.wav", "silence_after_ms": 125}\n\n', encoding="utf-8", ) def fail_if_runtime_imports_are_checked(): raise AssertionError("concat dry-run must not check runtime packages") import indextts.cli_v2 as cli_v2 original_import_check = cli_v2._import_required_packages cli_v2._import_required_packages = fail_if_runtime_imports_are_checked try: exit_code, stdout, stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(output_path), "--dry-run", ] ) finally: cli_v2._import_required_packages = original_import_check output_parent_exists = output_path.parent.exists() output_exists = output_path.exists() self.assertEqual(exit_code, 0) self.assertEqual(stdout, "Concat file OK: 1 segments\n") self.assertEqual(stderr, "") self.assertFalse(output_parent_exists) self.assertFalse(output_exists) def test_concat_dry_run_rejects_non_object_json_with_1_based_line_number(self): with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) concat_file = temp_path / "manifest.jsonl" concat_file.write_text('\n["not", "an", "object"]\n', encoding="utf-8") exit_code, stdout, stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(temp_path / "out.wav"), "--dry-run", ] ) self.assertEqual(exit_code, 1) self.assertEqual(stdout, "") self.assertIn("line 2", stderr) self.assertIn("JSON object", stderr) def test_concat_dry_run_rejects_unknown_fields_and_comment_lines(self): with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) audio_path = temp_path / "clip.wav" concat_file = temp_path / "manifest.jsonl" write_wav(audio_path) concat_file.write_text( '# {"audio": "clip.wav"}\n{"audio": "clip.wav", "text": "ignored"}\n', encoding="utf-8", ) first_exit_code, first_stdout, first_stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(temp_path / "out.wav"), "--dry-run", ] ) concat_file.write_text('{"audio": "clip.wav", "text": "ignored"}\n', encoding="utf-8") second_exit_code, second_stdout, second_stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(temp_path / "out.wav"), "--dry-run", ] ) self.assertEqual(first_exit_code, 1) self.assertEqual(first_stdout, "") self.assertIn("line 1", first_stderr) self.assertIn("not valid JSON", first_stderr) self.assertEqual(second_exit_code, 1) self.assertEqual(second_stdout, "") self.assertIn("line 1", second_stderr) self.assertIn("unknown fields", second_stderr) self.assertIn("text", second_stderr) def test_concat_dry_run_rejects_invalid_segment_fields_with_line_number(self): cases = [ ('{"silence_after_ms": 0}\n', "missing required field: audio"), ('{"audio": ""}\n', "must not be empty"), ('{"audio": 123}\n', "must be a string"), ('{"audio": "clip.wav", "silence_after_ms": -1}\n', "non-negative integer"), ('{"audio": "clip.wav", "silence_after_ms": 1.5}\n', "non-negative integer"), ('{"audio": "clip.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) write_wav(temp_path / "clip.wav") concat_file = temp_path / "manifest.jsonl" concat_file.write_text("\n" + manifest, encoding="utf-8") exit_code, stdout, stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(temp_path / "out.wav"), "--dry-run", ] ) self.assertEqual(exit_code, 1) self.assertEqual(stdout, "") self.assertIn("line 2", stderr) self.assertIn(expected_message, stderr) def test_concat_dry_run_resolves_command_paths_from_cwd_and_audio_from_concat_file_directory(self): with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) cwd_path = temp_path / "cwd" concat_dir = temp_path / "manifests" assets_dir = concat_dir / "assets" cwd_path.mkdir() concat_dir.mkdir() assets_dir.mkdir() write_wav(assets_dir / "CLIP.WAV") concat_file = concat_dir / "manifest.jsonl" output_path = cwd_path / "OUT.WAV" concat_file.write_text('{"audio": "assets/CLIP.WAV"}\n', encoding="utf-8") with working_directory(cwd_path): exit_code, stdout, stderr = self.run_concat( [ "concat", "--concat-file", str(Path("..") / "manifests" / "manifest.jsonl"), "--output", "OUT.WAV", "--dry-run", ] ) self.assertEqual(exit_code, 0) self.assertEqual(stdout, "Concat file OK: 1 segments\n") self.assertEqual(stderr, "") self.assertFalse(output_path.exists()) def test_concat_dry_run_rejects_non_wav_output_and_audio_extensions_case_insensitively(self): with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) audio_path = temp_path / "clip.mp3" concat_file = temp_path / "manifest.jsonl" audio_path.write_bytes(b"not checked after extension failure") concat_file.write_text('{"audio": "clip.mp3"}\n', encoding="utf-8") output_exit_code, output_stdout, output_stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(temp_path / "out.mp3"), "--dry-run", ] ) audio_exit_code, audio_stdout, audio_stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(temp_path / "out.wav"), "--dry-run", ] ) self.assertEqual(output_exit_code, 1) self.assertEqual(output_stdout, "") self.assertIn("--output must be a .wav file", output_stderr) self.assertEqual(audio_exit_code, 1) self.assertEqual(audio_stdout, "") self.assertIn("line 1", audio_stderr) self.assertIn("field 'audio' must be a .wav file", audio_stderr) def test_concat_dry_run_checks_wav_existence_readability_non_empty_and_format_match(self): cases = [ ("missing.wav", None, 2, "audio file does not exist"), ("broken.wav", b"not a wav", 1, "not a readable WAV"), ("empty.wav", "empty", 1, "audio file is empty"), ] for audio_name, audio_content, expected_exit_code, expected_message in cases: with self.subTest(expected_message=expected_message): with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) concat_file = temp_path / "manifest.jsonl" if audio_content == "empty": write_empty_wav(temp_path / audio_name) elif audio_content is not None: (temp_path / audio_name).write_bytes(audio_content) concat_file.write_text(f'{{"audio": "{audio_name}"}}\n', encoding="utf-8") exit_code, stdout, stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(temp_path / "out.wav"), "--dry-run", ] ) self.assertEqual(exit_code, expected_exit_code) self.assertEqual(stdout, "") self.assertIn("line 1", stderr) self.assertIn(expected_message, stderr) with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) concat_file = temp_path / "manifest.jsonl" write_wav(temp_path / "first.wav", frame_rate=24000) write_wav(temp_path / "second.wav", frame_rate=22050) concat_file.write_text( '{"audio": "first.wav"}\n{"audio": "second.wav"}\n', encoding="utf-8", ) exit_code, stdout, stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(temp_path / "out.wav"), "--dry-run", ] ) self.assertEqual(exit_code, 1) self.assertEqual(stdout, "") self.assertIn("line 2", stderr) self.assertIn("baseline line 1", stderr) self.assertIn("WAV format does not match", stderr) def test_concat_dry_run_rejects_empty_manifest(self): with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) concat_file = temp_path / "manifest.jsonl" concat_file.write_text("\n \n", encoding="utf-8") exit_code, stdout, stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(temp_path / "out.wav"), "--dry-run", ] ) self.assertEqual(exit_code, 1) self.assertEqual(stdout, "") self.assertIn("at least one segment", stderr) def test_concat_dry_run_rejects_output_path_conflicts(self): with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) audio_path = temp_path / "clip.wav" concat_file = temp_path / "manifest.jsonl" blocked_parent = temp_path / "blocked" write_wav(audio_path) blocked_parent.write_text("file blocks directory creation", encoding="utf-8") concat_file.write_text('{"audio": "clip.wav"}\n', encoding="utf-8") same_concat_exit_code, same_concat_stdout, same_concat_stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(concat_file), "--dry-run", ] ) same_audio_exit_code, same_audio_stdout, same_audio_stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(audio_path), "--dry-run", ] ) blocked_parent_exit_code, blocked_parent_stdout, blocked_parent_stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(blocked_parent / "out.wav"), "--dry-run", ] ) self.assertEqual(same_concat_exit_code, 1) self.assertEqual(same_concat_stdout, "") self.assertIn("--output must not be the same path as --concat-file", same_concat_stderr) self.assertEqual(same_audio_exit_code, 1) self.assertEqual(same_audio_stdout, "") self.assertIn("line 1", same_audio_stderr) self.assertIn("conflicts with --output", same_audio_stderr) self.assertEqual(blocked_parent_exit_code, 1) self.assertEqual(blocked_parent_stdout, "") self.assertIn("output parent path cannot be created", blocked_parent_stderr) def test_concat_dry_run_rejects_existing_output_unless_force_without_modifying_it(self): with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) audio_path = temp_path / "clip.wav" concat_file = temp_path / "manifest.jsonl" output_path = temp_path / "out.wav" write_wav(audio_path) output_path.write_bytes(b"existing output") concat_file.write_text('{"audio": "clip.wav"}\n', encoding="utf-8") reject_exit_code, reject_stdout, reject_stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(output_path), "--dry-run", ] ) force_exit_code, force_stdout, force_stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(output_path), "--dry-run", "--force", ] ) 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, "Concat file OK: 1 segments\n") self.assertEqual(force_stderr, "") self.assertEqual(output_bytes, b"existing output") def test_concat_generates_wav_with_manifest_order_and_silence(self): with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) first_path = temp_path / "first.wav" second_path = temp_path / "second.wav" concat_file = temp_path / "manifest.jsonl" output_path = temp_path / "out.wav" write_wav_frames(first_path, b"\x01\x02") write_wav_frames(second_path, b"\x03") concat_file.write_text( '{"audio": "first.wav", "silence_after_ms": 2}\n' '{"audio": "second.wav", "silence_after_ms": 1}\n', encoding="utf-8", ) exit_code, stdout, stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(output_path), ] ) output_wav = read_wav(output_path) self.assertEqual(exit_code, 0) self.assertEqual(stdout, f"Generated: {output_path}\n") self.assertEqual(stderr, "") self.assertEqual(output_wav["channels"], 1) self.assertEqual(output_wav["sample_width"], 1) self.assertEqual(output_wav["frame_rate"], 1000) self.assertEqual(output_wav["frames"], b"\x01\x02\x00\x00\x03\x00") def test_concat_execution_does_not_initialize_or_check_model_resources(self): with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) audio_path = temp_path / "clip.wav" concat_file = temp_path / "manifest.jsonl" output_path = temp_path / "out.wav" write_wav_frames(audio_path, b"\x04\x05") concat_file.write_text('{"audio": "clip.wav"}\n', encoding="utf-8") with mock.patch( "indextts.cli_v2._ensure_user_state", side_effect=AssertionError("concat must not initialize user state"), ), mock.patch( "indextts.cli_v2._missing_model_files", side_effect=AssertionError("concat must not check model resources"), ): exit_code, stdout, stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(output_path), ] ) self.assertEqual(exit_code, 0) self.assertEqual(stdout, f"Generated: {output_path}\n") self.assertEqual(stderr, "") def test_concat_force_overwrites_existing_output_during_real_execution(self): with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) audio_path = temp_path / "clip.wav" concat_file = temp_path / "manifest.jsonl" output_path = temp_path / "out.wav" write_wav_frames(audio_path, b"\x04\x05") write_wav_frames(output_path, b"\x09") concat_file.write_text('{"audio": "clip.wav"}\n', encoding="utf-8") reject_exit_code, reject_stdout, reject_stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(output_path), ] ) force_exit_code, force_stdout, force_stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(output_path), "--force", ] ) output_wav = read_wav(output_path) 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}\n") self.assertEqual(force_stderr, "") self.assertEqual(output_wav["frames"], b"\x04\x05") def test_concat_execution_failure_returns_code_4_and_removes_temporary_wav(self): with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) audio_path = temp_path / "clip.wav" concat_file = temp_path / "manifest.jsonl" output_path = temp_path / "out.wav" write_wav_frames(audio_path, b"\x06") concat_file.write_text('{"audio": "clip.wav"}\n', encoding="utf-8") import indextts.cli_v2 as cli_v2 original_replace = cli_v2.os.replace def fail_replace(_source, _target): raise OSError("replace failed") cli_v2.os.replace = fail_replace try: exit_code, stdout, stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(output_path), ] ) finally: cli_v2.os.replace = original_replace temporary_wavs = list(temp_path.glob(f".{output_path.name}.*.wav")) self.assertEqual(exit_code, 4) self.assertEqual(stdout, "") self.assertIn("ERROR: concat failed: replace failed", stderr) self.assertNotIn("WARNING: cleanup failed", stderr) self.assertFalse(output_path.exists()) self.assertEqual(temporary_wavs, []) def test_concat_cleanup_failure_is_appended_without_overriding_primary_failure(self): with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) audio_path = temp_path / "clip.wav" concat_file = temp_path / "manifest.jsonl" output_path = temp_path / "out.wav" write_wav_frames(audio_path, b"\x07") concat_file.write_text('{"audio": "clip.wav"}\n', encoding="utf-8") import indextts.cli_v2 as cli_v2 original_replace = cli_v2.os.replace original_cleanup = cli_v2._cleanup_concat_temp_file def fail_replace(_source, _target): raise OSError("replace failed") def fail_cleanup(_temp_path): return OSError("cannot remove temp") cli_v2.os.replace = fail_replace cli_v2._cleanup_concat_temp_file = fail_cleanup try: exit_code, stdout, stderr = self.run_concat( [ "concat", "--concat-file", str(concat_file), "--output", str(output_path), ] ) finally: cli_v2.os.replace = original_replace cli_v2._cleanup_concat_temp_file = original_cleanup for temp_wav in temp_path.glob(f".{output_path.name}.*.wav"): temp_wav.unlink() self.assertEqual(exit_code, 4) self.assertEqual(stdout, "") self.assertIn("ERROR: concat failed: replace failed", stderr) self.assertIn("WARNING: cleanup failed: cannot remove temp", stderr) self.assertLess( stderr.index("ERROR: concat failed: replace failed"), stderr.index("WARNING: cleanup failed: cannot remove temp"), ) self.assertFalse(output_path.exists()) if __name__ == "__main__": unittest.main()