diff --git a/tests/services/test_service_registry.py b/tests/services/test_service_registry.py new file mode 100644 index 00000000..711be196 --- /dev/null +++ b/tests/services/test_service_registry.py @@ -0,0 +1,66 @@ +import sys +import types + +import pytest + +from py.services.service_registry import ServiceRegistry + + +@pytest.fixture(autouse=True) +def clear_service_registry(): + ServiceRegistry.clear_services() + yield + ServiceRegistry.clear_services() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "method_name,module_path,class_name,service_key", + [ + ("get_lora_scanner", "py.services.lora_scanner", "LoraScanner", "lora_scanner"), + ("get_checkpoint_scanner", "py.services.checkpoint_scanner", "CheckpointScanner", "checkpoint_scanner"), + ("get_recipe_scanner", "py.services.recipe_scanner", "RecipeScanner", "recipe_scanner"), + ], +) +async def test_lazy_loaded_scanners(monkeypatch, method_name, module_path, class_name, service_key): + calls = 0 + fake_instance = object() + + class FakeScanner: + @classmethod + async def get_instance(cls): + nonlocal calls + calls += 1 + return fake_instance + + module = types.ModuleType(module_path) + setattr(module, class_name, FakeScanner) + monkeypatch.setitem(sys.modules, module_path, module) + + method = getattr(ServiceRegistry, method_name) + + first = await method() + assert first is fake_instance + assert await ServiceRegistry.get_service(service_key) is fake_instance + + second = await method() + assert second is fake_instance + assert calls == 1 + + +@pytest.mark.asyncio +async def test_lazy_loaded_websocket_manager(monkeypatch): + fake_manager = object() + module = types.ModuleType("py.services.websocket_manager") + module.ws_manager = fake_manager + monkeypatch.setitem(sys.modules, "py.services.websocket_manager", module) + + first = await ServiceRegistry.get_websocket_manager() + assert first is fake_manager + + # Update registry to simulate external registration drift + sentinel = object() + ServiceRegistry._services["websocket_manager"] = sentinel + + second = await ServiceRegistry.get_websocket_manager() + assert second is sentinel diff --git a/tests/services/test_settings_manager.py b/tests/services/test_settings_manager.py index 3101b938..3b8ecf83 100644 --- a/tests/services/test_settings_manager.py +++ b/tests/services/test_settings_manager.py @@ -1,3 +1,4 @@ +import copy import json import os @@ -34,6 +35,32 @@ def test_environment_variable_overrides_settings(tmp_path, monkeypatch): assert mgr.get("civitai_api_key") == "secret" +def _create_manager_with_settings(tmp_path, monkeypatch, initial_settings, *, save_spy=None): + """Helper to instantiate SettingsManager with predefined settings.""" + + fake_settings_path = tmp_path / "settings.json" + + monkeypatch.setattr( + "py.services.settings_manager.ensure_settings_file", + lambda logger=None: str(fake_settings_path), + ) + + if save_spy is None: + monkeypatch.setattr(SettingsManager, "_save_settings", lambda self: None) + else: + monkeypatch.setattr(SettingsManager, "_save_settings", save_spy) + + monkeypatch.setattr( + SettingsManager, + "_load_settings", + lambda self: copy.deepcopy(initial_settings), + ) + + mgr = SettingsManager() + mgr.settings_file = str(fake_settings_path) + return mgr + + def test_download_path_template_parses_json_string(manager): templates = {"lora": "{author}", "checkpoint": "{author}", "embedding": "{author}"} manager.settings["download_path_templates"] = json.dumps(templates) @@ -98,6 +125,89 @@ def test_migrate_creates_default_library(manager): assert libraries["default"].get("folder_paths", {}) == manager.settings.get("folder_paths", {}) +def test_migrate_sanitizes_legacy_libraries(tmp_path, monkeypatch): + initial = { + "libraries": {"legacy": "not-a-dict"}, + "active_library": "legacy", + "folder_paths": {"loras": ["/old"]}, + } + + manager = _create_manager_with_settings(tmp_path, monkeypatch, initial) + + libraries = manager.get_libraries() + assert set(libraries.keys()) == {"legacy"} + payload = libraries["legacy"] + assert payload["folder_paths"] == {} + assert payload["default_lora_root"] == "" + assert payload["default_checkpoint_root"] == "" + assert payload["default_embedding_root"] == "" + assert manager.get_active_library_name() == "legacy" + + +def test_active_library_syncs_top_level_settings(tmp_path, monkeypatch): + initial = { + "libraries": { + "default": { + "folder_paths": {"loras": ["/loras"]}, + "default_lora_root": "/loras", + "default_checkpoint_root": "/ckpt", + "default_embedding_root": "/embed", + }, + "studio": { + "folder_paths": {"loras": ["/studio"]}, + "default_lora_root": "/studio", + "default_checkpoint_root": "/studio_ckpt", + "default_embedding_root": "/studio_embed", + }, + }, + "active_library": "studio", + # Drifted top-level values that should be corrected during init + "folder_paths": {"loras": ["/loras"]}, + "default_lora_root": "/loras", + "default_checkpoint_root": "/ckpt", + "default_embedding_root": "/embed", + } + + manager = _create_manager_with_settings(tmp_path, monkeypatch, initial) + + assert manager.get_active_library_name() == "studio" + assert manager.get("folder_paths")["loras"] == ["/studio"] + assert manager.get("default_lora_root") == "/studio" + assert manager.get("default_checkpoint_root") == "/studio_ckpt" + assert manager.get("default_embedding_root") == "/studio_embed" + + # Drift the top-level values again and ensure activate_library repairs them + manager.settings["folder_paths"] = {"loras": ["/loras"]} + manager.settings["default_lora_root"] = "/loras" + manager.activate_library("studio") + + assert manager.get("folder_paths")["loras"] == ["/studio"] + assert manager.get("default_lora_root") == "/studio" + + +def test_refresh_environment_variables_updates_stored_value(tmp_path, monkeypatch): + calls = [] + + def save_spy(self): + calls.append(self.settings.get("civitai_api_key")) + + initial = { + "civitai_api_key": "stale", + "libraries": {"default": {"folder_paths": {}, "default_lora_root": "", "default_checkpoint_root": "", "default_embedding_root": ""}}, + "active_library": "default", + } + + monkeypatch.setenv("CIVITAI_API_KEY", "from-init") + manager = _create_manager_with_settings(tmp_path, monkeypatch, initial, save_spy=save_spy) + + assert calls[-1] == "from-init" + + monkeypatch.setenv("CIVITAI_API_KEY", "refreshed") + manager.refresh_environment_variables() + + assert calls[-1] == "refreshed" + + def test_upsert_library_creates_entry_and_activates(manager, tmp_path): lora_dir = tmp_path / "loras" lora_dir.mkdir()