diff options
Diffstat (limited to 'tests/test_make_audiocpp_server_json.py')
| -rw-r--r-- | tests/test_make_audiocpp_server_json.py | 457 |
1 files changed, 457 insertions, 0 deletions
diff --git a/tests/test_make_audiocpp_server_json.py b/tests/test_make_audiocpp_server_json.py new file mode 100644 index 0000000..63260ca --- /dev/null +++ b/tests/test_make_audiocpp_server_json.py @@ -0,0 +1,457 @@ +"""Tests for the audio.cpp server.json generator tool.""" + +import json +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from converter import config +from tools import make_audiocpp_server_json as make_server + +FAKE_CONFIG = ( + 'LANGUAGE = "English"\n' + "\n" + 'AUDIOCPP_API_URL = "http://127.0.0.1:9999" # audio.cpp audiocpp_server\n' + "\n" + "CHUNK_SIZE = 250\n" +) + + +class FindWavFilesTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def _touch(self, name): + path = self.folder / name + path.write_bytes(b"x") + return path + + def test_finds_only_wavs_case_insensitive(self): + self._touch("b.wav") + self._touch("a.WAV") + self._touch("notes.txt") + (self.folder / "sub").mkdir() + (self.folder / "sub" / "c.wav").write_bytes(b"x") + names = [path.name for path in make_server.find_wav_files(self.folder)] + self.assertEqual(names, ["a.WAV", "b.wav"]) + + def test_sorted_alphabetically_case_insensitive(self): + for name in ("Zed.wav", "alpha.wav", "Beta.wav"): + self._touch(name) + names = [path.name for path in make_server.find_wav_files(self.folder)] + self.assertEqual(names, ["alpha.wav", "Beta.wav", "Zed.wav"]) + + def test_empty_directory_returns_empty_list(self): + self.assertEqual(make_server.find_wav_files(self.folder), []) + + +class ConfigPortTests(unittest.TestCase): + def test_port_parsed_from_config_url(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:8080"): + self.assertEqual(make_server.config_port(), 8080) + + def test_missing_port_falls_back(self): + with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1"): + self.assertEqual(make_server.config_port(), + make_server.FALLBACK_PORT) + + def test_invalid_url_falls_back(self): + with patch.object(config, "AUDIOCPP_API_URL", "not a url"): + self.assertEqual(make_server.config_port(), + make_server.FALLBACK_PORT) + + def test_url_with_port_replaces_port(self): + self.assertEqual( + make_server._url_with_port("http://127.0.0.1:8080", 9000), + "http://127.0.0.1:9000") + + def test_url_without_port_adds_port(self): + self.assertEqual( + make_server._url_with_port("http://localhost", 8080), + "http://localhost:8080") + + +class UpdateConfigPortTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.config_path = Path(self._tmp.name) / "config.py" + self.config_path.write_text(FAKE_CONFIG, encoding="utf-8") + + def tearDown(self): + self._tmp.cleanup() + + def test_rewrites_port_preserving_comment(self): + changed = make_server.update_config_api_url_port( + 8080, config_path=self.config_path) + self.assertTrue(changed) + text = self.config_path.read_text(encoding="utf-8") + self.assertIn( + 'AUDIOCPP_API_URL = "http://127.0.0.1:8080" # audio.cpp audiocpp_server', + text) + self.assertIn('LANGUAGE = "English"', text) + self.assertIn("CHUNK_SIZE = 250", text) + + def test_returns_false_when_no_url_line(self): + path = Path(self._tmp.name) / "other.py" + path.write_text('CHUNK_SIZE = 250\n', encoding="utf-8") + self.assertFalse(make_server.update_config_api_url_port( + 8080, config_path=path)) + + def test_returns_false_when_port_unchanged(self): + self.assertFalse(make_server.update_config_api_url_port( + 9999, config_path=self.config_path)) + self.assertEqual(self.config_path.read_text(encoding="utf-8"), + FAKE_CONFIG) + + def test_returns_false_when_file_missing(self): + self.assertFalse(make_server.update_config_api_url_port( + 8080, config_path=Path(self._tmp.name) / "nope.py")) + + +class BuildVoicePresetsTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + self.narrator = self.folder / "narrator.wav" + self.narrator.write_bytes(b"x") + self.other = self.folder / "other.wav" + self.other.write_bytes(b"x") + + def tearDown(self): + self._tmp.cleanup() + + def test_presets_named_after_basenames_with_absolute_paths(self): + transcripts = {str(self.narrator): "First transcript.", + str(self.other): "Second transcript."} + with patch.object(make_server, "transcribe_reference_audio", + side_effect=lambda path, model_name="base": + transcripts[path]): + presets = make_server.build_voice_presets( + [self.narrator, self.other], "base") + self.assertEqual(list(presets), ["narrator", "other"]) + self.assertEqual(presets["narrator"]["reference_text"], + "First transcript.") + self.assertEqual(Path(presets["narrator"]["voice_ref"]), + self.narrator.resolve()) + + def test_failed_transcription_keeps_entry_with_empty_text(self): + with patch.object(make_server, "transcribe_reference_audio", + return_value=None): + presets = make_server.build_voice_presets([self.narrator], "base") + self.assertEqual(presets["narrator"]["reference_text"], "") + + def test_whisper_model_name_is_passed_through(self): + with patch.object(make_server, "transcribe_reference_audio", + return_value="text") as mock_transcribe: + make_server.build_voice_presets([self.narrator], "large-v3") + self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], + "large-v3") + + +class BuildServerConfigTests(unittest.TestCase): + def test_both_models_with_presets(self): + presets = {"narrator": {"voice_ref": "/x.wav", + "reference_text": "hi"}} + server_config = make_server.build_server_config( + host="127.0.0.1", port=8080, backend="cuda", lazy_load=False, + include_custom=True, include_clone=True, + custom_voice_id="qwen", clone_model_id="qwen-clone", + custom_voice_path="models/custom", base_path="models/base", + voice_presets=presets) + self.assertEqual(server_config["host"], "127.0.0.1") + self.assertEqual(server_config["port"], 8080) + self.assertEqual(server_config["backend"], "cuda") + self.assertFalse(server_config["lazy_load"]) + self.assertEqual([model["id"] for model in server_config["models"]], + ["qwen", "qwen-clone"]) + custom_entry, clone_entry = server_config["models"] + self.assertNotIn("voice_presets", custom_entry) + self.assertEqual(custom_entry["family"], "qwen3_tts") + self.assertEqual(custom_entry["path"], "models/custom") + self.assertEqual(clone_entry["path"], "models/base") + self.assertEqual(clone_entry["voice_presets"], presets) + + def test_custom_only_has_single_entry(self): + server_config = make_server.build_server_config( + host="0.0.0.0", port=9000, backend="cpu", lazy_load=True, + include_custom=True, include_clone=False, + custom_voice_id="qwen", clone_model_id="qwen-clone", + custom_voice_path="models/custom", base_path=None, + voice_presets={}) + self.assertEqual(len(server_config["models"]), 1) + self.assertEqual(server_config["models"][0]["id"], "qwen") + self.assertNotIn("voice_presets", server_config["models"][0]) + + def test_clone_only_without_presets_omits_key(self): + server_config = make_server.build_server_config( + host="127.0.0.1", port=8080, backend="vulkan", lazy_load=False, + include_custom=False, include_clone=True, + custom_voice_id="qwen", clone_model_id="qwen-clone", + custom_voice_path=None, base_path="models/base", + voice_presets={}) + self.assertEqual(len(server_config["models"]), 1) + self.assertEqual(server_config["models"][0]["id"], "qwen-clone") + self.assertNotIn("voice_presets", server_config["models"][0]) + + +class PromptHelperTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def test_ask_wav_dir_reprompts_until_valid(self): + with patch("builtins.input", + side_effect=[str(self.folder / "nope"), + str(self.folder)]): + self.assertEqual(make_server.ask_wav_dir(), self.folder) + + def test_ask_wav_dir_empty_skips(self): + with patch("builtins.input", return_value=""): + self.assertIsNone(make_server.ask_wav_dir()) + + def test_ask_wav_dir_eof_returns_none(self): + with patch("builtins.input", side_effect=EOFError): + self.assertIsNone(make_server.ask_wav_dir()) + + def test_ask_port_reprompts_until_valid(self): + with patch("builtins.input", side_effect=["abc", "8081"]): + self.assertEqual(make_server.ask_port(8080), 8081) + + def test_ask_port_eof_returns_default(self): + with patch("builtins.input", side_effect=EOFError): + self.assertEqual(make_server.ask_port(8080), 8080) + + def test_ask_menu_reprompts_until_valid(self): + options = [("One", "one"), ("Two", "two")] + with patch("builtins.input", side_effect=["9", "2"]): + self.assertEqual( + make_server.ask_menu("Pick:", options, default_index=1), + "two") + + def test_ask_menu_eof_returns_default(self): + options = [("One", "one"), ("Two", "two")] + with patch("builtins.input", side_effect=EOFError): + self.assertEqual( + make_server.ask_menu("Pick:", options, default_index=1), + "one") + + +class MainTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + self.output = self.folder / "server.json" + # Isolate the config.py rewrite target so no test can ever + # modify the repository's real converter/config.py. + self.fake_config = self.folder / "config.py" + self.fake_config.write_text(FAKE_CONFIG, encoding="utf-8") + patcher = patch.object(make_server, "CONFIG_PATH", self.fake_config) + patcher.start() + self.addCleanup(patcher.stop) + + def tearDown(self): + self._tmp.cleanup() + + def _run(self, argv, inputs=None, transcribe=None): + argv = ["make_audiocpp_server_json.py"] + argv + input_effect = inputs if inputs is not None else EOFError + transcribe_effect = transcribe if transcribe is not None else MagicMock() + with patch.object(sys, "argv", argv), \ + patch("builtins.input", side_effect=input_effect), \ + patch.object(make_server, "transcribe_reference_audio", + side_effect=transcribe_effect): + return make_server.main() + + def _defaults(self, models="", host="", port="", backend="", + lazy="", custom_path="", clone_path="", wav_dir="", + confirm="y", prefix=()): + return list(prefix) + [models, host, port, backend, lazy, + custom_path, clone_path, wav_dir, confirm] + + def test_default_run_hosts_both_models(self): + exit_code = self._run(["--output", str(self.output)], + inputs=self._defaults()) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["host"], "127.0.0.1") + self.assertEqual(data["port"], make_server.config_port()) + self.assertEqual(data["backend"], "cuda") + self.assertFalse(data["lazy_load"]) + self.assertEqual( + [model["id"] for model in data["models"]], + [config.AUDIOCPP_MODEL_ID, config.AUDIOCPP_CLONE_MODEL_ID]) + self.assertEqual( + [model["path"] for model in data["models"]], + [make_server.DEFAULT_CUSTOM_VOICE_PATH, + make_server.DEFAULT_BASE_PATH]) + self.assertNotIn("voice_presets", data["models"][1]) + + def test_eof_uses_all_defaults(self): + exit_code = self._run(["--output", str(self.output)]) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["host"], "127.0.0.1") + self.assertEqual(data["port"], make_server.config_port()) + self.assertEqual(data["backend"], "cuda") + self.assertFalse(data["lazy_load"]) + self.assertEqual(len(data["models"]), 2) + + def test_clone_only_with_positional_wav_dir(self): + (self.folder / "narrator.wav").write_bytes(b"x") + (self.folder / "alpha.wav").write_bytes(b"x") + inputs = ["3", "", "", "", "", "", "y"] + exit_code = self._run( + [str(self.folder), "--output", str(self.output)], + inputs=inputs, + transcribe=lambda path, model_name="base": + f"transcript of {Path(path).name}") + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 1) + clone_entry = data["models"][0] + self.assertEqual(clone_entry["id"], config.AUDIOCPP_CLONE_MODEL_ID) + self.assertEqual(sorted(clone_entry["voice_presets"]), + ["alpha", "narrator"]) + self.assertEqual(clone_entry["voice_presets"]["narrator"], + {"voice_ref": str((self.folder / "narrator.wav").resolve()), + "reference_text": "transcript of narrator.wav"}) + + def test_custom_only_single_model(self): + inputs = ["", "", "", "", "", "y"] + exit_code = self._run( + ["--output", str(self.output), "--models", "custom"], + inputs=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual([model["id"] for model in data["models"]], + [config.AUDIOCPP_MODEL_ID]) + + def test_duplicate_ids_prompt_for_distinct_clone_id(self): + with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen"), \ + patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen"): + inputs = ["1", "qwen-clone-2", "", "", "", "", "", "", "", "y"] + exit_code = self._run(["--output", str(self.output)], + inputs=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual([model["id"] for model in data["models"]], + ["qwen", "qwen-clone-2"]) + + def test_duplicate_ids_eof_exits(self): + with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen"), \ + patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen"): + with self.assertRaises(SystemExit) as ctx: + self._run(["--output", str(self.output)]) + self.assertNotEqual(ctx.exception.code, 0) + self.assertFalse(self.output.exists()) + + def test_port_sync_accepted_updates_config(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:9999"): + inputs = ["", "", "y", "", "", "", "", "", "y"] + exit_code = self._run(["--output", str(self.output), + "--port", "8080"], + inputs=inputs) + self.assertEqual(exit_code, 0) + self.assertIn('"http://127.0.0.1:8080"', + self.fake_config.read_text(encoding="utf-8")) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["port"], 8080) + + def test_port_sync_declined_keeps_config(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:9999"): + inputs = ["", "", "n", "", "", "", "", "", "y"] + exit_code = self._run(["--output", str(self.output), + "--port", "8080"], + inputs=inputs) + self.assertEqual(exit_code, 0) + self.assertIn('"http://127.0.0.1:9999"', + self.fake_config.read_text(encoding="utf-8")) + + def test_matching_port_does_not_prompt_for_sync(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:8080"): + inputs = self._defaults() + exit_code = self._run(["--output", str(self.output)], + inputs=inputs) + self.assertEqual(exit_code, 0) + self.assertEqual(self.fake_config.read_text(encoding="utf-8"), + FAKE_CONFIG) + + def test_invalid_menu_choice_reprompts(self): + inputs = ["9", "", "", "", "", "", "", "", "", "y"] + exit_code = self._run(["--output", str(self.output)], + inputs=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 2) + + def test_confirm_declined_writes_nothing(self): + inputs = self._defaults(confirm="n") + exit_code = self._run(["--output", str(self.output)], + inputs=inputs) + self.assertEqual(exit_code, 1) + self.assertFalse(self.output.exists()) + + def test_existing_output_declined_keeps_file(self): + self.output.write_text('{"old": true}', encoding="utf-8") + exit_code = self._run(["--output", str(self.output)], + inputs=["n"]) + self.assertEqual(exit_code, 1) + self.assertEqual(json.loads(self.output.read_text(encoding="utf-8")), + {"old": True}) + + def test_existing_output_accepted_overwrites(self): + self.output.write_text('{"old": true}', encoding="utf-8") + inputs = ["y"] + self._defaults() + exit_code = self._run(["--output", str(self.output)], + inputs=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 2) + + def test_force_overwrites_without_prompt(self): + self.output.write_text('{"old": true}', encoding="utf-8") + inputs = self._defaults() + exit_code = self._run(["--output", str(self.output), "--force"], + inputs=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 2) + + def test_flags_skip_prompts(self): + exit_code = self._run( + ["--output", str(self.output), "--models", "both", + "--host", "0.0.0.0", "--port", "9000", "--backend", "cpu", + "--lazy-load"], + inputs=["y", "", "", "", "y"]) + self.assertEqual(exit_code, 0) + self.assertIn('"http://127.0.0.1:9000"', + self.fake_config.read_text(encoding="utf-8")) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["host"], "0.0.0.0") + self.assertEqual(data["port"], 9000) + self.assertEqual(data["backend"], "cpu") + self.assertTrue(data["lazy_load"]) + + def test_missing_positional_wav_dir_errors(self): + with self.assertRaises(SystemExit) as ctx: + self._run([str(self.folder / "nope"), + "--output", str(self.output)], + inputs=self._defaults()) + self.assertEqual(ctx.exception.code, 2) + + +if __name__ == "__main__": + unittest.main() |