diff options
| author | historia <historiavg@proton.me> | 2026-09-07 06:47:47 -0400 |
|---|---|---|
| committer | historia <historiavg@proton.me> | 2026-09-07 06:47:47 -0400 |
| commit | 84dd2d068317998f6fb59400c534ef5be6b51b53 (patch) | |
| tree | 025293e9d9229e02960374771ae522d9de2628ce /lib/tests/test_cli.py | |
| parent | 39b0f2bbed74f6487a41b82501ae3c6799e4b5c4 (diff) | |
| download | producer-main.tar.gz | |
Diffstat (limited to 'lib/tests/test_cli.py')
| -rw-r--r-- | lib/tests/test_cli.py | 357 |
1 files changed, 0 insertions, 357 deletions
diff --git a/lib/tests/test_cli.py b/lib/tests/test_cli.py deleted file mode 100644 index 76952ba..0000000 --- a/lib/tests/test_cli.py +++ /dev/null @@ -1,357 +0,0 @@ -import sys -from types import SimpleNamespace - -import numpy as np -import pytest -import soundfile as sf -from conftest import speechish - -from producer import io as pio -from producer.cli import _apply_args, build_parser, process_one -from producer.config import Options - - -def _mk_wav(tmp_path, name="in.wav", stereo=False): - x = speechish(3.0, level_dbfs=-30.0) - if stereo: - data = np.stack([x, x * 0.5], axis=1) - sf.write(str(tmp_path / name), data, 44100, subtype="PCM_16") - else: - sf.write(str(tmp_path / name), x, 44100, subtype="PCM_16") - return tmp_path / name - - -def _opts(**kw): - opts = Options() - opts.denoise = "off" - opts.enhance = "off" - for k, v in kw.items(): - setattr(opts, k, v) - return opts - - -def test_decode_stereo_mixdown(tmp_path): - p = _mk_wav(tmp_path, "st.wav", stereo=True) - x, sr = pio.decode(p) - assert sr == 44100 - assert x.dtype.name == "float32" - assert x.ndim == 1 - - -def test_encode_wav_bitdepths(tmp_path): - x = speechish(2.0, level_dbfs=-20.0) - for depth in (16, 24, 32): - out = tmp_path / f"o{depth}.wav" - pio.encode(x, 44100, out, "wav", depth) - y, sr = pio.decode(out) - assert sr == 44100 - assert ( - abs( - float(np.sqrt(np.mean(y.astype(np.float64) ** 2))) - - float(np.sqrt(np.mean(x.astype(np.float64) ** 2))) - ) - < 1e-3 - ) - - -def test_encode_flac(tmp_path): - x = speechish(2.0, level_dbfs=-20.0) - out = tmp_path / "o.flac" - pio.encode(x, 44100, out, "flac", 24) - y, sr = pio.decode(out) - assert sr == 44100 - assert np.corrcoef(x, y)[0, 1] > 0.999 - - -def test_mp3_roundtrip(tmp_path): - import pytest as _pt - - if not pio.ffmpeg_available(): - _pt.skip("ffmpeg missing") - x = speechish(4.0, level_dbfs=-20.0) - out = tmp_path / "o.mp3" - pio.encode(x, 44100, out, "mp3", 16) - y, sr = pio.decode(out) - assert sr == 44100 - from producer import meters - - assert abs(meters.rms_db(x) - meters.rms_db(y)) < 0.7 - - -def test_decode_via_ffmpeg_fallback(tmp_path): - import pytest as _pt - - if not pio.ffmpeg_available(): - _pt.skip("ffmpeg missing") - x = speechish(4.0, level_dbfs=-20.0) - out = tmp_path / "o.mp3" - pio.encode(x, 44100, out, "mp3", 16) - y, sr = pio.decode(out) - assert sr == 44100 - assert y.size > 0 - - -def test_process_one_end_to_end(tmp_path, capsys): - import json - - from producer import meters - - inp = _mk_wav(tmp_path, "e2e.wav") - out = tmp_path / "e2e_processed.wav" - rc = process_one(inp, _opts(report=True), single=True) - assert rc == out - assert out.exists() - rep_path = out.with_name("e2e_processed.report.json") - assert rep_path.exists() - rep = json.loads(rep_path.read_text()) - y, sr = pio.decode(out) - assert abs(meters.rms_db(y) + 20.0) < 0.6 - assert meters.true_peak_db(y, sr) <= -2.9 - assert rep["after"]["rms_db"] != 0 - - -def test_dry_run_listing(tmp_path, capsys): - from producer.cli import main - - inp = _mk_wav(tmp_path, "dry.wav") - rc = main([str(inp), "--dry-run", "--denoise", "off"]) - assert rc == 0 - out = capsys.readouterr().out - assert "denoise" in out and "dsp" in out and "levelling" in out - - -def test_arg_parsing_precedence(): - parser = build_parser() - args = parser.parse_args( - ["in.wav", "--profile", "podcast", "--warmth", "0.1", "--ceiling", "-2.0"] - ) - opts = Options() - _apply_args(opts, args) - assert opts.profile == "podcast" - assert opts.loudness_mode() == "lufs" - assert opts.strengths["warmth"] == 0.1 - assert opts.ceiling() == -2.0 - assert opts.strengths["air"] is None - - -def test_radio_profile_has_tuned_defaults(): - from producer import pipeline - - opts = Options(profile="radio") - assert opts.eff("tape") > 0 and opts.eff("soothe") > 0 - assert opts.loudness_mode() == "lufs" - stages = pipeline.build_stages(opts) - assert [st.name for st in stages] == ["denoise", "enhance", "dsp", "levelling"] - assert opts.denoise_strength is not None - - -def test_tape_and_soothe_flags_override(): - args = build_parser().parse_args( - ["in.wav", "--profile", "radio", "--tape", "0.4", "--soothe", "0.7"] - ) - opts = Options() - _apply_args(opts, args) - assert opts.profile == "radio" - assert opts.strengths["tape"] == 0.4 - assert opts.strengths["soothe"] == 0.7 - - -def test_default_output_suffix_processed(tmp_path): - from producer.cli import _resolve_output - - inp = tmp_path / "song.wav" - assert _resolve_output(inp, _opts(), single=True) == tmp_path / "song_processed.wav" - odir = tmp_path / "out" - opts = _opts(output=str(odir)) - assert _resolve_output(inp, opts, single=False) == odir / "song_processed.wav" - - -def test_engine_chunk_flags(): - args = build_parser().parse_args(["in.wav", "--engine-chunk", "15", "--engine-overlap", "1.0"]) - opts = Options() - _apply_args(opts, args) - assert opts.engine_chunk_s == 15.0 - assert opts.engine_overlap_s == 1.0 - - -def test_default_options_whole_file_and_soft_denoise(): - opts = Options() - assert opts.engine_chunk_s == 0.0 - assert opts.denoise_strength == 0.9 - - -def test_engine_chunk_validation(): - with pytest.raises(SystemExit): - args = build_parser().parse_args(["in.wav", "--engine-chunk", "5", "--engine-overlap", "5"]) - _apply_args(Options(), args) - with pytest.raises(SystemExit): - args = build_parser().parse_args(["in.wav", "--engine-chunk", "-1"]) - _apply_args(Options(), args) - - -def test_engine_chunk_config_override(tmp_path): - from producer import config as cfgmod - - cfg = tmp_path / "config.toml" - cfg.write_text("engine_chunk = 10.0\nengine_overlap = 1.0\n") - opts = Options() - cfgmod.apply_config(opts, cfgmod.load_config(cfg)) - assert opts.engine_chunk_s == 10.0 - assert opts.engine_overlap_s == 1.0 - - -def test_denoise_pf_flag_and_config_plumbing(tmp_path): - from producer import config as cfgmod - - opts = Options() - _apply_args(opts, build_parser().parse_args(["in.wav"])) - assert opts.denoise_pf is False - - opts = Options() - _apply_args(opts, build_parser().parse_args(["in.wav", "--denoise-pf"])) - assert opts.denoise_pf is True - - cfg = tmp_path / "config.toml" - cfg.write_text('[denoise]\nengine = "dfn3"\nstrength = 0.8\npf = true\n') - opts = Options() - cfgmod.apply_config(opts, cfgmod.load_config(cfg)) - assert opts.denoise_strength == 0.8 - assert opts.denoise_pf is True - - -def test_spectral_denoise_choice(tmp_path): - from producer import config as cfgmod - - opts = Options() - _apply_args(opts, build_parser().parse_args(["in.wav", "--denoise", "spectral"])) - assert opts.denoise == "spectral" - - cfg = tmp_path / "config.toml" - cfg.write_text('[denoise]\nengine = "spectral"\nstrength = 0.8\n') - opts = Options() - cfgmod.apply_config(opts, cfgmod.load_config(cfg)) - assert opts.denoise == "spectral" - assert opts.denoise_strength == 0.8 - - -def test_conflict_auto_renames_when_not_a_tty(tmp_path): - inp = _mk_wav(tmp_path, "in.wav") - first = process_one(inp, _opts(), single=True) - assert first == tmp_path / "in_processed.wav" - assert process_one(inp, _opts(), single=True) == tmp_path / "in_processed_1.wav" - assert process_one(inp, _opts(), single=True) == tmp_path / "in_processed_2.wav" - assert (tmp_path / "in_processed.wav").exists() - - -def test_conflict_prompt_overwrite(tmp_path, monkeypatch): - inp = _mk_wav(tmp_path, "in.wav") - out = tmp_path / "in_processed.wav" - process_one(inp, _opts(), single=True) - monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True)) - monkeypatch.setattr("builtins.input", lambda _prompt: "o") - assert process_one(inp, _opts(), single=True) == out - assert out.exists() - - -def test_conflict_prompt_rename(tmp_path, monkeypatch): - inp = _mk_wav(tmp_path, "in.wav") - out = tmp_path / "in_processed.wav" - process_one(inp, _opts(), single=True) - monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True)) - monkeypatch.setattr("builtins.input", lambda _prompt: "r") - second = process_one(inp, _opts(), single=True) - assert second == tmp_path / "in_processed_1.wav" - assert second.exists() - assert out.exists() - - -def test_conflict_prompt_invalid_then_overwrite(tmp_path, monkeypatch): - inp = _mk_wav(tmp_path, "in.wav") - out = tmp_path / "in_processed.wav" - process_one(inp, _opts(), single=True) - monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True)) - answers = iter(["maybe", "O"]) - monkeypatch.setattr("builtins.input", lambda _prompt: next(answers)) - assert process_one(inp, _opts(), single=True) == out - - -def test_conflict_prompt_cancel(tmp_path, monkeypatch, capsys): - inp = _mk_wav(tmp_path, "in.wav") - out = tmp_path / "in_processed.wav" - process_one(inp, _opts(), single=True) - before = out.read_bytes() - monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True)) - monkeypatch.setattr("builtins.input", lambda _prompt: "c") - assert process_one(inp, _opts(), single=True) is None - assert out.read_bytes() == before - assert "skipped" in capsys.readouterr().out - - -def test_conflict_prompt_eof_cancels(tmp_path, monkeypatch): - inp = _mk_wav(tmp_path, "in.wav") - process_one(inp, _opts(), single=True) - monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True)) - - def _eof(_prompt): - raise EOFError - - monkeypatch.setattr("builtins.input", _eof) - assert process_one(inp, _opts(), single=True) is None - - -def test_process_one_reports_stage_status(tmp_path, capsys): - inp = _mk_wav(tmp_path, "status.wav") - out_path = process_one(inp, _opts(), single=True) - assert out_path is not None - out = capsys.readouterr().out - assert f"[producer] {inp} (" in out - assert "dsp done in" in out - assert "levelling done in" in out - assert f"[producer] wrote {out_path}" in out - - -def test_multi_file_run_gets_position_prefixes(tmp_path, capsys): - from producer.cli import main - - a = _mk_wav(tmp_path, "a.wav") - b = _mk_wav(tmp_path, "b.wav") - rc = main([str(a), str(b), "--denoise", "off"]) - assert rc == 0 - out = capsys.readouterr().out - assert "[1/2]" in out and "[2/2]" in out - assert (tmp_path / "a_processed.wav").exists() - assert (tmp_path / "b_processed.wav").exists() - - -def test_output_file_rejected_for_multiple_inputs(tmp_path): - from producer.cli import main - - a = _mk_wav(tmp_path, "a.wav") - b = _mk_wav(tmp_path, "b.wav") - with pytest.raises(SystemExit): - main([str(a), str(b), "-o", str(tmp_path / "out.wav"), "--denoise", "off"]) - - -def test_output_dir_accepted_for_multiple_inputs(tmp_path): - from producer.cli import main - - a = _mk_wav(tmp_path, "a.wav") - b = _mk_wav(tmp_path, "b.wav") - outdir = tmp_path / "masters" - rc = main([str(a), str(b), "-o", str(outdir), "--denoise", "off"]) - assert rc == 0 - assert (outdir / "a_processed.wav").exists() - assert (outdir / "b_processed.wav").exists() - - -def test_batch_failure_continues_to_next_file(tmp_path, capsys): - from producer.cli import main - - good = _mk_wav(tmp_path, "good.wav") - missing = tmp_path / "missing.wav" - rc = main([str(missing), str(good), "--denoise", "off"]) - assert rc == 1 - captured = capsys.readouterr() - assert "[2/2]" in captured.out - assert "[producer] ERROR" in captured.err - assert (tmp_path / "good_processed.wav").exists() |
