diff options
Diffstat (limited to 'lib/project/tests')
| -rw-r--r-- | lib/project/tests/conftest.py | 43 | ||||
| -rw-r--r-- | lib/project/tests/test_ai.py | 366 | ||||
| -rw-r--r-- | lib/project/tests/test_cli.py | 287 | ||||
| -rw-r--r-- | lib/project/tests/test_config.py | 134 | ||||
| -rw-r--r-- | lib/project/tests/test_install.py | 225 | ||||
| -rw-r--r-- | lib/project/tests/test_pipeline.py | 273 |
6 files changed, 1328 insertions, 0 deletions
diff --git a/lib/project/tests/conftest.py b/lib/project/tests/conftest.py new file mode 100644 index 0000000..7c75358 --- /dev/null +++ b/lib/project/tests/conftest.py @@ -0,0 +1,43 @@ +from pathlib import Path +import sys + +import numpy as np +import pytest +import soundfile as sf + +# Support running the suite before the project is installed. +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from voiceforge.config import resolve +from voiceforge.setup import ensure_ffmpeg + + +@pytest.fixture +def progress(): + return lambda *_: None + + +@pytest.fixture +def wav(tmp_path): + def write(samples=None, *, rate=48000, name="input.wav"): + if samples is None: + time = np.arange(rate * 3) / rate + samples = 0.15 * np.sin(2 * np.pi * 440 * time) + path = tmp_path / name + sf.write(path, np.asarray(samples), rate, format="WAV", subtype="FLOAT") + return path + + return write + + +@pytest.fixture +def bypass(): + return resolve(profile="cleanup-only", overrides={"denoiser": "none", "highpass": False}) + + +@pytest.fixture +def ffmpeg(): + try: + return ensure_ffmpeg() + except RuntimeError as error: + pytest.skip(str(error)) diff --git a/lib/project/tests/test_ai.py b/lib/project/tests/test_ai.py new file mode 100644 index 0000000..e8c5872 --- /dev/null +++ b/lib/project/tests/test_ai.py @@ -0,0 +1,366 @@ +"""Offline AI contract tests; no Torch installation, model download, or GPU needed.""" +import hashlib +import io +import json +import os +from pathlib import Path +import sys +import time +from types import SimpleNamespace +from unittest.mock import Mock +import zipfile + +import numpy as np +import pytest +import soundfile as sf + +from voiceforge import ai, ai_worker, setup + + +MODEL_BYTES = 7986207 +MODEL_HASH = "49c52edc8947ae1f9bf50d81530beaf3a2c3245aeaf34b6f31ff535cd22284d2" + + +@pytest.mark.parametrize("size,message", [ + (10, "checksum mismatch"), + (MODEL_BYTES, "checksum mismatch"), + (MODEL_BYTES + 1, "exceeds the expected size"), +]) +def test_model_rejects_truncated_corrupt_and_oversized_downloads(tmp_path, monkeypatch, size, message): + monkeypatch.setattr(ai_worker.urllib.request, "urlopen", Mock(return_value=io.BytesIO(b"x" * size))) + progress = Mock() + monkeypatch.setattr(ai_worker, "report", progress) + with pytest.raises(RuntimeError, match=message): + ai_worker.ensure_model(tmp_path / "model") + assert list(tmp_path.iterdir()) == [] + events = [call.args for call in progress.call_args_list] + assert events[0] == ("Downloading DeepFilterNet3 model (bytes)", 0, MODEL_BYTES) + counts = [completed for _, completed, total in events if total == MODEL_BYTES] + assert counts == sorted(counts) + assert all(0 <= count <= MODEL_BYTES for count in counts) + + +@pytest.mark.parametrize("unsafe", [False, True]) +def test_mocked_model_archive_progress_extraction_and_reuse(tmp_path, monkeypatch, unsafe): + # Mock only the release digest for a synthetic archive; corruption tests above + # exercise the real SHA-256 gate. Keep its real byte count and ZIP extraction. + def archive(padding): + stream = io.BytesIO() + with zipfile.ZipFile(stream, "w") as bundle: + bundle.writestr("DeepFilterNet3/config.ini", "[df]\nsr=48000\n") + bundle.writestr("DeepFilterNet3/checkpoints/model_120.ckpt.best", "checkpoint") + bundle.writestr("../escape" if unsafe else "padding", b"x" * padding) + return stream.getvalue() + + payload = archive(MODEL_BYTES - len(archive(0))) + assert len(payload) == MODEL_BYTES + digest = Mock(wraps=hashlib.sha256()) + digest.hexdigest.return_value = MODEL_HASH + monkeypatch.setattr(ai_worker.hashlib, "sha256", lambda: digest) + download = Mock(return_value=io.BytesIO(payload)) + monkeypatch.setattr(ai_worker.urllib.request, "urlopen", download) + progress = Mock() + monkeypatch.setattr(ai_worker, "report", progress) + directory = tmp_path / "model" + if unsafe: + with pytest.raises(RuntimeError, match="Unsafe path"): + ai_worker.ensure_model(directory) + assert not (tmp_path / "escape").exists() + assert list(tmp_path.iterdir()) == [] + else: + model = ai_worker.ensure_model(directory) + assert (model / "config.ini").is_file() + assert (model / "checkpoints/model_120.ckpt.best").is_file() + assert ai_worker.ensure_model(directory) == model + download.assert_called_once() + assert b"".join(call.args[0] for call in digest.update.call_args_list) == payload + events = [call.args for call in progress.call_args_list] + byte_events = [event for event in events if "(bytes)" in event[0]] + assert byte_events[0][1:] == (0, MODEL_BYTES) + assert byte_events[-1][1:] == (MODEL_BYTES, MODEL_BYTES) + assert len(byte_events) > 2 + assert events[-1] == ("Verifying and extracting DeepFilterNet3 model",) + + +def test_incomplete_model_fails_without_network(tmp_path, monkeypatch): + directory = tmp_path / "model" + directory.mkdir() + download = Mock(side_effect=AssertionError("Unexpected network call")) + monkeypatch.setattr(ai_worker.urllib.request, "urlopen", download) + with pytest.raises(RuntimeError, match="Incomplete model directory"): + ai_worker.ensure_model(directory) + download.assert_not_called() + + +@pytest.mark.parametrize("close_output", [False, True]) +def test_subprocess_heartbeat_does_not_block_on_silence_or_closed_output(close_output): + events = [] + code = "import os,time; " + if close_output: + code += "os.close(1); os.close(2); " + code += "time.sleep(0.8)" + setup.run_process([sys.executable, "-c", code], "Waiting", lambda *event: events.append(event)) + assert events[0] == ("Waiting", None, None) + elapsed = [completed for _, completed, total in events[1:] if total is None] + assert len(elapsed) >= 2 + assert elapsed == sorted(elapsed) + + +def test_subprocess_parses_split_byte_progress_records(): + events = [] + record = "VOICEFORGE_PROGRESS " + json.dumps(["Download (bytes)", 123, 456]) + "\n" + code = f"import os,time; os.write(1,{record[:12].encode()!r}); time.sleep(.3); os.write(1,{record[12:].encode()!r})" + setup.run_process([sys.executable, "-c", code], "Download", lambda *event: events.append(event)) + assert events[-1] == ("Download (bytes)", 123, 456) + + +@pytest.mark.parametrize("exception", [KeyboardInterrupt, RuntimeError]) +def test_callback_cancellation_terminates_and_reaps_child(monkeypatch, exception): + popen = setup.subprocess.Popen + children = [] + + def launch(*args, **kwargs): + child = popen(*args, **kwargs) + children.append(child) + return child + + def cancel(stage, completed, total): + if completed is not None: + raise exception("cancelled") + + monkeypatch.setattr(setup.subprocess, "Popen", launch) + started = time.monotonic() + with pytest.raises(exception, match="cancelled"): + setup.run_process([sys.executable, "-c", "import time; time.sleep(30)"], "Waiting", cancel) + assert len(children) == 1 and children[0].poll() is not None + assert time.monotonic() - started < 10 + + +def test_cancellation_kills_descendants_that_outlive_the_leader(tmp_path, monkeypatch): + sentinel = tmp_path / "grandchild.pid" + monkeypatch.setenv("GRANDCHILD_PID_FILE", str(sentinel)) + child_code = ( + "import os, subprocess, sys, time\n" + "subprocess.Popen([sys.executable, '-c', " + "\"import os, time; open(os.environ['GRANDCHILD_PID_FILE'], 'w')" + ".write(str(os.getpid())); time.sleep(60)\"])\n" + "while not os.path.exists(os.environ['GRANDCHILD_PID_FILE']):\n" + " time.sleep(0.02)\n" + ) + + def cancel(stage, completed, total): + if completed is not None and children and children[0].poll() is not None: + raise KeyboardInterrupt("cancelled") + + popen = setup.subprocess.Popen + children = [] + + def launch(*args, **kwargs): + child = popen(*args, **kwargs) + children.append(child) + return child + + monkeypatch.setattr(setup.subprocess, "Popen", launch) + with pytest.raises(KeyboardInterrupt, match="cancelled"): + setup.run_process([sys.executable, "-c", child_code], "Waiting", cancel) + + def alive(pid): + try: + for line in Path(f"/proc/{pid}/status").read_text().splitlines(): + if line.startswith("State:"): + return "Z" not in line + return True + except FileNotFoundError: + return False + + grandchild = int(sentinel.read_text()) + for _ in range(50): + if not alive(grandchild): + break + time.sleep(0.1) + else: + pytest.fail("A descendant survived the cancellation of its process group") + + +def test_subprocess_failure_keeps_bounded_diagnostics(): + code = "import sys; print('x'*100000); print('specific failure'); sys.exit(7)" + with pytest.raises(RuntimeError, match="exit 7") as error: + setup.run_process([sys.executable, "-c", code], "Worker") + assert "specific failure" in str(error.value) + assert len(str(error.value)) < 66000 + + +@pytest.mark.parametrize("strength", [-1, 2, float("nan"), float("inf")]) +def test_invalid_strength_fails_before_setup(tmp_path, monkeypatch, strength): + install = Mock(side_effect=AssertionError("Unexpected setup")) + monkeypatch.setattr(ai, "ensure_ai", install) + with pytest.raises(ValueError, match="strength"): + ai.denoise(tmp_path / "input.wav", tmp_path / "output.wav", strength=strength) + install.assert_not_called() + + +def _ready_environment(root, flavor="cpu"): + environment = root / f"ai-df-0.5.6-torch-2.5.1-{flavor}-py311-v1" + (environment / "bin").mkdir(parents=True) + (environment / "bin/python").write_text("#!/bin/sh\n") + (environment / ".voiceforge-ready").touch() + return environment + + +def test_ready_environment_skips_the_check_worker(tmp_path, monkeypatch): + root = tmp_path / "prefix" + environment = _ready_environment(root) + monkeypatch.setenv("VOICEFORGE_HOME", str(root)) + monkeypatch.setattr(ai.os, "confstr", lambda *_: "glibc 2.39") + executed = [] + monkeypatch.setattr(ai, "run_process", lambda args, *rest, **kwargs: executed.append(args)) + python = ai.ensure_ai("cpu") + assert executed == [] + assert python == environment / "bin/python" + + +def test_setup_verifies_even_a_ready_environment(tmp_path, monkeypatch): + root = tmp_path / "prefix" + _ready_environment(root) + monkeypatch.setenv("VOICEFORGE_HOME", str(root)) + monkeypatch.setattr(ai.os, "confstr", lambda *_: "glibc 2.39") + executed = [] + monkeypatch.setattr(ai, "run_process", lambda args, *rest, **kwargs: executed.append(args)) + ai.ensure_ai("cpu", verify=True) + assert len(executed) == 1 + assert executed[0][-1] == "--check" + + +def test_broken_environment_is_rebuilt_and_self_tested(tmp_path, monkeypatch): + root = tmp_path / "prefix" + environment = _ready_environment(root) + (environment / "bin/python").unlink() + (environment / "bin/python").symlink_to("/nonexistent/python3.11") + leftover = environment / "leftover.txt" + leftover.write_text("from the previous location") + monkeypatch.setenv("VOICEFORGE_HOME", str(root)) + monkeypatch.setattr(ai.os, "confstr", lambda *_: "glibc 2.39") + stages, commands = [], [] + + def fake_run(args, stage, progress=None, **kwargs): + commands.append(args) + stages.append(stage) + if stage == "Checking DeepFilterNet3": + environment.mkdir(parents=True, exist_ok=True) + (environment / ".voiceforge-ready").touch() + + monkeypatch.setattr(ai, "run_process", fake_run) + monkeypatch.setattr(ai, "ensure_uv", lambda progress=None: "uv") + ai.ensure_ai("cpu") + assert not leftover.exists() + assert stages == ["Preparing AI Python 3.11", "Installing PyTorch (cpu)", + "Installing DeepFilterNet3", "Checking DeepFilterNet3"] + assert commands[0][-1] == str(environment) + assert (environment / ".voiceforge-ready").is_file() + + +def test_copied_environment_is_rebuilt_not_trusted(tmp_path, monkeypatch): + # Copying an installation while the original remains leaves the copy's + # interpreter links resolving into the original prefix; that must not + # count as ready, or the copy breaks when the original is removed. + root = tmp_path / "prefix" + environment = _ready_environment(root) + original = _ready_environment(tmp_path / "other") + (environment / "bin/python").unlink() + (environment / "bin/python").symlink_to(original / "bin/python") + monkeypatch.setenv("VOICEFORGE_HOME", str(root)) + monkeypatch.setattr(ai.os, "confstr", lambda *_: "glibc 2.39") + stages = [] + + def fake_run(args, stage, progress=None, **kwargs): + stages.append(stage) + if stage == "Checking DeepFilterNet3": + environment.mkdir(parents=True, exist_ok=True) + (environment / ".voiceforge-ready").touch() + + monkeypatch.setattr(ai, "run_process", fake_run) + monkeypatch.setattr(ai, "ensure_uv", lambda progress=None: "uv") + ai.ensure_ai("cpu") + assert stages[0] == "Preparing AI Python 3.11" + + +@pytest.fixture +def mocked_worker(tmp_path, monkeypatch): + class Tensor: + def __init__(self, array): + self.array = array + + def unsqueeze(self, axis): + return Tensor(np.expand_dims(self.array, axis)) + + def squeeze(self, axis): + return Tensor(np.squeeze(self.array, axis)) + + def numpy(self): + return self.array + + calls = [] + + def enhance(model, state, tensor, pad): + assert pad is True + assert tensor.array.shape[1] % 480 == 0 + calls.append(tensor.array.shape[1]) + return Tensor(tensor.array * 0.5) + + state = SimpleNamespace(sr=lambda: 48000) + backend = SimpleNamespace(enhance=enhance, init_df=lambda *a, **kw: (None, state, None)) + for name, module in { + "torch": SimpleNamespace(set_num_threads=lambda n: None, from_numpy=Tensor, + cuda=SimpleNamespace(is_available=lambda: False)), + "df": SimpleNamespace(), + "df.enhance": backend, + "df.model": SimpleNamespace(ModelParams=lambda: SimpleNamespace( + sr=48000, fft_size=960, hop_size=480, nb_erb=32, min_nb_freqs=2)), + "libdf": SimpleNamespace(DF=lambda **kw: state), + }.items(): + monkeypatch.setitem(sys.modules, name, module) + monkeypatch.setattr(ai_worker, "ensure_model", lambda path: path) + monkeypatch.setattr(ai_worker.signal, "signal", lambda *args: None) + monkeypatch.setenv("DEVICE", "cpu") + progress = Mock() + monkeypatch.setattr(ai_worker, "report", progress) + + def run(source, target, strength): + monkeypatch.setattr(sys, "argv", ["worker", "--device", "cpu", "--model-dir", str(tmp_path), + "--source", str(source), "--target", str(target), + "--strength", str(strength)]) + ai_worker.main() + + return run, calls, backend, progress + + +@pytest.mark.parametrize("frames", [0, 1, 479, 481, 480001, 960017]) +@pytest.mark.parametrize("strength", [0, 0.85, 1]) +def test_mocked_worker_exact_length_mix_and_bounded_windows(tmp_path, wav, mocked_worker, frames, strength): + run, calls, _, progress = mocked_worker + audio = np.random.default_rng(42).normal(0, 0.1, frames).astype("float32") + source = wav(audio) + target = tmp_path / "clean.wav" + run(source, target, strength) + result, rate = sf.read(target, dtype="float32") + assert len(result) == frames and rate == 48000 + assert sf.info(target).subtype == "FLOAT" + np.testing.assert_allclose(result, audio * (1 - strength * 0.5), rtol=2e-6, atol=1e-8) + if strength == 0: + np.testing.assert_array_equal(result, audio) + assert calls == [] + assert max(calls, default=0) <= 674400 + assert progress.call_args.args == ("Denoising (cpu)", frames, frames) + assert not list(tmp_path.glob(".voiceforge-ai-*")) + + +def test_worker_failure_preserves_existing_target(tmp_path, wav, mocked_worker): + run, _, backend, _ = mocked_worker + backend.enhance = Mock(side_effect=RuntimeError("inference failed")) + source = wav() + target = tmp_path / "existing.wav" + target.write_bytes(b"existing output") + with pytest.raises(RuntimeError, match="inference failed"): + run(source, target, 0.85) + assert target.read_bytes() == b"existing output" + assert not list(tmp_path.glob(".voiceforge-ai-*")) diff --git a/lib/project/tests/test_cli.py b/lib/project/tests/test_cli.py new file mode 100644 index 0000000..a5437b7 --- /dev/null +++ b/lib/project/tests/test_cli.py @@ -0,0 +1,287 @@ +import json +from pathlib import Path +import sys + +import numpy as np +import pytest +import soundfile as sf + +from voiceforge import cli, pipeline + + +@pytest.fixture +def invoke(monkeypatch): + def run(*args): + monkeypatch.setattr(sys, "argv", ["voiceforge", *map(str, args)]) + return cli.main() + + return run + + +def test_preview_real_audio_preserves_original_and_reports_provenance( + wav, tmp_path, ffmpeg, invoke, capsys, monkeypatch +): + time = np.arange(48000 * 7) / 48000 + source = wav(0.15 * np.sin(2 * np.pi * 440 * time)) + original = source.read_bytes() + output_dir = tmp_path / "previews" + modes = [] + real_run = pipeline.ffmpeg_run + + def record_render(*args, **kwargs): + log = real_run(*args, **kwargs) + if args[2] == "Mastering (pass 2)": + modes.append(json.loads(log[log.rfind("{"):log.rfind("}") + 1])["normalization_type"]) + return log + + monkeypatch.setattr(pipeline, "ffmpeg_run", record_render) + invoke("preview", source, "--no-denoise", "--start", "1", "--duration", "5", + "--profiles", "natural", "radio", "--no-normalize", "--no-limiter", + "--output-dir", output_dir) + + captured = capsys.readouterr() + assert captured.out == "" + assert "Preview:" in captured.err + assert source.read_bytes() == original + assert len(modes) == 2 + assert {path.name for path in output_dir.iterdir()} == { + f"input.{profile}.preview{suffix}" + for profile in ("natural", "radio") for suffix in (".wav", ".report.json") + } + for profile, mode in zip(("natural", "radio"), modes): + target = output_dir / f"input.{profile}.preview.wav" + report = json.loads(target.with_suffix(".report.json").read_text()) + info = sf.info(target) + assert info.duration == pytest.approx(5, abs=0.02) + assert info.channels == 1 + assert info.subtype == "PCM_24" + assert report["output"] == str(target) + assert report["provenance"] == { + "original": str(source.resolve()), "preview_start_seconds": 1, + "requested_duration_seconds": 5, "normalization_forced": True, + } + assert not Path(report["source"]).exists() + assert report["settings"]["profile"] == profile + assert report["settings"]["denoiser"] == "none" + assert report["settings"]["normalize"] is True + assert report["settings"]["limiter"] is True + assert report["settings"]["compressor_ratio"] == {"natural": 2.0, "radio": 3.5}[profile] + assert report["normalization_mode"] == mode + assert report["master_loudness"]["input_i"] == pytest.approx(-19, abs=0.5) + assert report["master_loudness"]["input_tp"] <= -1.4 + + +def test_config_stdout_json_and_set_precedence(tmp_path, invoke, capsys): + config = tmp_path / "settings.toml" + config.write_text('profile = "cleanup-only"\ntarget_lufs = -24.0\n') + invoke("config", "--json", "--config", config, "--profile", "narrator", + "--target-lufs", "-21", "--set", "target_lufs=-18", + "--set", "profile=radio", "--no-denoise") + captured = capsys.readouterr() + settings = json.loads(captured.out) + assert captured.err == "" + assert settings["profile"] == "radio" + assert settings["compressor_ratio"] == 3.5 + assert settings["compressor_attack_ms"] == 8 + assert settings["normalize"] is True + assert settings["target_lufs"] == -18 + assert settings["denoiser"] == "none" + + +def test_analyze_stdout_is_json(wav, ffmpeg, invoke, capsys): + source = wav() + original = source.read_bytes() + invoke("analyze", source, "--no-denoise") + stats = json.loads(capsys.readouterr().out) + assert stats["duration_seconds"] == pytest.approx(3) + assert stats["channels"] == 1 + assert stats["frames"] == 144000 + assert np.isfinite(stats["loudness"]["input_i"]) + assert source.read_bytes() == original + assert list(source.parent.iterdir()) == [source] + + +@pytest.mark.parametrize("start", ["nan", "inf", "-inf"]) +def test_preview_rejects_nonfinite_start_before_tool_setup( + tmp_path, invoke, capsys, monkeypatch, start +): + monkeypatch.setattr(cli, "ensure_ffmpeg", lambda: pytest.fail("Invalid start reached FFmpeg")) + output_dir = tmp_path / "previews" + with pytest.raises(SystemExit) as error: + invoke("preview", tmp_path / "missing.wav", f"--start={start}", + "--output-dir", output_dir) + assert error.value.code == 1 + captured = capsys.readouterr() + assert captured.out == "" + assert "Preview start" in captured.err + assert not output_dir.exists() + + +def test_unknown_setting_is_a_cli_error(invoke, capsys): + with pytest.raises(SystemExit) as error: + invoke("config", "--json", "--set", "unknown_setting=1") + assert error.value.code == 1 + captured = capsys.readouterr() + assert captured.out == "" + assert "Unknown setting(s): unknown_setting" in captured.err + + +def test_missing_input_fails_before_tool_setup(tmp_path, invoke, capsys, monkeypatch): + monkeypatch.setattr(pipeline, "ensure_ffmpeg", lambda: pytest.fail("Missing input reached FFmpeg")) + target = tmp_path / "master.wav" + with pytest.raises(SystemExit) as error: + invoke("process", tmp_path / "missing.wav", "--no-denoise", "-o", target) + assert error.value.code == 1 + assert "existing WAV file" in capsys.readouterr().err + assert not target.exists() + + +def test_existing_output_is_not_overwritten(wav, tmp_path, invoke, capsys, monkeypatch): + source = wav() + original = source.read_bytes() + target = tmp_path / "master.wav" + target.write_bytes(b"previous master") + monkeypatch.setattr(pipeline, "ensure_ffmpeg", lambda: pytest.fail("Existing output reached FFmpeg")) + with pytest.raises(SystemExit) as error: + invoke("process", source, "--no-denoise", "-o", target) + assert error.value.code == 1 + assert "Output exists" in capsys.readouterr().err + assert target.read_bytes() == b"previous master" + assert source.read_bytes() == original + assert not target.with_suffix(".report.json").exists() + + +def test_overwrite_cannot_replace_input(wav, invoke, capsys, monkeypatch): + source = wav() + original = source.read_bytes() + monkeypatch.setattr(cli, "process", lambda *a, **kw: pytest.fail("Input collision reached processing")) + with pytest.raises(SystemExit) as error: + invoke("process", source, "-o", source, "--overwrite") + assert error.value.code == 1 + assert "overwrite an input recording" in capsys.readouterr().err + assert source.read_bytes() == original + + +def test_duplicate_batch_names_fail_before_processing(wav, tmp_path, invoke, capsys, monkeypatch): + (tmp_path / "other").mkdir() + sources = [wav(), wav(name="other/input.wav")] + originals = [source.read_bytes() for source in sources] + output_dir = tmp_path / "masters" + monkeypatch.setattr(cli, "process", lambda *a, **kw: pytest.fail("Duplicate names reached processing")) + with pytest.raises(SystemExit) as error: + invoke("process", *sources, "--output-dir", output_dir, "--overwrite") + assert error.value.code == 1 + assert "duplicate output names" in capsys.readouterr().err + assert not output_dir.exists() + assert [source.read_bytes() for source in sources] == originals + + +def test_default_outputs_are_written_beside_each_input(wav, tmp_path, ffmpeg, invoke, capsys): + (tmp_path / "a").mkdir() + (tmp_path / "b").mkdir() + sources = [wav(name="a/take.wav"), wav(name="b/take.wav")] + invoke("process", *sources, "--no-denoise") + for source in sources: + assert {path.name for path in source.parent.iterdir()} == { + "take.wav", "take.natural.wav", "take.natural.report.json"} + captured = capsys.readouterr() + assert captured.out == "" + assert captured.err.count("Written:") == 2 + + +def test_output_and_output_dir_cannot_be_combined(wav, tmp_path, invoke, capsys): + with pytest.raises(SystemExit) as error: + invoke("process", wav(), "-o", tmp_path / "x.wav", "--output-dir", tmp_path / "outs") + assert error.value.code == 1 + assert "--output-dir cannot be combined" in capsys.readouterr().err + + +def test_missing_later_input_fails_before_any_processing(wav, tmp_path, invoke, capsys, monkeypatch): + source = wav() + monkeypatch.setattr(cli, "process", lambda *a, **kw: pytest.fail("Preflight must reject the batch")) + with pytest.raises(SystemExit) as error: + invoke("process", source, tmp_path / "missing.wav", "--no-denoise") + assert error.value.code == 1 + assert "existing WAV file" in capsys.readouterr().err + assert not (tmp_path / "input.natural.wav").exists() + + +def test_existing_output_in_batch_fails_before_any_processing(wav, tmp_path, invoke, capsys, monkeypatch): + (tmp_path / "a").mkdir() + (tmp_path / "b").mkdir() + sources = [wav(name="a/take.wav"), wav(name="b/take.wav")] + (tmp_path / "b" / "take.natural.wav").write_bytes(b"previous result") + monkeypatch.setattr(cli, "process", lambda *a, **kw: pytest.fail("Preflight must reject the batch")) + with pytest.raises(SystemExit) as error: + invoke("process", *sources, "--no-denoise") + assert error.value.code == 1 + assert "Output exists" in capsys.readouterr().err + assert not (tmp_path / "a" / "take.natural.wav").exists() + + +def test_preview_rejects_duplicate_profiles_and_profile_override(wav, invoke, capsys, monkeypatch): + source = wav() + monkeypatch.setattr(cli, "ensure_ffmpeg", lambda: pytest.fail("Invalid preview plan reached FFmpeg")) + with pytest.raises(SystemExit) as error: + invoke("preview", source, "--no-denoise", "--profiles", "natural", "natural") + assert error.value.code == 1 + assert "unique" in capsys.readouterr().err + with pytest.raises(SystemExit) as error: + invoke("preview", source, "--no-denoise", "--profiles", "natural", + "--set", "profile=radio") + assert error.value.code == 1 + assert "remove the profile override" in capsys.readouterr().err + assert list(source.parent.iterdir()) == [source] + + +def test_preview_default_output_is_written_beside_input(wav, tmp_path, ffmpeg, invoke, capsys): + source = wav() + original = source.read_bytes() + invoke("preview", source, "--no-denoise", "--start", "1", "--duration", "2", + "--profiles", "natural") + assert {path.name for path in tmp_path.iterdir()} == { + "input.wav", "input.natural.preview.wav", "input.natural.preview.report.json"} + report = json.loads((tmp_path / "input.natural.preview.report.json").read_text()) + assert report["settings"]["profile"] == "natural" + assert source.read_bytes() == original + + +def test_preview_cancellation_cleans_excerpt(wav, tmp_path, ffmpeg, invoke, capsys, monkeypatch): + source = wav() + original = source.read_bytes() + output_dir = tmp_path / "previews" + excerpts = [] + + def cancel(excerpt, *args, **kwargs): + assert excerpt.is_file() + excerpts.append(excerpt) + raise KeyboardInterrupt + + monkeypatch.setattr(cli, "process", cancel) + with pytest.raises(SystemExit) as error: + invoke("preview", source, "--no-denoise", "--output-dir", output_dir) + assert error.value.code == 130 + assert "Cancelled" in capsys.readouterr().err + assert len(excerpts) == 1 + assert not excerpts[0].exists() + assert list(output_dir.iterdir()) == [] + assert source.read_bytes() == original + + +@pytest.mark.parametrize("suffix", [".wav", ".report.json", ".mp3"]) +def test_nonfile_output_with_overwrite_fails_before_tool_setup( + wav, tmp_path, bypass, progress, monkeypatch, suffix +): + source = wav() + original = source.read_bytes() + target = tmp_path / "master.wav" + nonfile = target.with_suffix(suffix) + nonfile.mkdir() + bypass.mp3 = True + monkeypatch.setattr(pipeline, "ensure_ffmpeg", lambda: pytest.fail("Non-file output reached FFmpeg")) + with pytest.raises(ValueError, match="Output is not a regular file"): + pipeline.process(source, target, bypass, progress, overwrite=True) + assert nonfile.is_dir() + assert list(nonfile.iterdir()) == [] + assert set(tmp_path.iterdir()) == {source, nonfile} + assert source.read_bytes() == original diff --git a/lib/project/tests/test_config.py b/lib/project/tests/test_config.py new file mode 100644 index 0000000..8144fc8 --- /dev/null +++ b/lib/project/tests/test_config.py @@ -0,0 +1,134 @@ +from dataclasses import asdict, fields + +import pytest + +from voiceforge.cli import main, parser, settings_for +from voiceforge.config import PROFILES, RANGES, Settings, resolve, toml +from voiceforge.pipeline import filters + + +@pytest.mark.parametrize("profile", PROFILES) +def test_profiles_and_toml_round_trip(profile, tmp_path): + settings = resolve(profile=profile) + assert asdict(settings) == asdict(Settings()) | PROFILES[profile] | {"profile": profile} + path = tmp_path / "settings.toml" + path.write_text(toml(settings)) + assert resolve(path) == settings + + +def test_precedence_defaults_profile_file_flags_then_set(tmp_path): + path = tmp_path / "settings.toml" + path.write_text('profile = "narrator"\ncompressor_ratio = 2.2\nwarmth_db = 3\n') + args = parser().parse_args([ + "config", "--config", str(path), "--profile", "radio", + "--warmth-db", "4", "--set", "warmth_db=5", "--set", "warmth_db=6", + ]) + settings = settings_for(args) + assert settings.profile == "radio" + assert settings.compressor_attack_ms == 8 # Selected profile beats file profile. + assert settings.compressor_ratio == 2.2 # File values beat profile defaults. + assert settings.warmth_db == 6 # Last --set beats named flags. + assert settings.sample_rate == 48000 + + +@pytest.mark.parametrize("field", [f.name for f in fields(Settings) if type(getattr(Settings(), f.name)) is bool]) +def test_boolean_flags_preserve_unspecified_and_accept_both_forms(field): + assert getattr(parser().parse_args(["config"]), field) is None + for enabled in (True, False): + flag = "--" + ("" if enabled else "no-") + field.replace("_", "-") + extra = ["--limiter"] if field == "normalize" and enabled else [] + args = parser().parse_args(["config", "--profile", "cleanup-only", flag, *extra]) + assert getattr(settings_for(args), field) is enabled + + +def test_false_cli_toggle_overrides_true_file_value(tmp_path): + path = tmp_path / "settings.toml" + path.write_text("eq = true\n") + args = parser().parse_args(["config", "--config", str(path), "--no-eq"]) + assert settings_for(args).eq is False + + +def test_no_denoise_and_unquoted_set_strings(): + assert settings_for(parser().parse_args(["config", "--no-denoise"])).denoiser == "none" + args = parser().parse_args(["config", "--no-denoise", "--set", "denoiser=fft"]) + assert settings_for(args).denoiser == "fft" + + +@pytest.mark.parametrize("key", RANGES) +def test_numeric_ranges_are_inclusive_and_reject_outside(key): + low, high = RANGES[key] + for value in (low, high): + assert getattr(resolve(overrides={key: value}), key) == value + for value in (low - 0.01, high + 0.01): + with pytest.raises(ValueError, match=key): + resolve(overrides={key: value}) + + +@pytest.mark.parametrize("overrides,match", [ + ({"typo": True}, "Unknown setting"), + ({"highpass": 1}, "highpass must be bool"), + ({"sample_rate": 48000.0}, "sample_rate must be int"), + ({"target_lufs": True}, "finite number"), + ({"target_lufs": float("nan")}, "finite number"), + ({"target_lufs": float("inf")}, "finite number"), + ({"denoiser": "unknown"}, "denoiser must be one of"), + ({"device": "gpu"}, "device must be one of"), + ({"channel": "stereo"}, "channel must be one of"), + ({"hum_hz": 55}, "hum_hz must be one of"), + ({"sample_rate": 96000}, "sample_rate must be one of"), + ({"mp3_bitrate": 64}, "mp3_bitrate must be one of"), + ({"limiter": False}, "Normalization includes true-peak limiting"), +]) +def test_invalid_configuration(overrides, match): + with pytest.raises(ValueError, match=match): + resolve(overrides=overrides) + + +def test_unknown_profile_and_malformed_toml(tmp_path): + with pytest.raises(ValueError, match="Unknown profile"): + resolve(profile="missing") + path = tmp_path / "bad.toml" + path.write_text("normalize = [") + with pytest.raises(ValueError): + resolve(path) + + +def test_invalid_cli_configuration_fails_before_processing(monkeypatch, capsys): + def unexpected(*args, **kwargs): + pytest.fail("Invalid configuration must not reach processing or tool setup") + + monkeypatch.setattr("voiceforge.cli.process", unexpected) + monkeypatch.setattr("voiceforge.cli.ensure_ffmpeg", unexpected) + monkeypatch.setattr("sys.argv", ["voiceforge", "process", "missing.wav", "--target-lufs", "0"]) + with pytest.raises(SystemExit) as error: + main() + assert error.value.code == 1 + assert "target_lufs must be between" in capsys.readouterr().err + + +def test_set_requires_assignment(): + with pytest.raises(ValueError, match="KEY=VALUE"): + settings_for(parser().parse_args(["config", "--set", "normalize"])) + + +@pytest.mark.parametrize("profile", ["radio", "not-a-profile"]) +def test_set_profile_uses_normal_profile_resolution(profile): + args = parser().parse_args(["config", "--set", f"profile={profile}"]) + if profile not in PROFILES: + with pytest.raises(ValueError, match="Unknown profile"): + settings_for(args) + else: + assert settings_for(args) == resolve(profile=profile) + + +@pytest.mark.parametrize("toggle,prefix,count", [ + ("expansion", "agate=", 1), ("eq", "equalizer=", 3), + ("compression", "acompressor=", 1), ("deess", "deesser=", 1), + ("lowpass", "lowpass=", 1), +]) +def test_shaping_filter_toggles(bypass, toggle, prefix, count): + assert filters(bypass) == ["anull"] + setattr(bypass, toggle, True) + chain = filters(bypass) + assert len(chain) == count + assert all(item.startswith(prefix) for item in chain) diff --git a/lib/project/tests/test_install.py b/lib/project/tests/test_install.py new file mode 100644 index 0000000..f1dfcf2 --- /dev/null +++ b/lib/project/tests/test_install.py @@ -0,0 +1,225 @@ +"""Data-dir resolution, contained tool environment, and launcher script checks.""" +import os +import subprocess +from pathlib import Path +import sys +import types + +import pytest + +from voiceforge import setup + +ROOT = Path(__file__).resolve().parents[3] +LAUNCHER = ROOT / "producer.sh" + + +@pytest.fixture +def shell_env(tmp_path): + env = os.environ.copy() + env.pop("VOICEFORGE_HOME", None) + env.pop("XDG_DATA_HOME", None) + env["HOME"] = str(tmp_path / "home") + return env + + +# --- data_dir resolution --------------------------------------------------- + +def test_data_dir_defaults_and_precedence(tmp_path, monkeypatch): + monkeypatch.delenv("VOICEFORGE_HOME", raising=False) + monkeypatch.delenv("XDG_DATA_HOME", raising=False) + monkeypatch.setenv("HOME", str(tmp_path)) + assert setup.data_dir() == tmp_path / ".local/share/voiceforge" + monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "xdg")) + assert setup.data_dir() == tmp_path / "xdg/voiceforge" + monkeypatch.setenv("VOICEFORGE_HOME", str(tmp_path / "prefix")) + assert setup.data_dir() == tmp_path / "prefix" + + +def test_relative_voiceforge_home_resolves_against_cwd(tmp_path, monkeypatch): + monkeypatch.setenv("VOICEFORGE_HOME", "voiceforge-home") + monkeypatch.chdir(tmp_path) + assert setup.data_dir() == tmp_path / "voiceforge-home" + + +def test_relative_xdg_data_home_falls_back_to_home(tmp_path, monkeypatch): + monkeypatch.delenv("VOICEFORGE_HOME", raising=False) + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("XDG_DATA_HOME", "relative/path") + assert setup.data_dir() == tmp_path / ".local/share/voiceforge" + + +# --- contained tool environment -------------------------------------------- + +def test_tool_env_is_sanitized_and_contained(tmp_path, monkeypatch): + monkeypatch.setenv("VOICEFORGE_HOME", str(tmp_path / "prefix")) + monkeypatch.setenv("PYTHONPATH", "/host/site-packages") + monkeypatch.setenv("PYTHONHOME", "/host/python") + monkeypatch.setenv("VIRTUAL_ENV", "/host/venv") + monkeypatch.setenv("PIP_TARGET", "/host/target") + monkeypatch.setenv("PIP_PREFIX", "/host/prefix") + monkeypatch.setenv("PIP_USER", "1") + monkeypatch.setenv("PIP_FIND_LINKS", "/host/wheels") + monkeypatch.setenv("PIP_INDEX_URL", "https://host.example/simple") + monkeypatch.setenv("PIP_EXTRA_INDEX_URL", "https://host.example/extra") + monkeypatch.setenv("PIP_CONSTRAINT", "/host/constraints.txt") + monkeypatch.setenv("UV_INDEX_URL", "https://host.example/simple") + monkeypatch.setenv("UV_DEFAULT_INDEX", "https://host.example/simple") + monkeypatch.setenv("UV_EXTRA_INDEX_URL", "https://host.example/extra") + monkeypatch.setenv("UV_INDEX", "https://host.example/extra") + monkeypatch.setenv("UV_FIND_LINKS", "/host/wheels") + monkeypatch.setenv("UV_CONSTRAINT", "/host/constraints.txt") + monkeypatch.delenv("PIP_CACHE_DIR", raising=False) + env = setup.tool_env() + prefix = str(tmp_path / "prefix") + assert "PYTHONPATH" not in env and "PYTHONHOME" not in env and "VIRTUAL_ENV" not in env + for name in ("PIP_TARGET", "PIP_PREFIX", "PIP_USER", "PIP_FIND_LINKS", + "PIP_INDEX_URL", "PIP_EXTRA_INDEX_URL", "PIP_CONSTRAINT", + "UV_INDEX_URL", "UV_DEFAULT_INDEX", "UV_EXTRA_INDEX_URL", + "UV_INDEX", "UV_FIND_LINKS", "UV_CONSTRAINT"): + assert name not in env, name + assert env["PYTHONNOUSERSITE"] == "1" + assert env["PIP_CONFIG_FILE"] == os.devnull + assert env["UV_NO_CONFIG"] == "1" + assert env["XDG_CACHE_HOME"] == f"{prefix}/cache" + assert env["PIP_CACHE_DIR"] == f"{prefix}/cache/pip" + assert env["UV_CACHE_DIR"] == f"{prefix}/cache/uv" + assert env["UV_PYTHON_INSTALL_DIR"] == f"{prefix}/uv/python" + assert env["TMPDIR"] == f"{prefix}/tmp" + assert (tmp_path / "prefix/tmp").is_dir() + + +def test_tool_env_tolerates_an_uncreatable_data_dir(monkeypatch): + monkeypatch.setenv("VOICEFORGE_HOME", "/vf-prefix-unwritable") + env = setup.tool_env() + assert env["PIP_CACHE_DIR"] == "/vf-prefix-unwritable/cache/pip" + assert "TMPDIR" not in env + + +def test_tool_env_leaves_process_environment_untouched(tmp_path, monkeypatch): + monkeypatch.setenv("VOICEFORGE_HOME", str(tmp_path / "prefix")) + monkeypatch.setenv("PYTHONPATH", "/host/site-packages") + setup.tool_env() + assert os.environ["PYTHONPATH"] == "/host/site-packages" + + +def test_ensure_uv_bootstraps_into_data_dir_with_contained_pip_cache(tmp_path, monkeypatch): + monkeypatch.setenv("VOICEFORGE_HOME", str(tmp_path / "prefix")) + monkeypatch.setattr(setup.shutil, "which", lambda name: None) + calls = [] + uv_binary = tmp_path / "prefix/bootstrap/bin/uv" + + def fake_run(args, stage, progress=None, *, env=None): + calls.append((args, env)) + # Simulate pip actually producing the pinned uv. + if stage == "Installing uv from PyPI": + uv_binary.parent.mkdir(parents=True, exist_ok=True) + uv_binary.write_text("#!/bin/sh\n") + uv_binary.chmod(0o755) + + monkeypatch.setattr(setup, "run_process", fake_run) + uv = setup.ensure_uv() + assert uv == str(uv_binary) + venv_args, venv_env = calls[0] + pip_args, pip_env = calls[1] + assert venv_args[-1] == str(tmp_path / "prefix/bootstrap") + assert pip_env["PIP_CACHE_DIR"] == str(tmp_path / "prefix/cache/pip") + assert pip_env["UV_CACHE_DIR"] == str(tmp_path / "prefix/cache/uv") + assert pip_env["PIP_CONFIG_FILE"] == os.devnull + assert "PYTHONPATH" not in pip_env + assert "uv==0.8.17" in pip_args + + +def test_ensure_uv_reuses_contained_bootstrap(tmp_path, monkeypatch): + monkeypatch.setenv("VOICEFORGE_HOME", str(tmp_path / "prefix")) + uv_binary = tmp_path / "prefix/bootstrap/bin/uv" + uv_binary.parent.mkdir(parents=True) + uv_binary.write_text("#!/bin/sh\n") + uv_binary.chmod(0o755) + monkeypatch.setattr(setup.shutil, "which", lambda name: None) + assert setup.ensure_uv() == str(uv_binary) + + +def test_ensure_uv_prefers_the_contained_bootstrap_over_a_host_uv(tmp_path, monkeypatch): + monkeypatch.setenv("VOICEFORGE_HOME", str(tmp_path / "prefix")) + uv_binary = tmp_path / "prefix/bootstrap/bin/uv" + uv_binary.parent.mkdir(parents=True) + uv_binary.write_text("#!/bin/sh\n") + uv_binary.chmod(0o755) + monkeypatch.setattr(setup.shutil, "which", lambda name: "/host/bin/uv") + assert setup.ensure_uv() == str(uv_binary) + + +def test_ensure_uv_ignores_a_host_uv_without_required_capability(tmp_path, monkeypatch): + monkeypatch.setenv("VOICEFORGE_HOME", str(tmp_path / "prefix")) + monkeypatch.setattr(setup.shutil, "which", lambda name: "/host/bin/uv") + monkeypatch.setattr(setup, "_uv_supports_relocatable", lambda uv: False) + stages = [] + uv_binary = tmp_path / "prefix/bootstrap/bin/uv" + + def fake_run(args, stage, progress=None, *, env=None): + stages.append(stage) + if stage == "Installing uv from PyPI": + uv_binary.parent.mkdir(parents=True, exist_ok=True) + uv_binary.write_text("#!/bin/sh\n") + uv_binary.chmod(0o755) + + monkeypatch.setattr(setup, "run_process", fake_run) + assert setup.ensure_uv() == str(uv_binary) + assert stages == ["Creating uv bootstrap", "Installing uv from PyPI"] + + +def test_ensure_uv_fails_when_bootstrap_does_not_produce_uv(tmp_path, monkeypatch): + monkeypatch.setenv("VOICEFORGE_HOME", str(tmp_path / "prefix")) + monkeypatch.setattr(setup.shutil, "which", lambda name: None) + monkeypatch.setattr(setup, "run_process", lambda *args, **kwargs: None) + with pytest.raises(RuntimeError, match="did not produce a usable uv"): + setup.ensure_uv() + + +# --- FFmpeg selection --------------------------------------------------------- + +def test_ensure_ffmpeg_prefers_the_bundled_binary_over_host(tmp_path, monkeypatch): + bundled = tmp_path / "bundled/ffmpeg" + bundled.parent.mkdir(parents=True) + bundled.write_text("#!/bin/sh\n") + monkeypatch.setitem(sys.modules, "imageio_ffmpeg", + types.SimpleNamespace(get_ffmpeg_exe=lambda: str(bundled))) + monkeypatch.setattr(setup.shutil, "which", lambda name: "/usr/bin/ffmpeg") + assert setup.ensure_ffmpeg() == str(bundled) + + +def test_ensure_ffmpeg_falls_back_to_system_without_the_bundle(monkeypatch): + monkeypatch.setitem(sys.modules, "imageio_ffmpeg", None) + monkeypatch.setattr(setup.shutil, "which", lambda name: "/usr/bin/ffmpeg") + assert setup.ensure_ffmpeg() == "/usr/bin/ffmpeg" + + +def test_ensure_ffmpeg_error_points_at_the_launcher(monkeypatch): + monkeypatch.setitem(sys.modules, "imageio_ffmpeg", None) + monkeypatch.setattr(setup.shutil, "which", lambda name: None) + with pytest.raises(RuntimeError, match=r"\./producer\.sh --rebuild") as error: + setup.ensure_ffmpeg() + assert "install.sh" not in str(error.value) + + +# --- producer.sh launcher ---------------------------------------------------- + +@pytest.mark.parametrize("argv", [["--help"], ["--rebuild", "--help"]]) +def test_launcher_forwards_help_without_bootstrapping(argv, shell_env, tmp_path, monkeypatch): + # A missing lib/ must not be created by a pure help request. + staging = tmp_path / "checkout" + staging.mkdir() + (staging / "producer.sh").write_text(LAUNCHER.read_text()) + (staging / "producer.sh").chmod(0o755) + monkeypatch.setenv("PATH", "/usr/bin:/bin") + result = subprocess.run(["./producer.sh", *argv], cwd=staging, env=shell_env, + capture_output=True, text=True, timeout=30) + assert result.returncode == 0 + assert "process" in result.stdout + assert not (staging / "lib").exists() + + +def test_launcher_script_parses(): + result = subprocess.run(["bash", "-n", str(LAUNCHER)], + capture_output=True, text=True, timeout=30) + assert result.returncode == 0, result.stderr diff --git a/lib/project/tests/test_pipeline.py b/lib/project/tests/test_pipeline.py new file mode 100644 index 0000000..0fa1c52 --- /dev/null +++ b/lib/project/tests/test_pipeline.py @@ -0,0 +1,273 @@ +import json + +import numpy as np +import pytest +import soundfile as sf + +from voiceforge import pipeline +from voiceforge.audio import loudness, measure +from voiceforge.config import resolve + + +@pytest.mark.parametrize("kind", ["missing", "input-extension", "output-extension"]) +def test_invalid_paths_fail_before_tool_setup(wav, tmp_path, bypass, progress, monkeypatch, kind): + source = wav(name="input.txt" if kind == "input-extension" else "input.wav") + if kind == "missing": + source = tmp_path / "missing.wav" + target = tmp_path / ("master.mp3" if kind == "output-extension" else "master.wav") + monkeypatch.setattr(pipeline, "ensure_ffmpeg", lambda: pytest.fail("Invalid paths must fail preflight")) + with pytest.raises(ValueError, match="WAV file|.wav extension"): + pipeline.process(source, target, bypass, progress) + assert not target.exists() + + +@pytest.mark.parametrize("alias", ["same", "symlink", "hardlink"]) +@pytest.mark.parametrize("overwrite", [False, True]) +def test_original_is_never_overwritten(wav, tmp_path, bypass, progress, monkeypatch, alias, overwrite): + source = wav() + original = source.read_bytes() + target = source if alias == "same" else tmp_path / "alias.wav" + if alias == "symlink": + target.symlink_to(source) + elif alias == "hardlink": + target.hardlink_to(source) + monkeypatch.setattr(pipeline, "ensure_ffmpeg", lambda: pytest.fail("Preflight must reject the input")) + with pytest.raises(ValueError, match="original recording"): + pipeline.process(source, target, bypass, progress, overwrite=overwrite) + assert source.read_bytes() == original + + +@pytest.mark.parametrize("suffix", [".wav", ".report.json", ".mp3"]) +def test_existing_output_or_sidecar_rejected_before_processing(wav, tmp_path, bypass, progress, monkeypatch, suffix): + source = wav() + target = tmp_path / "master.wav" + existing = target.with_suffix(suffix) + existing.write_bytes(b"existing output") + bypass.mp3 = True + monkeypatch.setattr(pipeline, "ensure_ffmpeg", lambda: pytest.fail("Preflight must reject existing outputs")) + with pytest.raises(FileExistsError, match="Output exists"): + pipeline.process(source, target, bypass, progress) + assert existing.read_bytes() == b"existing output" + + +@pytest.mark.parametrize("kind,message", [ + ("empty", "empty"), ("silence", "silent or too quiet"), + ("quiet", "silent or too quiet"), ("surround", "Only mono or stereo"), + ("nonfinite", "non-finite"), ("stereo", "select --channel"), + ("mono-right", "right channel of a mono"), +]) +def test_unusable_inputs_leave_no_outputs(wav, tmp_path, bypass, progress, ffmpeg, kind, message): + samples = { + "empty": np.empty(0), "silence": np.zeros(48000), + "quiet": np.full(48000, 1e-6), "surround": np.ones((4800, 3)) * 0.1, + "nonfinite": np.array([0.1, np.nan]), "stereo": np.ones((4800, 2)) * 0.1, + "mono-right": np.ones(4800) * 0.1, + }[kind] + source = wav(samples) + target = tmp_path / "outputs" / "master.wav" + if kind == "mono-right": + bypass.channel = "right" + with pytest.raises(ValueError, match=message): + pipeline.process(source, target, bypass, progress) + assert not target.parent.exists() + + +def test_measure_reports_channels_peaks_clipping_and_progress(wav): + samples = np.tile([1.0, -0.25], (4800, 1)) + events = [] + result = measure(wav(samples), lambda *event: events.append(event)) + assert result["duration_seconds"] == 0.1 + assert result["channels"] == 2 + assert result["sample_peak_dbfs"] == 0 + assert result["channel_rms_dbfs"] == pytest.approx([0, 20 * np.log10(0.25)]) + assert result["dc_offset"] == pytest.approx([1, -0.25]) + assert result["near_full_scale_samples"] == 4800 + assert events[0][1:] == (0, 4800) + assert events[-1][1:] == (4800, 4800) + + +@pytest.mark.parametrize("channel", ["left", "right", "mix"]) +def test_explicit_stereo_selection_preserves_signal_and_duration(wav, tmp_path, bypass, progress, ffmpeg, channel): + time = np.arange(48000 * 3) / 48000 + left = 0.2 * np.sin(2 * np.pi * 440 * time) + right = 0.1 * np.sin(2 * np.pi * 880 * time) + source = wav(np.column_stack([left, right])) + bypass.channel = channel + target = tmp_path / "master.wav" + report = pipeline.process(source, target, bypass, progress) + actual, rate = sf.read(target) + expected = {"left": left, "right": right, "mix": (left + right) / 2}[channel] + assert actual.ndim == 1 + assert rate == 48000 + assert len(actual) == len(expected) + np.testing.assert_allclose(actual, expected, atol=5e-7, rtol=0) + assert report["normalization_mode"] == "disabled" + assert report["level_correction_db"] == 0 + assert sf.info(target).subtype == "PCM_24" + + +@pytest.mark.parametrize("rate", [44100, 48000]) +def test_normalization_peak_ceiling_and_report(wav, tmp_path, progress, ffmpeg, rate): + source = wav() + original = source.read_bytes() + target = tmp_path / "nested" / "master.wav" + settings = resolve(overrides={"denoiser": "none", "sample_rate": rate}) + report = pipeline.process(source, target, settings, progress) + stats = sf.info(target) + assert stats.channels == 1 + assert stats.samplerate == rate + assert stats.subtype == "PCM_24" + assert abs(stats.duration - 3) <= 0.02 + verified = loudness(ffmpeg, target, settings, stats.duration, progress) + assert verified["input_i"] == pytest.approx(settings.target_lufs, abs=0.5) + assert verified["input_tp"] <= settings.true_peak_db + 0.1 + assert report["master"]["near_full_scale_samples"] == 0 + assert report["master"]["sample_peak_dbfs"] <= settings.true_peak_db + 0.1 + assert json.loads(target.with_suffix(".report.json").read_text()) == report + assert source.read_bytes() == original + assert not list(target.parent.glob(".voiceforge-*")) + + +def test_limiter_without_normalization(wav, tmp_path, bypass, progress, ffmpeg): + time = np.arange(48000 * 3) / 48000 + source = wav(0.99 * np.sin(2 * np.pi * 440 * time)) + bypass.limiter = True + bypass.true_peak_db = -6 + report = pipeline.process(source, tmp_path / "master.wav", bypass, progress) + assert report["normalization_mode"] == "disabled" + assert report["master_loudness"]["input_tp"] <= -5.9 + assert report["master"]["sample_peak_dbfs"] > -7 + + +def test_speech_estimate_survives_pause_dominated_recordings(wav): + time = np.arange(48000 * 3) / 48000 + audio = np.concatenate([np.zeros(48000 * 27), 0.02 * np.sin(2 * np.pi * 440 * time)]) + stats = measure(wav(audio), lambda *_: None) + assert stats["quiet_blocks_dbfs"] <= -110 + assert stats["speech_level_estimate_dbfs"] == pytest.approx(-37, abs=2) + + +def test_leveling_uses_active_speech_not_pauses(wav, tmp_path, progress, ffmpeg): + time = np.arange(48000 * 3) / 48000 + audio = np.concatenate([np.zeros(48000 * 27), 0.02 * np.sin(2 * np.pi * 440 * time)]) + source = wav(audio) + settings = resolve(profile="cleanup-only", overrides={"denoiser": "none", "leveling": True}) + report = pipeline.process(source, tmp_path / "master.wav", settings, progress) + assert 10 < report["level_correction_db"] < settings.max_gain_db + assert not any("gain limit" in warning for warning in report["warnings"]) + + +def test_leveling_skipped_when_speech_estimate_is_unreliable(wav, tmp_path, bypass, progress, ffmpeg, monkeypatch): + real_measure = pipeline.measure + + def measured(path, callback, stage="Analyzing audio"): + stats = real_measure(path, callback, stage) + if stage == "Measuring cleaned voice": + stats["speech_level_estimate_dbfs"] = -120 + return stats + + monkeypatch.setattr(pipeline, "measure", measured) + bypass.leveling = True + report = pipeline.process(wav(), tmp_path / "master.wav", bypass, progress) + assert report["level_correction_db"] == 0 + assert any("leveling skipped" in warning for warning in report["warnings"]) + + +def test_antiphase_mix_is_rejected_after_selection(wav, tmp_path, bypass, progress, ffmpeg): + time = np.arange(48000 * 3) / 48000 + tone = 0.2 * np.sin(2 * np.pi * 440 * time) + source = wav(np.column_stack([tone, -tone])) + bypass.channel = "mix" + target = tmp_path / "master.wav" + with pytest.raises(ValueError, match="selected audio is silent or too quiet"): + pipeline.process(source, target, bypass, progress) + assert not target.exists() + assert not target.with_suffix(".report.json").exists() + assert not list(tmp_path.glob(".voiceforge-*")) + + +def test_silent_selected_channel_is_rejected_after_selection(wav, tmp_path, bypass, progress, ffmpeg): + time = np.arange(48000 * 3) / 48000 + tone = 0.2 * np.sin(2 * np.pi * 440 * time) + source = wav(np.column_stack([tone, np.zeros_like(tone)])) + bypass.channel = "right" + target = tmp_path / "master.wav" + with pytest.raises(ValueError, match="selected audio is silent or too quiet"): + pipeline.process(source, target, bypass, progress) + assert not target.exists() + + +def test_zero_denoise_strength_skips_ai(wav, tmp_path, bypass, progress, ffmpeg, monkeypatch): + def unexpected(*args, **kwargs): + pytest.fail("Zero denoise strength must not invoke AI") + + monkeypatch.setattr("voiceforge.ai.denoise", unexpected) + bypass.denoiser = "deepfilter" + bypass.denoise_strength = 0 + report = pipeline.process(wav(), tmp_path / "master.wav", bypass, progress) + assert report["cleaned"]["frames"] == report["input"]["frames"] + + +def test_fft_and_mp3_export_can_replace_generated_outputs(wav, tmp_path, bypass, progress, ffmpeg): + source = wav() + original = source.read_bytes() + bypass.denoiser = "fft" + bypass.mp3 = True + target = tmp_path / "master.wav" + for suffix in (".wav", ".report.json", ".mp3"): + target.with_suffix(suffix).write_bytes(b"old output") + events = [] + report = pipeline.process(source, target, bypass, lambda *event: events.append(event), overwrite=True) + assert any(event[0] == "FFT noise reduction" for event in events) + assert target.with_suffix(".mp3").stat().st_size > 1000 + assert report["mp3_loudness"]["input_i"] is not None + assert json.loads(target.with_suffix(".report.json").read_text()) == report + assert sf.info(target).channels == 1 + assert source.read_bytes() == original + + +def test_render_failure_preserves_existing_outputs_and_removes_temporary_files(wav, tmp_path, bypass, progress, ffmpeg, monkeypatch): + source = wav() + target = tmp_path / "master.wav" + bypass.mp3 = True + outputs = [target.with_suffix(suffix) for suffix in (".wav", ".report.json", ".mp3")] + for path in outputs: + path.write_bytes(b"previous publication") + real_run = pipeline.ffmpeg_run + + def fail_encoding(executable, args, stage, duration, callback): + if stage == "Encoding MP3": + raise RuntimeError("injected encoder failure") + return real_run(executable, args, stage, duration, callback) + + monkeypatch.setattr(pipeline, "ffmpeg_run", fail_encoding) + with pytest.raises(RuntimeError, match="injected encoder failure"): + pipeline.process(source, target, bypass, progress, overwrite=True) + assert all(path.read_bytes() == b"previous publication" for path in outputs) + assert not list(tmp_path.glob(".voiceforge-*")) + + +@pytest.mark.parametrize("overwrite", [False, True]) +def test_publish_collision_respects_overwrite(tmp_path, overwrite): + staged = tmp_path / "staged" + target = tmp_path / "target" + staged.write_bytes(b"new") + target.write_bytes(b"concurrent output") + if overwrite: + pipeline.publish(staged, target, overwrite=True) + assert target.read_bytes() == b"new" + assert not staged.exists() + else: + with pytest.raises(FileExistsError): + pipeline.publish(staged, target, overwrite=False) + assert target.read_bytes() == b"concurrent output" + assert staged.read_bytes() == b"new" + + +def test_publish_new_output_moves_staged_file(tmp_path): + staged = tmp_path / "staged" + target = tmp_path / "target" + staged.write_bytes(b"finished") + pipeline.publish(staged, target, overwrite=False) + assert target.read_bytes() == b"finished" + assert not staged.exists() |