From 55bc4004a5dfced8b4c53a88f735b4c2b64f6d48 Mon Sep 17 00:00:00 2001 From: historia Date: Sun, 6 Sep 2026 15:45:33 -0400 Subject: feat: output conflict handling --- lib/tests/test_cli.py | 82 +++++++++++++++++++++++++++++++++++++++++++++-- lib/tests/test_engines.py | 60 ++++++++++++++++++++++++++++++++++ 2 files changed, 140 insertions(+), 2 deletions(-) (limited to 'lib/tests') diff --git a/lib/tests/test_cli.py b/lib/tests/test_cli.py index e43489c..1a157f7 100644 --- a/lib/tests/test_cli.py +++ b/lib/tests/test_cli.py @@ -1,3 +1,6 @@ +import sys +from types import SimpleNamespace + import numpy as np import soundfile as sf from conftest import speechish @@ -93,11 +96,11 @@ def test_process_one_end_to_end(tmp_path, capsys): from producer import meters inp = _mk_wav(tmp_path, "e2e.wav") - out = tmp_path / "e2e_master.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_master.report.json") + 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) @@ -150,3 +153,78 @@ def test_tape_and_soothe_flags_override(): 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_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 diff --git a/lib/tests/test_engines.py b/lib/tests/test_engines.py index 0efccc6..781b46e 100644 --- a/lib/tests/test_engines.py +++ b/lib/tests/test_engines.py @@ -1,8 +1,68 @@ import os +import sys +import types +import warnings import numpy as np import pytest +AUDIO_META_MSG = ( + "`torchaudio.backend.common.AudioMetaData` has been moved to " + "`torchaudio.AudioMetaData`. Please update the import path." +) + + +def _stub_df_modules(calls: list[str]) -> tuple[types.ModuleType, list[types.ModuleType]]: + pkg = types.ModuleType("df") + mods = [] + for full in ("df.utils", "df.logger", "df.io"): + mod = types.ModuleType(full) + + def probe(name: str): + def fn(*_args): + calls.append(name) + return "deadbeef" + + return fn + + for fn_name in ("get_git_root", "get_commit_hash", "get_branch_name"): + setattr(mod, fn_name, probe(f"{full}.{fn_name}")) + setattr(pkg, full.split(".")[1], mod) + mods.append(mod) + return pkg, mods + + +def test_dfn_shim_neutralizes_git_probes(monkeypatch): + from producer.engines import denoise_dfn + + calls: list[str] = [] + pkg, mods = _stub_df_modules(calls) + for name, mod in zip(("df", "df.utils", "df.logger", "df.io"), (pkg, *mods), strict=True): + monkeypatch.setitem(sys.modules, name, mod) + assert pkg.logger.get_commit_hash() == "deadbeef" + calls.clear() + denoise_dfn._shim_df_git() + for mod in mods: + for fn_name in ("get_git_root", "get_commit_hash", "get_branch_name"): + assert mod.__dict__[fn_name]() is None + assert calls == [] + + +def test_dfn_shim_silences_torchaudio_warning(): + from producer.engines import denoise_dfn + + code = "import warnings\nwarnings.warn(MESSAGE, UserWarning)" + denoise_dfn._shim_torchaudio_backend() + with warnings.catch_warnings(record=True) as caught: + exec(compile(code, "df/io.py", "exec"), {"__name__": "df.io", "MESSAGE": AUDIO_META_MSG}) + assert caught == [] + with warnings.catch_warnings(record=True) as caught: + exec( + compile(code, "other/mod.py", "exec"), + {"__name__": "other.mod", "MESSAGE": AUDIO_META_MSG}, + ) + assert len(caught) == 1 + def _metrics_floor(x, sr): from producer import meters -- cgit v1.2.3