feat: configure [backend] loads existing backend config rather than just overwriting it

1 files changed, 296 insertions, 0 deletions

diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py
index 563ed78..3d042db 100644
--- a/app/tests/test_backends_audiocpp.py
+++ b/app/tests/test_backends_audiocpp.py
@@ -1270,5 +1270,301 @@ class DetectServerSpecTests(unittest.TestCase):
                             for line in status.details))
 
 
+class InstalledModelEntriesTests(unittest.TestCase):
+    """installed_model_entries: the complement of missing_model_entries."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.dir = Path(self._tmp.name)
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def _server_json(self, models):
+        path = self.dir / "server.json"
+        path.write_text(json.dumps({"models": models}), encoding="utf-8")
+        return path
+
+    def test_lists_entries_whose_files_are_on_disk(self):
+        (self.dir / "models" / "present").mkdir(parents=True)
+        (self.dir / "models" / "present" / "m.gguf").write_bytes(b"x")
+        path = self._server_json([
+            {"id": "a", "path": "models/present"},
+            {"id": "b", "path": "models/absent"},
+        ])
+        installed = make_server.installed_model_entries(path)
+        self.assertEqual([m["id"] for m in installed], ["a"])
+
+    def test_unreadable_json_returns_empty(self):
+        path = self.dir / "server.json"
+        path.write_text("not json", encoding="utf-8")
+        self.assertEqual(make_server.installed_model_entries(path), [])
+
+
+class MissingModelInstallGuidanceTests(unittest.TestCase):
+    """missing_model_install_guidance: missing paths -> (id, install_id)."""
+
+    def test_maps_paths_and_skips_unmapped(self):
+        with tempfile.TemporaryDirectory() as td:
+            checkout = Path(td)
+            specs = checkout / "model_specs"
+            specs.mkdir()
+            (specs / "qwen3_tts.json").write_text(json.dumps({
+                "family": "qwen3_tts", "category": "tts",
+                "tasks": ["tts"],
+                "packages": [{
+                    "id": "qwen3_tts_0_6b_base_q8_0", "format": "gguf",
+                    "target_directory": "Qwen3-TTS-12Hz-0.6B-Base-GGUF",
+                }],
+            }), encoding="utf-8")
+            missing = [
+                {"id": "qwen", "rel": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"},
+                {"id": "x", "rel": "models/nope"},
+            ]
+            guidance = make_server.missing_model_install_guidance(
+                checkout, missing)
+        self.assertEqual(guidance,
+                         [("qwen", "qwen3_tts_0_6b_base_q8_0")])
+
+
+class LoadServerConfigTests(unittest.TestCase):
+    """load_server_config: read server.json, or None when unusable."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.dir = Path(self._tmp.name)
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_reads_dict_document(self):
+        path = self.dir / "server.json"
+        path.write_text(json.dumps({"host": "0.0.0.0", "models": []}),
+                        encoding="utf-8")
+        self.assertEqual(make_server.load_server_config(path),
+                         {"host": "0.0.0.0", "models": []})
+
+    def test_missing_file_returns_none(self):
+        self.assertIsNone(make_server.load_server_config(
+            self.dir / "nope.json"))
+
+    def test_unreadable_json_returns_none(self):
+        path = self.dir / "server.json"
+        path.write_text("not json", encoding="utf-8")
+        self.assertIsNone(make_server.load_server_config(path))
+
+    def test_non_dict_document_returns_none(self):
+        path = self.dir / "server.json"
+        path.write_text("[1, 2, 3]", encoding="utf-8")
+        self.assertIsNone(make_server.load_server_config(path))
+
+
+class ServerConfigSelectionsTests(unittest.TestCase):
+    """server_config_selections: map server.json models back to the catalog."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.checkout = _make_checkout(Path(self._tmp.name))
+        self.catalog = make_server.load_model_catalog(self.checkout)
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_maps_paths_to_family_dirs_and_tasks(self):
+        config = {"models": [
+            {"id": "qwen", "family": "qwen3_tts",
+             "path": "models/Qwen3-TTS-12Hz-1.7B-Base-GGUF", "task": "tts"},
+            {"id": "qwen-design", "family": "qwen3_tts",
+             "path": "models/Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF",
+             "task": "vdes"},
+            {"id": "higgs", "family": "higgs_audio_tts",
+             "path": "models/Higgs-Audio-v3-TTS-4B-GGUF", "task": "tts"},
+        ]}
+        selected, tasks = make_server.server_config_selections(config,
+                                                               self.catalog)
+        self.assertEqual(selected["qwen3_tts"],
+                         ["Qwen3-TTS-12Hz-1.7B-Base-GGUF",
+                          "Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF"])
+        self.assertEqual(selected["higgs_audio_tts"],
+                         ["Higgs-Audio-v3-TTS-4B-GGUF"])
+        self.assertEqual(tasks[("qwen3_tts",
+                                "Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF")],
+                         "vdes")
+        self.assertEqual(tasks[("qwen3_tts",
+                                "Qwen3-TTS-12Hz-1.7B-Base-GGUF")], "tts")
+
+    def test_unknown_family_ignored(self):
+        config = {"models": [
+            {"id": "x", "family": "not_a_family", "path": "models/x"},
+        ]}
+        selected, tasks = make_server.server_config_selections(config,
+                                                               self.catalog)
+        self.assertEqual(selected, {})
+        self.assertEqual(tasks, {})
+
+    def test_absolute_and_unprefixed_paths_kept_as_targets(self):
+        config = {"models": [
+            {"id": "qwen", "family": "qwen3_tts",
+             "path": "/abs/Qwen3-TTS-12Hz-1.7B-Base-GGUF", "task": "tts"},
+        ]}
+        selected, tasks = make_server.server_config_selections(config,
+                                                               self.catalog)
+        self.assertEqual(selected["qwen3_tts"],
+                         ["/abs/Qwen3-TTS-12Hz-1.7B-Base-GGUF"])
+
+    def test_empty_models_yield_empty_selections(self):
+        selected, tasks = make_server.server_config_selections({"models": []},
+                                                               self.catalog)
+        self.assertEqual(selected, {})
+        self.assertEqual(tasks, {})
+
+
+class UnusedInstalledEntriesTests(unittest.TestCase):
+    """unused_installed_entries: installed models dropped by a new selection."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.dir = Path(self._tmp.name)
+        (self.dir / "models" / "kept").mkdir(parents=True)
+        (self.dir / "models" / "kept" / "m.gguf").write_bytes(b"x")
+        (self.dir / "models" / "dropped").mkdir()
+        (self.dir / "models" / "dropped" / "m.gguf").write_bytes(b"x")
+        (self.dir / "models" / "missing").mkdir()  # empty: not installed
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def _server_json(self, models):
+        path = self.dir / "server.json"
+        path.write_text(json.dumps({"models": models}), encoding="utf-8")
+        return path
+
+    def test_returns_installed_entries_not_in_new_paths(self):
+        path = self._server_json([
+            {"id": "kept", "path": "models/kept"},
+            {"id": "dropped", "path": "models/dropped"},
+            {"id": "missing", "path": "models/missing"},
+        ])
+        unused = make_server.unused_installed_entries(
+            path, {"models/kept"})
+        self.assertEqual([entry["id"] for entry in unused], ["dropped"])
+
+    def test_nothing_unused_when_all_kept(self):
+        path = self._server_json([
+            {"id": "kept", "path": "models/kept"},
+        ])
+        unused = make_server.unused_installed_entries(
+            path, {"models/kept"})
+        self.assertEqual(unused, [])
+
+
+class DeleteModelFilesTests(unittest.TestCase):
+    """delete_model_files: remove on-disk model files for {id, rel} entries."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.dir = Path(self._tmp.name)
+        (self.dir / "models" / "a").mkdir(parents=True)
+        (self.dir / "models" / "a" / "m.gguf").write_bytes(b"x")
+        (self.dir / "models" / "b").mkdir()
+        (self.dir / "models" / "b" / "m.gguf").write_bytes(b"x")
+        (self.dir / "models" / "c").mkdir(parents=True)
+        self.server_json = self.dir / "server.json"
+        self.server_json.write_text(json.dumps({
+            "models": [
+                {"id": "a", "path": "models/a"},
+                {"id": "b", "path": "models/b"},
+                {"id": "c", "path": "models/c"},
+            ],
+        }), encoding="utf-8")
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_removes_dirs_and_counts(self):
+        removed = make_server.delete_model_files(
+            self.server_json,
+            [{"id": "a", "rel": "models/a"}, {"id": "b", "rel": "models/b"}])
+        self.assertEqual(removed, 2)
+        self.assertFalse((self.dir / "models" / "a").exists())
+        self.assertFalse((self.dir / "models" / "b").exists())
+        self.assertTrue((self.dir / "models" / "c").exists())
+
+    def test_missing_paths_ignored(self):
+        removed = make_server.delete_model_files(
+            self.server_json, [{"id": "ghost", "rel": "models/ghost"}])
+        self.assertEqual(removed, 0)
+
+    def test_removes_single_file(self):
+        file_path = self.dir / "models" / "single.gguf"
+        file_path.write_bytes(b"x")
+        removed = make_server.delete_model_files(
+            self.server_json, [{"id": "s", "rel": "models/single.gguf"}])
+        self.assertEqual(removed, 1)
+        self.assertFalse(file_path.exists())
+
+    def test_absolute_rel_path_honored(self):
+        target = self.dir / "absolute"
+        target.mkdir()
+        (target / "m.gguf").write_bytes(b"x")
+        removed = make_server.delete_model_files(
+            self.server_json, [{"id": "a", "rel": str(target)}])
+        self.assertEqual(removed, 1)
+        self.assertFalse(target.exists())
+
+
+class InstallModelsTests(unittest.TestCase):
+    """install_models: runs the install helper with download=True."""
+
+    def test_downloads_delegating_to_install_models(self):
+        with tempfile.TemporaryDirectory() as td:
+            checkout = Path(td)
+            guidance = [("qwen", "qwen3_tts_0_6b_base_q8_0")]
+            with patch.object(make_server, "_install_models") as mk:
+                make_server.install_models(checkout, guidance)
+            mk.assert_called_once_with(checkout, guidance, download=True)
+
+
+class HandInstallGuidanceTests(unittest.TestCase):
+    """hand_install_guidance: explains how to install models by hand."""
+
+    def test_lists_each_model_and_its_path(self):
+        with tempfile.TemporaryDirectory() as td:
+            checkout = Path(td)
+            message = make_server.hand_install_guidance(checkout, [
+                {"id": "qwen", "rel": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"},
+                {"id": "higgs", "rel": "models/Higgs-Audio-4B-GGUF"},
+            ])
+        self.assertIn("qwen", message)
+        self.assertIn("models/Qwen3-TTS-12Hz-0.6B-Base-GGUF", message)
+        self.assertIn("higgs", message)
+        self.assertIn("models/Higgs-Audio-4B-GGUF", message)
+        self.assertIn("download", message.lower())
+
+
+class UninstallTests(unittest.TestCase):
+    """uninstall: stop the server and remove the checkout."""
+
+    def test_removes_checkout(self):
+        with tempfile.TemporaryDirectory() as td:
+            checkout = Path(td) / "audio.cpp"
+            checkout.mkdir()
+            with patch.object(make_server, "find_local_checkout",
+                              return_value=checkout), \
+                    patch.object(make_server.servers, "stop") as mk_stop:
+                rc = make_server.uninstall()
+        self.assertEqual(rc, 0)
+        self.assertFalse(checkout.exists())
+        mk_stop.assert_called_once_with("audiocpp")
+
+    def test_no_checkout_is_a_noop(self):
+        with patch.object(make_server, "find_local_checkout",
+                          return_value=None), \
+                patch.object(make_server.servers, "stop") as mk_stop:
+            rc = make_server.uninstall()
+        self.assertEqual(rc, 0)
+        mk_stop.assert_called_once_with("audiocpp")
+
+
 if __name__ == "__main__":
     unittest.main()