path: root/lib/project/tests/test_cli.py

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