mirror of
https://github.com/index-tts/index-tts.git
synced 2026-11-03 12:53:39 +08:00
nanaotoandnanaoto
bfaf561d7b
* test: resolve temp paths in the CLI tests so they pass off Linux
These assertions compare a path the CLI printed against one built from
`tempfile.TemporaryDirectory()`. On Linux those agree; on macOS they do not,
because `/var` is a symlink to `/private/var` and the CLI resolves it:
AssertionError: 'model_dir = "/var/folders/.../models"'
not found in 'model_dir = "/private/var/folders/.../models"'
18 of 153 non-GPU tests failed that way. Windows has its own version of the same
problem — `RUNNER~1` 8.3 short names and drive-letter case — so this blocks CI on
any non-Linux runner.
Resolve at the point the temp directory is wrapped, which is what
`test_cli_v2_download.py` already does at its seven sites and why that file was
the only one passing. 129 call sites across the other four files; no behaviour
changes, the comparison just uses the same canonical form on both sides.
153 passed, 0 failed afterwards.
* ci: run the lightweight checks on Windows too
Everything platform-specific in this repo was unverified. `pyproject.toml`
carries three `sys_platform` markers — triton-windows gated to win32, and the
wetext / WeTextProcessing split on linux — and `cli_tests` branches three ways on
`sys.platform` for config and model-cache locations. CI only ever exercised the
Linux side of any of it. The triton-windows marker bug (#742, #749) was exactly
this shape: a Windows-only resolution failure that no amount of green Linux CI
could catch.
No GPU is needed. `-m "not gpu"` deselects the 22 GPU tests, and the one
remaining case that passes `--device cuda` runs against a hand-built
`SimpleNamespace` whose `is_available` is `lambda: True`
(`cli_tests/test_cli_v2_check.py:93-98`), so nothing looks for a real device.
`fail-fast: false` so a Windows failure still lets the Linux result through, and
the uv cache is enabled — the Windows job resolves a cu128 torch and there is no
reason to refetch it every run.
Blocking rather than advisory, because the previous commit fixed the path
assertions that would otherwise have failed there: 153 passed, 0 failed.
---------
Co-authored-by: nanaoto <10inspiral@gmail.com>
644 lines
25 KiB
Python
644 lines
25 KiB
Python
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).resolve()
|
|
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).resolve()
|
|
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).resolve()
|
|
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).resolve()
|
|
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).resolve()
|
|
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).resolve()
|
|
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).resolve()
|
|
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).resolve()
|
|
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).resolve()
|
|
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).resolve()
|
|
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).resolve()
|
|
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).resolve()
|
|
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).resolve()
|
|
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).resolve()
|
|
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).resolve()
|
|
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).resolve()
|
|
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()
|