refactor(settings): lazily initialize manager

This commit is contained in:
pixelpaws
2025-10-08 08:56:57 +08:00
parent 3118f3b43c
commit c53fa8692b
31 changed files with 299 additions and 172 deletions

View File

@@ -73,6 +73,30 @@ nodes_mock.NODE_CLASS_MAPPINGS = {}
sys.modules['nodes'] = nodes_mock
@pytest.fixture(autouse=True)
def _isolate_settings_dir(tmp_path_factory, monkeypatch):
"""Redirect settings.json into a temporary directory for each test."""
settings_dir = tmp_path_factory.mktemp("settings_dir")
def fake_get_settings_dir(create: bool = True) -> str:
if create:
settings_dir.mkdir(exist_ok=True)
return str(settings_dir)
monkeypatch.setattr("py.utils.settings_paths.get_settings_dir", fake_get_settings_dir)
monkeypatch.setattr(
"py.utils.settings_paths.user_config_dir",
lambda *_args, **_kwargs: str(settings_dir),
)
from py.services import settings_manager as settings_manager_module
settings_manager_module.reset_settings_manager()
yield
settings_manager_module.reset_settings_manager()
def pytest_pyfunc_call(pyfuncitem):
"""Allow bare async tests to run without pytest.mark.asyncio."""
test_function = pyfuncitem.function

View File

