path: root/tests/test_make_audiocpp_server_json.py

"""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()