@@ -8,7 +8,7 @@ import pytest
from py.services.download_manager import DownloadManager
from py.services import download_manager
from py.services.service_registry import ServiceRegistry
from py.services.settings_manager import settings
from py.services.settings_manager import SettingsManager, get_settings_manager
from py.utils.metadata_manager import MetadataManager
@@ -23,7 +23,8 @@ def reset_download_manager():
@pytest.fixture(autouse=True)
def isolate_settings(monkeypatch, tmp_path):
"""Point settings writes at a temporary directory to avoid touching real files."""
default_settings = settings._get_default_settings()
manager = get_settings_manager()
default_settings = manager._get_default_settings()
default_settings.update(
{
"default_lora_root": str(tmp_path),
@@ -37,8 +38,8 @@ def isolate_settings(monkeypatch, tmp_path):
"base_model_path_mappings": {"BaseModel": "MappedModel"},
}
)
monkeypatch.setattr(settings, "settings", default_settings)
monkeypatch.setattr(type(settings), "_save_settings", lambda self: None)
monkeypatch.setattr(manager, "settings", default_settings)
monkeypatch.setattr(SettingsManager, "_save_settings", lambda self: None)
@pytest.fixture(autouse=True)
@@ -187,7 +188,7 @@ async def test_successful_download_uses_defaults(monkeypatch, scanners, metadata
assert manager._active_downloads[result["download_id"]]["status"] == "completed"
assert captured["relative_path"] == "MappedModel/fantasy"
expected_dir = Path(settings.get("default_lora_root")) / "MappedModel" / "fantasy"
expected_dir = Path(get_settings_manager().get("default_lora_root")) / "MappedModel" / "fantasy"
assert captured["save_dir"] == expected_dir
assert captured["model_type"] == "lora"
assert captured["download_urls"] == [

View File

@@ -6,7 +6,7 @@ import pytest
from py.services.example_images_cleanup_service import ExampleImagesCleanupService
from py.services.service_registry import ServiceRegistry
from py.services.settings_manager import settings
from py.services.settings_manager import get_settings_manager
class StubScanner:
@@ -21,8 +21,9 @@ class StubScanner:
async def test_cleanup_moves_empty_and_orphaned(tmp_path, monkeypatch):
service = ExampleImagesCleanupService()
previous_path = settings.get('example_images_path')
settings.settings['example_images_path'] = str(tmp_path)
settings_manager = get_settings_manager()
previous_path = settings_manager.get('example_images_path')
settings_manager.settings['example_images_path'] = str(tmp_path)
try:
empty_folder = tmp_path / 'empty_folder'
@@ -64,23 +65,24 @@ async def test_cleanup_moves_empty_and_orphaned(tmp_path, monkeypatch):
finally:
if previous_path is None:
settings.settings.pop('example_images_path', None)
settings_manager.settings.pop('example_images_path', None)
else:
settings.settings['example_images_path'] = previous_path
settings_manager.settings['example_images_path'] = previous_path
@pytest.mark.asyncio
async def test_cleanup_handles_missing_path(monkeypatch):
service = ExampleImagesCleanupService()
previous_path = settings.get('example_images_path')
settings.settings.pop('example_images_path', None)
settings_manager = get_settings_manager()
previous_path = settings_manager.get('example_images_path')
settings_manager.settings.pop('example_images_path', None)
try:
result = await service.cleanup_example_image_folders()
finally:
if previous_path is not None:
settings.settings['example_images_path'] = previous_path
settings_manager.settings['example_images_path'] = previous_path
assert result['success'] is False
assert result['error_code'] == 'path_not_configured'

View File

@@ -7,7 +7,7 @@ from types import SimpleNamespace
import pytest
from py.services.settings_manager import settings
from py.services.settings_manager import SettingsManager, get_settings_manager
from py.utils import example_images_download_manager as download_module
@@ -43,11 +43,15 @@ def _patch_scanner(monkeypatch: pytest.MonkeyPatch, scanner: StubScanner) -> Non
@pytest.mark.usefixtures("tmp_path")
async def test_start_download_rejects_parallel_runs(monkeypatch: pytest.MonkeyPatch, tmp_path):
async def test_start_download_rejects_parallel_runs(
monkeypatch: pytest.MonkeyPatch,
tmp_path,
settings_manager,
):
ws_manager = RecordingWebSocketManager()
manager = download_module.DownloadManager(ws_manager=ws_manager)
monkeypatch.setitem(settings.settings, "example_images_path", str(tmp_path))
monkeypatch.setitem(settings_manager.settings, "example_images_path", str(tmp_path))
model = {
"sha256": "abc123",
@@ -106,11 +110,15 @@ async def test_start_download_rejects_parallel_runs(monkeypatch: pytest.MonkeyPa
@pytest.mark.usefixtures("tmp_path")
async def test_pause_resume_blocks_processing(monkeypatch: pytest.MonkeyPatch, tmp_path):
async def test_pause_resume_blocks_processing(
monkeypatch: pytest.MonkeyPatch,
tmp_path,
settings_manager,
):
ws_manager = RecordingWebSocketManager()
manager = download_module.DownloadManager(ws_manager=ws_manager)
monkeypatch.setitem(settings.settings, "example_images_path", str(tmp_path))
monkeypatch.setitem(settings_manager.settings, "example_images_path", str(tmp_path))
models = [
{
@@ -231,13 +239,17 @@ async def test_pause_resume_blocks_processing(monkeypatch: pytest.MonkeyPatch, t
@pytest.mark.usefixtures("tmp_path")
async def test_legacy_folder_migrated_and_skipped(monkeypatch: pytest.MonkeyPatch, tmp_path):
async def test_legacy_folder_migrated_and_skipped(
monkeypatch: pytest.MonkeyPatch,
tmp_path,
settings_manager,
):
ws_manager = RecordingWebSocketManager()
manager = download_module.DownloadManager(ws_manager=ws_manager)
monkeypatch.setitem(settings.settings, "example_images_path", str(tmp_path))
monkeypatch.setitem(settings.settings, "libraries", {"default": {}, "extra": {}})
monkeypatch.setitem(settings.settings, "active_library", "extra")
monkeypatch.setitem(settings_manager.settings, "example_images_path", str(tmp_path))
monkeypatch.setitem(settings_manager.settings, "libraries", {"default": {}, "extra": {}})
monkeypatch.setitem(settings_manager.settings, "active_library", "extra")
model_hash = "d" * 64
model_path = tmp_path / "model.safetensors"
@@ -310,13 +322,17 @@ async def test_legacy_folder_migrated_and_skipped(monkeypatch: pytest.MonkeyPatc
@pytest.mark.usefixtures("tmp_path")
async def test_legacy_progress_file_migrates(monkeypatch: pytest.MonkeyPatch, tmp_path):
async def test_legacy_progress_file_migrates(
monkeypatch: pytest.MonkeyPatch,
tmp_path,
settings_manager,
):
ws_manager = RecordingWebSocketManager()
manager = download_module.DownloadManager(ws_manager=ws_manager)
monkeypatch.setitem(settings.settings, "example_images_path", str(tmp_path))
monkeypatch.setitem(settings.settings, "libraries", {"default": {}, "extra": {}})
monkeypatch.setitem(settings.settings, "active_library", "extra")
monkeypatch.setitem(settings_manager.settings, "example_images_path", str(tmp_path))
monkeypatch.setitem(settings_manager.settings, "libraries", {"default": {}, "extra": {}})
monkeypatch.setitem(settings_manager.settings, "active_library", "extra")
model_hash = "e" * 64
model_path = tmp_path / "model-two.safetensors"
@@ -380,20 +396,24 @@ async def test_legacy_progress_file_migrates(monkeypatch: pytest.MonkeyPatch, tm
@pytest.mark.usefixtures("tmp_path")
async def test_download_remains_in_initial_library(monkeypatch: pytest.MonkeyPatch, tmp_path):
async def test_download_remains_in_initial_library(
monkeypatch: pytest.MonkeyPatch,
tmp_path,
settings_manager,
):
ws_manager = RecordingWebSocketManager()
manager = download_module.DownloadManager(ws_manager=ws_manager)
monkeypatch.setitem(settings.settings, "example_images_path", str(tmp_path))
monkeypatch.setitem(settings.settings, "libraries", {"LibraryA": {}, "LibraryB": {}})
monkeypatch.setitem(settings.settings, "active_library", "LibraryA")
monkeypatch.setitem(settings_manager.settings, "example_images_path", str(tmp_path))
monkeypatch.setitem(settings_manager.settings, "libraries", {"LibraryA": {}, "LibraryB": {}})
monkeypatch.setitem(settings_manager.settings, "active_library", "LibraryA")
state = {"active": "LibraryA"}
def fake_get_active_library_name(self):
return state["active"]
monkeypatch.setattr(type(settings), "get_active_library_name", fake_get_active_library_name)
monkeypatch.setattr(SettingsManager, "get_active_library_name", fake_get_active_library_name)
model_hash = "f" * 64
model_path = tmp_path / "example-model.safetensors"
@@ -454,3 +474,7 @@ async def test_download_remains_in_initial_library(monkeypatch: pytest.MonkeyPat
assert (model_dir / "local.txt").exists()
assert not (library_b_root / ".download_progress.json").exists()
assert not (library_b_root / model_hash).exists()
@pytest.fixture
def settings_manager():
return get_settings_manager()

View File

@@ -5,7 +5,7 @@ from typing import Any, Dict
import pytest
from py.services.settings_manager import settings
from py.services.settings_manager import get_settings_manager
from py.utils import example_images_download_manager as download_module
@@ -19,19 +19,21 @@ class RecordingWebSocketManager:
@pytest.fixture(autouse=True)
def restore_settings() -> None:
original = settings.settings.copy()
manager = get_settings_manager()
original = manager.settings.copy()
try:
yield
finally:
settings.settings.clear()
settings.settings.update(original)
manager.settings.clear()
manager.settings.update(original)
async def test_start_download_requires_configured_path(monkeypatch: pytest.MonkeyPatch) -> None:
manager = download_module.DownloadManager(ws_manager=RecordingWebSocketManager())
# Ensure example_images_path is not configured
settings.settings.pop('example_images_path', None)
settings_manager = get_settings_manager()
settings_manager.settings.pop('example_images_path', None)
with pytest.raises(download_module.DownloadConfigurationError) as exc_info:
await manager.start_download({})
@@ -44,9 +46,10 @@ async def test_start_download_requires_configured_path(monkeypatch: pytest.Monke
async def test_start_download_bootstraps_progress_and_task(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
settings.settings["example_images_path"] = str(tmp_path)
settings.settings["libraries"] = {"default": {}}
settings.settings["active_library"] = "default"
settings_manager = get_settings_manager()
settings_manager.settings["example_images_path"] = str(tmp_path)
settings_manager.settings["libraries"] = {"default": {}}
settings_manager.settings["active_library"] = "default"
ws_manager = RecordingWebSocketManager()
manager = download_module.DownloadManager(ws_manager=ws_manager)
@@ -84,9 +87,10 @@ async def test_start_download_bootstraps_progress_and_task(monkeypatch: pytest.M
async def test_pause_and_resume_flow(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
settings.settings["example_images_path"] = str(tmp_path)
settings.settings["libraries"] = {"default": {}}
settings.settings["active_library"] = "default"
settings_manager = get_settings_manager()
settings_manager.settings["example_images_path"] = str(tmp_path)
settings_manager.settings["libraries"] = {"default": {}}
settings_manager.settings["active_library"] = "default"
ws_manager = RecordingWebSocketManager()
manager = download_module.DownloadManager(ws_manager=ws_manager)

View File

@@ -7,7 +7,7 @@ from typing import Any, Dict
import pytest
from py.services.settings_manager import settings
from py.services.settings_manager import get_settings_manager
from py.utils.example_images_file_manager import ExampleImagesFileManager
@@ -22,16 +22,18 @@ class JsonRequest:
@pytest.fixture(autouse=True)
def restore_settings() -> None:
original = settings.settings.copy()
manager = get_settings_manager()
original = manager.settings.copy()
try:
yield
finally:
settings.settings.clear()
settings.settings.update(original)
manager.settings.clear()
manager.settings.update(original)
async def test_open_folder_requires_existing_model_directory(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
settings.settings["example_images_path"] = str(tmp_path)
settings_manager = get_settings_manager()
settings_manager.settings["example_images_path"] = str(tmp_path)
model_hash = "a" * 64
model_folder = tmp_path / model_hash
model_folder.mkdir()
@@ -65,7 +67,8 @@ async def test_open_folder_requires_existing_model_directory(monkeypatch: pytest
async def test_open_folder_rejects_invalid_paths(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
settings.settings["example_images_path"] = str(tmp_path)
settings_manager = get_settings_manager()
settings_manager.settings["example_images_path"] = str(tmp_path)
def fake_get_model_folder(_hash):
return str(tmp_path.parent / "outside")
@@ -81,7 +84,8 @@ async def test_open_folder_rejects_invalid_paths(monkeypatch: pytest.MonkeyPatch
async def test_get_files_lists_supported_media(tmp_path) -> None:
settings.settings["example_images_path"] = str(tmp_path)
settings_manager = get_settings_manager()
settings_manager.settings["example_images_path"] = str(tmp_path)
model_hash = "b" * 64
model_folder = tmp_path / model_hash
model_folder.mkdir()
@@ -99,7 +103,8 @@ async def test_get_files_lists_supported_media(tmp_path) -> None:
async def test_has_images_reports_presence(tmp_path) -> None:
settings.settings["example_images_path"] = str(tmp_path)
settings_manager = get_settings_manager()
settings_manager.settings["example_images_path"] = str(tmp_path)
model_hash = "c" * 64
model_folder = tmp_path / model_hash
model_folder.mkdir()

View File

@@ -6,7 +6,7 @@ from pathlib import Path
import pytest
from py.services.settings_manager import settings
from py.services.settings_manager import get_settings_manager
from py.utils.example_images_paths import (
ensure_library_root_exists,
get_model_folder,
@@ -18,18 +18,24 @@ from py.utils.example_images_paths import (
@pytest.fixture(autouse=True)
def restore_settings():
original = copy.deepcopy(settings.settings)
manager = get_settings_manager()
original = copy.deepcopy(manager.settings)
try:
yield
finally:
settings.settings.clear()
settings.settings.update(original)
manager.settings.clear()
manager.settings.update(original)
def test_get_model_folder_single_library(tmp_path):
settings.settings['example_images_path'] = str(tmp_path)
settings.settings['libraries'] = {'default': {}}
settings.settings['active_library'] = 'default'
@pytest.fixture
def settings_manager():
return get_settings_manager()
def test_get_model_folder_single_library(tmp_path, settings_manager):
settings_manager.settings['example_images_path'] = str(tmp_path)
settings_manager.settings['libraries'] = {'default': {}}
settings_manager.settings['active_library'] = 'default'
model_hash = 'a' * 64
folder = get_model_folder(model_hash)
@@ -39,13 +45,13 @@ def test_get_model_folder_single_library(tmp_path):
assert relative == model_hash
def test_get_model_folder_multi_library(tmp_path):
settings.settings['example_images_path'] = str(tmp_path)
settings.settings['libraries'] = {
def test_get_model_folder_multi_library(tmp_path, settings_manager):
settings_manager.settings['example_images_path'] = str(tmp_path)
settings_manager.settings['libraries'] = {
'default': {},
'Alt Library': {},
}
settings.settings['active_library'] = 'Alt Library'
settings_manager.settings['active_library'] = 'Alt Library'
model_hash = 'b' * 64
expected_folder = tmp_path / 'Alt_Library' / model_hash
@@ -57,13 +63,13 @@ def test_get_model_folder_multi_library(tmp_path):
assert relative == os.path.join('Alt_Library', model_hash).replace('\\', '/')
def test_get_model_folder_migrates_legacy_structure(tmp_path):
settings.settings['example_images_path'] = str(tmp_path)
settings.settings['libraries'] = {
def test_get_model_folder_migrates_legacy_structure(tmp_path, settings_manager):
settings_manager.settings['example_images_path'] = str(tmp_path)
settings_manager.settings['libraries'] = {
'default': {},
'extra': {},
}
settings.settings['active_library'] = 'extra'
settings_manager.settings['active_library'] = 'extra'
model_hash = 'c' * 64
legacy_folder = tmp_path / model_hash
@@ -82,31 +88,31 @@ def test_get_model_folder_migrates_legacy_structure(tmp_path):
assert (expected_folder / 'image.png').exists()
def test_ensure_library_root_exists_creates_directories(tmp_path):
settings.settings['example_images_path'] = str(tmp_path)
settings.settings['libraries'] = {'default': {}, 'secondary': {}}
settings.settings['active_library'] = 'secondary'
def test_ensure_library_root_exists_creates_directories(tmp_path, settings_manager):
settings_manager.settings['example_images_path'] = str(tmp_path)
settings_manager.settings['libraries'] = {'default': {}, 'secondary': {}}
settings_manager.settings['active_library'] = 'secondary'
resolved = ensure_library_root_exists('secondary')
assert Path(resolved) == tmp_path / 'secondary'
assert (tmp_path / 'secondary').is_dir()
def test_iter_library_roots_returns_all_configured(tmp_path):
settings.settings['example_images_path'] = str(tmp_path)
settings.settings['libraries'] = {'default': {}, 'alt': {}}
settings.settings['active_library'] = 'alt'
def test_iter_library_roots_returns_all_configured(tmp_path, settings_manager):
settings_manager.settings['example_images_path'] = str(tmp_path)
settings_manager.settings['libraries'] = {'default': {}, 'alt': {}}
settings_manager.settings['active_library'] = 'alt'
roots = dict(iter_library_roots())
assert roots['default'] == str(tmp_path / 'default')
assert roots['alt'] == str(tmp_path / 'alt')
def test_is_valid_example_images_root_accepts_hash_directories(tmp_path):
settings.settings['example_images_path'] = str(tmp_path)
def test_is_valid_example_images_root_accepts_hash_directories(tmp_path, settings_manager):
settings_manager.settings['example_images_path'] = str(tmp_path)
# Ensure single-library mode (not multi-library mode)
settings.settings['libraries'] = {'default': {}}
settings.settings['active_library'] = 'default'
settings_manager.settings['libraries'] = {'default': {}}
settings_manager.settings['active_library'] = 'default'
hash_folder = tmp_path / ('d' * 64)
hash_folder.mkdir()

View File

@@ -7,18 +7,19 @@ from typing import Any, Dict, Tuple
import pytest
from py.services.settings_manager import settings
from py.services.settings_manager import get_settings_manager
from py.utils import example_images_processor as processor_module
@pytest.fixture(autouse=True)
def restore_settings() -> None:
original = settings.settings.copy()
manager = get_settings_manager()
original = manager.settings.copy()
try:
yield
finally:
settings.settings.clear()
settings.settings.update(original)
manager.settings.clear()
manager.settings.update(original)
def test_get_file_extension_from_magic_bytes() -> None:
@@ -90,9 +91,10 @@ def stub_scanners(monkeypatch: pytest.MonkeyPatch, tmp_path) -> StubScanner:
async def test_import_images_creates_hash_directory(monkeypatch: pytest.MonkeyPatch, tmp_path, stub_scanners: StubScanner) -> None:
settings.settings["example_images_path"] = str(tmp_path / "examples")
settings.settings["libraries"] = {"default": {}}
settings.settings["active_library"] = "default"
settings_manager = get_settings_manager()
settings_manager.settings["example_images_path"] = str(tmp_path / "examples")
settings_manager.settings["libraries"] = {"default": {}}
settings_manager.settings["active_library"] = "default"
source_file = tmp_path / "upload.png"
source_file.write_bytes(b"PNG data")
@@ -112,7 +114,7 @@ async def test_import_images_creates_hash_directory(monkeypatch: pytest.MonkeyPa
assert result["success"] is True
assert result["files"][0]["name"].startswith("custom_short")
model_folder = Path(settings.settings["example_images_path"]) / ("a" * 64)
model_folder = Path(settings_manager.settings["example_images_path"]) / ("a" * 64)
assert model_folder.exists()
created_files = list(model_folder.glob("custom_short*.png"))
assert len(created_files) == 1
@@ -132,7 +134,8 @@ async def test_import_images_rejects_missing_parameters(monkeypatch: pytest.Monk
async def test_import_images_raises_when_model_not_found(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
settings.settings["example_images_path"] = str(tmp_path)
settings_manager = get_settings_manager()
settings_manager.settings["example_images_path"] = str(tmp_path)
async def _empty_scanner(cls=None):
return StubScanner([])

View File

@@ -1,6 +1,6 @@
import pytest
from py.services.settings_manager import settings
from py.services.settings_manager import SettingsManager, get_settings_manager
from py.utils.utils import (
calculate_recipe_fingerprint,
calculate_relative_path_for_model,
@@ -9,7 +9,8 @@ from py.utils.utils import (
@pytest.fixture
def isolated_settings(monkeypatch):
default_settings = settings._get_default_settings()
manager = get_settings_manager()
default_settings = manager._get_default_settings()
default_settings.update(
{
"download_path_templates": {
@@ -20,8 +21,8 @@ def isolated_settings(monkeypatch):
"base_model_path_mappings": {},
}
)
monkeypatch.setattr(settings, "settings", default_settings)
monkeypatch.setattr(type(settings), "_save_settings", lambda self: None)
monkeypatch.setattr(manager, "settings", default_settings)
monkeypatch.setattr(SettingsManager, "_save_settings", lambda self: None)
return default_settings