mirror of
https://github.com/willmiao/ComfyUI-Lora-Manager.git
synced 2026-09-26 21:44:09 -03:00
feat: optional centralized storage for sidecar metadata and previews (#1045)
Add an opt-in 'centralized' sidecar storage mode alongside the default 'alongside' layout. In centralized mode, .metadata.json sidecars and preview assets live under a configurable root (sidecar_storage_path, default <settings_dir>/sidecars), mirroring the library-relative directory structure: <root>/<library>/<root_basename>/<rel_dir>/. Backend: - settings: sidecar_storage_mode / sidecar_storage_path with validation; changing either refreshes the preview allowlist - config: centralized root added to preview-serving allowlist - lifecycle: delete / move / rename / folder-rename / folder-delete and undoable-delete staging all operate on the mirror tree in centralized mode (model files themselves never move); EXDEV-tolerant cross- filesystem moves - scanners: pending-hash filesystem scan walks the mirror tree in centralized mode; preview discovery reads from the sidecar dir; .civitai.info stays co-located in both modes - migration: SidecarMigrationUseCase moves sidecars+previews between layouts both directions (keep-newer conflict resolution, preview_url rewriting, WebSocket progress), exposed as POST+GET /api/lm/sidecars/migrate with a mode guard (force=true for the settings-first flow) Frontend: - settings modal: sidecar storage section (mode select + path input with browse/validation), mode-change confirmation offering immediate migration (force=true), and a 'Migrate Sidecars Now' action - i18n keys synced to all locales ([TODO: Translate] placeholders) Docs: metadata-json-schema.md gains a storage-location section; AGENTS.md records the sidecar_paths helper convention.
This commit is contained in:
Will Miao
@@ -0,0 +1,284 @@
|
||||
import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
|
||||
|
||||
vi.mock('../../../static/js/managers/ModalManager.js', () => ({
|
||||
modalManager: {
|
||||
closeModal: vi.fn(),
|
||||
},
|
||||
}));
|
||||
|
||||
vi.mock('../../../static/js/utils/uiHelpers.js', () => ({
|
||||
showToast: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock('../../../static/js/state/index.js', () => {
|
||||
const settings = {};
|
||||
return {
|
||||
state: {
|
||||
global: {
|
||||
settings,
|
||||
},
|
||||
loadingManager: {
|
||||
showSimpleLoading: vi.fn(),
|
||||
hide: vi.fn(),
|
||||
},
|
||||
},
|
||||
createDefaultSettings: () => ({
|
||||
language: 'en',
|
||||
sidecar_storage_mode: 'alongside',
|
||||
sidecar_storage_path: '',
|
||||
}),
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock('../../../static/js/api/modelApiFactory.js', () => ({
|
||||
resetAndReload: vi.fn(),
|
||||
getModelApiClient: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock('../../../static/js/utils/constants.js', () => ({
|
||||
DOWNLOAD_PATH_TEMPLATES: {},
|
||||
DEFAULT_PATH_TEMPLATES: {},
|
||||
MAPPABLE_BASE_MODELS: [],
|
||||
PATH_TEMPLATE_PLACEHOLDERS: {},
|
||||
FILENAME_TEMPLATE_PLACEHOLDERS: [],
|
||||
DEFAULT_FILENAME_TEMPLATES: { lora: '', checkpoint: '', embedding: '' },
|
||||
DEFAULT_PRIORITY_TAG_CONFIG: {
|
||||
lora: 'character, style',
|
||||
checkpoint: 'base, guide',
|
||||
embedding: 'hint',
|
||||
},
|
||||
getMappableBaseModelsDynamic: () => [],
|
||||
}));
|
||||
|
||||
vi.mock('../../../static/js/utils/i18nHelpers.js', () => ({
|
||||
translate: (_key, _params, fallback) => fallback ?? '',
|
||||
}));
|
||||
|
||||
vi.mock('../../../static/js/i18n/index.js', () => ({
|
||||
i18n: {
|
||||
getCurrentLocale: () => 'en',
|
||||
setLanguage: vi.fn().mockResolvedValue(),
|
||||
},
|
||||
}));
|
||||
|
||||
vi.mock('../../../static/js/components/shared/ModelCard.js', () => ({
|
||||
configureModelCardVideo: vi.fn(),
|
||||
}));
|
||||
|
||||
import { SettingsManager } from '../../../static/js/managers/SettingsManager.js';
|
||||
import { showToast } from '../../../static/js/utils/uiHelpers.js';
|
||||
import { resetAndReload } from '../../../static/js/api/modelApiFactory.js';
|
||||
import { state } from '../../../static/js/state/index.js';
|
||||
|
||||
const createManager = () => {
|
||||
state.global.settings = {};
|
||||
const initSettingsSpy = vi
|
||||
.spyOn(SettingsManager.prototype, 'initializeSettings')
|
||||
.mockResolvedValue();
|
||||
const initializeSpy = vi
|
||||
.spyOn(SettingsManager.prototype, 'initialize')
|
||||
.mockImplementation(() => {});
|
||||
|
||||
const manager = new SettingsManager();
|
||||
|
||||
initSettingsSpy.mockRestore();
|
||||
initializeSpy.mockRestore();
|
||||
|
||||
return manager;
|
||||
};
|
||||
|
||||
const appendSidecarControls = () => {
|
||||
const select = document.createElement('select');
|
||||
select.id = 'sidecarStorageMode';
|
||||
['alongside', 'centralized'].forEach((value) => {
|
||||
const option = document.createElement('option');
|
||||
option.value = value;
|
||||
select.appendChild(option);
|
||||
});
|
||||
|
||||
const pathSetting = document.createElement('div');
|
||||
pathSetting.id = 'sidecarStoragePathSetting';
|
||||
pathSetting.style.display = 'none';
|
||||
|
||||
const pathInput = document.createElement('input');
|
||||
pathInput.id = 'sidecarStoragePath';
|
||||
|
||||
const migrateBtn = document.createElement('button');
|
||||
migrateBtn.id = 'migrateSidecarsBtn';
|
||||
|
||||
document.body.append(select, pathSetting, pathInput, migrateBtn);
|
||||
return { select, pathSetting, pathInput, migrateBtn };
|
||||
};
|
||||
|
||||
const appendMigrationModal = () => {
|
||||
const modal = document.createElement('div');
|
||||
modal.id = 'sidecarMigrationConfirmModal';
|
||||
modal.innerHTML = `
|
||||
<h2 data-role="title"></h2>
|
||||
<p data-role="message"></p>
|
||||
<button data-action="confirm-sidecar-migration"></button>
|
||||
<button data-action="cancel-sidecar-migration"></button>`;
|
||||
document.body.appendChild(modal);
|
||||
return modal;
|
||||
};
|
||||
|
||||
const mockFetchOk = (payload = { success: true }) => {
|
||||
global.fetch = vi.fn().mockResolvedValue({
|
||||
ok: true,
|
||||
json: () => Promise.resolve(payload),
|
||||
});
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
document.body.innerHTML = '';
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
delete global.fetch;
|
||||
});
|
||||
|
||||
describe('SettingsManager sidecar storage', () => {
|
||||
describe('loadSidecarStorageSettings', () => {
|
||||
it('loads alongside mode and hides the centralized path input', () => {
|
||||
const manager = createManager();
|
||||
const { select, pathSetting, pathInput } = appendSidecarControls();
|
||||
state.global.settings = { sidecar_storage_mode: 'alongside', sidecar_storage_path: '/data/sidecars' };
|
||||
|
||||
manager.loadSidecarStorageSettings();
|
||||
|
||||
expect(select.value).toBe('alongside');
|
||||
expect(pathInput.value).toBe('/data/sidecars');
|
||||
expect(pathSetting.style.display).toBe('none');
|
||||
expect(manager._loadedSidecarStorageMode).toBe('alongside');
|
||||
});
|
||||
|
||||
it('loads centralized mode and shows the path input', () => {
|
||||
const manager = createManager();
|
||||
const { select, pathSetting } = appendSidecarControls();
|
||||
state.global.settings = { sidecar_storage_mode: 'centralized' };
|
||||
|
||||
manager.loadSidecarStorageSettings();
|
||||
|
||||
expect(select.value).toBe('centralized');
|
||||
expect(pathSetting.style.display).toBe('block');
|
||||
expect(manager._loadedSidecarStorageMode).toBe('centralized');
|
||||
});
|
||||
|
||||
it('falls back to alongside for unknown stored modes', () => {
|
||||
const manager = createManager();
|
||||
const { select } = appendSidecarControls();
|
||||
state.global.settings = { sidecar_storage_mode: 'bogus' };
|
||||
|
||||
manager.loadSidecarStorageSettings();
|
||||
|
||||
expect(select.value).toBe('alongside');
|
||||
});
|
||||
});
|
||||
|
||||
describe('handleSidecarStorageModeChange', () => {
|
||||
it('does not prompt for migration when the mode is unchanged', async () => {
|
||||
const manager = createManager();
|
||||
const { select } = appendSidecarControls();
|
||||
appendMigrationModal();
|
||||
state.global.settings = { sidecar_storage_mode: 'alongside' };
|
||||
manager._loadedSidecarStorageMode = 'alongside';
|
||||
select.value = 'alongside';
|
||||
mockFetchOk();
|
||||
|
||||
await manager.handleSidecarStorageModeChange();
|
||||
|
||||
expect(global.fetch).not.toHaveBeenCalledWith(
|
||||
'/api/lm/sidecars/migrate',
|
||||
expect.anything()
|
||||
);
|
||||
expect(showToast).toHaveBeenCalledWith(
|
||||
'toast.settings.settingsUpdated',
|
||||
expect.anything(),
|
||||
'success'
|
||||
);
|
||||
});
|
||||
|
||||
it('migrates to centralized after the user confirms the prompt', async () => {
|
||||
const manager = createManager();
|
||||
const { select, pathSetting } = appendSidecarControls();
|
||||
const modal = appendMigrationModal();
|
||||
state.global.settings = { sidecar_storage_mode: 'alongside' };
|
||||
manager._loadedSidecarStorageMode = 'alongside';
|
||||
select.value = 'centralized';
|
||||
mockFetchOk();
|
||||
|
||||
const changePromise = manager.handleSidecarStorageModeChange();
|
||||
|
||||
await vi.waitFor(() => expect(modal.classList.contains('show')).toBe(true));
|
||||
modal.querySelector('[data-action="confirm-sidecar-migration"]').click();
|
||||
await changePromise;
|
||||
|
||||
expect(state.global.settings.sidecar_storage_mode).toBe('centralized');
|
||||
expect(pathSetting.style.display).toBe('block');
|
||||
expect(global.fetch).toHaveBeenCalledWith('/api/lm/sidecars/migrate', expect.objectContaining({
|
||||
method: 'POST',
|
||||
body: JSON.stringify({ direction: 'to_centralized', force: true }),
|
||||
}));
|
||||
expect(showToast).toHaveBeenCalledWith('settings.sidecarStorage.migrateSuccess', {}, 'success');
|
||||
expect(resetAndReload).toHaveBeenCalledWith(true);
|
||||
expect(modal.classList.contains('show')).toBe(false);
|
||||
});
|
||||
|
||||
it('shows a deferred notice and skips migration when the user cancels', async () => {
|
||||
const manager = createManager();
|
||||
const { select } = appendSidecarControls();
|
||||
const modal = appendMigrationModal();
|
||||
state.global.settings = { sidecar_storage_mode: 'centralized' };
|
||||
manager._loadedSidecarStorageMode = 'centralized';
|
||||
select.value = 'alongside';
|
||||
mockFetchOk();
|
||||
|
||||
const changePromise = manager.handleSidecarStorageModeChange();
|
||||
await vi.waitFor(() => expect(modal.classList.contains('show')).toBe(true));
|
||||
modal.querySelector('[data-action="cancel-sidecar-migration"]').click();
|
||||
await changePromise;
|
||||
|
||||
const migrateCalls = global.fetch.mock.calls.filter(([url]) => url === '/api/lm/sidecars/migrate');
|
||||
expect(migrateCalls).toHaveLength(0);
|
||||
expect(showToast).toHaveBeenCalledWith('settings.sidecarStorage.migrationDeferred', {}, 'info');
|
||||
});
|
||||
});
|
||||
|
||||
describe('confirmAndMigrateSidecars', () => {
|
||||
it('derives the migration direction from the saved mode', async () => {
|
||||
const manager = createManager();
|
||||
appendSidecarControls();
|
||||
const modal = appendMigrationModal();
|
||||
state.global.settings = { sidecar_storage_mode: 'centralized' };
|
||||
mockFetchOk();
|
||||
|
||||
const confirmPromise = manager.confirmAndMigrateSidecars();
|
||||
await vi.waitFor(() => expect(modal.classList.contains('show')).toBe(true));
|
||||
modal.querySelector('[data-action="confirm-sidecar-migration"]').click();
|
||||
await confirmPromise;
|
||||
|
||||
expect(global.fetch).toHaveBeenCalledWith('/api/lm/sidecars/migrate', expect.objectContaining({
|
||||
body: JSON.stringify({ direction: 'to_centralized', force: true }),
|
||||
}));
|
||||
});
|
||||
});
|
||||
|
||||
describe('migrateSidecars', () => {
|
||||
it('surfaces backend failures as an error toast', async () => {
|
||||
const manager = createManager();
|
||||
const { migrateBtn } = appendSidecarControls();
|
||||
mockFetchOk({ success: false, error: 'disk full' });
|
||||
|
||||
await manager.migrateSidecars('to_alongside');
|
||||
|
||||
expect(showToast).toHaveBeenCalledWith(
|
||||
'settings.sidecarStorage.migrateFailed',
|
||||
{ message: 'disk full' },
|
||||
'error'
|
||||
);
|
||||
expect(resetAndReload).not.toHaveBeenCalled();
|
||||
expect(migrateBtn.disabled).toBe(false);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -23,6 +23,7 @@ from py.routes.handlers.misc_handlers import (
|
||||
NodeRegistryHandler,
|
||||
ServiceRegistryAdapter,
|
||||
SettingsHandler,
|
||||
SidecarMigrationHandler,
|
||||
_collect_comfyui_session_logs,
|
||||
_is_wsl,
|
||||
_wsl_to_windows_path,
|
||||
@@ -2557,3 +2558,86 @@ async def test_get_model_versions_status_supported_type_stays_interactive():
|
||||
"hasBeenDownloaded": False,
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
class DummySidecarMigrationUseCase:
|
||||
def __init__(self, result):
|
||||
self.result = result
|
||||
self.calls = []
|
||||
|
||||
async def execute_with_error_handling(self, *, direction, progress_cb=None, force=False):
|
||||
self.calls.append({"direction": direction, "force": force})
|
||||
return self.result
|
||||
|
||||
|
||||
def _sidecar_migration_handler(result):
|
||||
use_case = DummySidecarMigrationUseCase(result)
|
||||
handler = SidecarMigrationHandler(
|
||||
use_case_factory=lambda: use_case,
|
||||
progress_callback_factory=lambda: None,
|
||||
)
|
||||
return handler, use_case
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sidecar_migration_handler_runs_to_centralized():
|
||||
result = {"success": True, "direction": "to_centralized", "moved": 3}
|
||||
handler, use_case = _sidecar_migration_handler(result)
|
||||
|
||||
response = await handler.migrate_sidecars(
|
||||
FakeRequest(json_data={"direction": "to_centralized", "force": True}) # pyright: ignore[reportArgumentType]
|
||||
)
|
||||
payload = _json_payload(response)
|
||||
|
||||
assert response.status == 200
|
||||
assert payload["success"] is True
|
||||
assert payload["moved"] == 3
|
||||
assert use_case.calls == [{"direction": "to_centralized", "force": True}]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sidecar_migration_handler_rejects_bad_direction():
|
||||
handler, use_case = _sidecar_migration_handler({"success": True})
|
||||
|
||||
response = await handler.migrate_sidecars(
|
||||
FakeRequest(json_data={"direction": "sideways"}) # pyright: ignore[reportArgumentType]
|
||||
)
|
||||
payload = _json_payload(response)
|
||||
|
||||
assert response.status == 400
|
||||
assert payload["success"] is False
|
||||
assert use_case.calls == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sidecar_migration_handler_accepts_get_query_params():
|
||||
result = {"success": True, "direction": "to_alongside", "moved": 0}
|
||||
handler, use_case = _sidecar_migration_handler(result)
|
||||
|
||||
response = await handler.migrate_sidecars(
|
||||
FakeRequest( # pyright: ignore[reportArgumentType]
|
||||
query={"direction": "to_alongside", "force": "true"},
|
||||
method="GET",
|
||||
)
|
||||
)
|
||||
payload = _json_payload(response)
|
||||
|
||||
assert response.status == 200
|
||||
assert payload["success"] is True
|
||||
assert use_case.calls == [{"direction": "to_alongside", "force": True}]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sidecar_migration_handler_guard_refusal_is_400():
|
||||
result = {"success": False, "error": "sidecar storage is already centralized"}
|
||||
handler, use_case = _sidecar_migration_handler(result)
|
||||
|
||||
response = await handler.migrate_sidecars(
|
||||
FakeRequest(json_data={"direction": "to_centralized"}) # pyright: ignore[reportArgumentType]
|
||||
)
|
||||
payload = _json_payload(response)
|
||||
|
||||
assert response.status == 400
|
||||
assert payload["success"] is False
|
||||
assert "already centralized" in payload["error"]
|
||||
assert use_case.calls == [{"direction": "to_centralized", "force": False}]
|
||||
|
||||
@@ -0,0 +1,361 @@
|
||||
"""Centralized sidecar storage: lifecycle flows (delete/move/rename/scans)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
|
||||
from py.config import config
|
||||
from py.services.checkpoint_scanner import CheckpointScanner
|
||||
from py.services.model_lifecycle_service import (
|
||||
ModelLifecycleService,
|
||||
delete_model_artifacts,
|
||||
)
|
||||
from py.services.pending_delete_service import PendingDeleteService
|
||||
from py.services.settings_manager import get_settings_manager
|
||||
from py.utils.metadata_manager import MetadataManager
|
||||
|
||||
|
||||
def _normalize(path) -> str:
|
||||
return str(path).replace(os.sep, "/")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def library_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
"""Single checkpoint root under tmp_path; every other root emptied."""
|
||||
|
||||
root = tmp_path / "checkpoints"
|
||||
root.mkdir()
|
||||
for attr, value in (
|
||||
("loras_roots", []),
|
||||
("base_models_roots", [str(root)]),
|
||||
("checkpoints_roots", [str(root)]),
|
||||
("embeddings_roots", []),
|
||||
("other_roots", []),
|
||||
("extra_loras_roots", []),
|
||||
("extra_checkpoints_roots", []),
|
||||
("extra_unet_roots", []),
|
||||
("extra_embeddings_roots", []),
|
||||
):
|
||||
monkeypatch.setattr(config, attr, value, raising=False)
|
||||
return root
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def centralized(library_root: Path, tmp_path: Path) -> Path:
|
||||
"""Enable centralized mode rooted at tmp_path/sidecars."""
|
||||
|
||||
sidecar_root = tmp_path / "sidecars"
|
||||
settings = get_settings_manager()
|
||||
settings.set("sidecar_storage_mode", "centralized")
|
||||
settings.set("sidecar_storage_path", str(sidecar_root))
|
||||
return sidecar_root
|
||||
|
||||
|
||||
def _mirror_dir(sidecar_root: Path, *rel: str) -> Path:
|
||||
"""Expected mirror directory for a library-relative path."""
|
||||
|
||||
library = get_settings_manager().get_active_library_name()
|
||||
return sidecar_root.joinpath(library, "checkpoints", *rel)
|
||||
|
||||
|
||||
def _write_sidecar(
|
||||
mirror_dir: Path,
|
||||
stem: str,
|
||||
*,
|
||||
file_path: str,
|
||||
preview_name: str | None = None,
|
||||
extra: Dict[str, Any] | None = None,
|
||||
) -> Path:
|
||||
mirror_dir.mkdir(parents=True, exist_ok=True)
|
||||
payload: Dict[str, Any] = {
|
||||
"file_name": stem,
|
||||
"file_path": _normalize(file_path),
|
||||
}
|
||||
if preview_name:
|
||||
payload["preview_url"] = _normalize(mirror_dir / preview_name)
|
||||
(mirror_dir / preview_name).write_bytes(b"preview")
|
||||
if extra:
|
||||
payload.update(extra)
|
||||
metadata_path = mirror_dir / f"{stem}.metadata.json"
|
||||
metadata_path.write_text(json.dumps(payload), encoding="utf-8")
|
||||
return metadata_path
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_model_artifacts_centralized(
|
||||
library_root: Path, centralized: Path
|
||||
):
|
||||
model = library_root / "model.safetensors"
|
||||
model.write_bytes(b"weights")
|
||||
mirror = _mirror_dir(centralized)
|
||||
_write_sidecar(mirror, "model", file_path=model, preview_name="model.preview.webp")
|
||||
|
||||
deleted = await delete_model_artifacts(str(library_root), "model")
|
||||
|
||||
assert not model.exists()
|
||||
assert not (mirror / "model.metadata.json").exists()
|
||||
assert not (mirror / "model.preview.webp").exists()
|
||||
assert any(path.endswith("model.safetensors") for path in deleted)
|
||||
assert "model.metadata.json" in deleted
|
||||
assert "model.preview.webp" in deleted
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_model_artifacts_alongside_still_siblings(
|
||||
library_root: Path,
|
||||
):
|
||||
"""Alongside mode (default) keeps deleting sidecars next to the model."""
|
||||
|
||||
model = library_root / "model.safetensors"
|
||||
model.write_bytes(b"weights")
|
||||
sidecar = library_root / "model.metadata.json"
|
||||
sidecar.write_text("{}", encoding="utf-8")
|
||||
preview = library_root / "model.preview.webp"
|
||||
preview.write_bytes(b"preview")
|
||||
|
||||
await delete_model_artifacts(str(library_root), "model")
|
||||
|
||||
assert not model.exists()
|
||||
assert not sidecar.exists()
|
||||
assert not preview.exists()
|
||||
|
||||
|
||||
def test_enumerate_model_artifacts_centralized(
|
||||
library_root: Path, centralized: Path
|
||||
):
|
||||
model = library_root / "model.safetensors"
|
||||
model.write_bytes(b"weights")
|
||||
mirror = _mirror_dir(centralized)
|
||||
_write_sidecar(mirror, "model", file_path=model, preview_name="model.preview.png")
|
||||
|
||||
service = PendingDeleteService.__new__(PendingDeleteService)
|
||||
artifacts = service._enumerate_model_artifacts(
|
||||
str(library_root), "model", ".safetensors"
|
||||
)
|
||||
|
||||
assert artifacts == [
|
||||
os.path.abspath(str(model)),
|
||||
os.path.abspath(str(mirror / "model.metadata.json")),
|
||||
os.path.abspath(str(mirror / "model.preview.png")),
|
||||
]
|
||||
|
||||
|
||||
class _RecordingScanner:
|
||||
def __init__(self):
|
||||
self.calls: List[tuple] = []
|
||||
self.model_type = "lora"
|
||||
|
||||
async def update_single_model_cache(self, old_path, new_path, metadata):
|
||||
self.calls.append((old_path, new_path, metadata))
|
||||
|
||||
|
||||
class _PassthroughMetadataManager:
|
||||
async def save_metadata(self, path: str, metadata):
|
||||
await MetadataManager.save_metadata(path, metadata)
|
||||
return True
|
||||
|
||||
|
||||
async def _json_metadata_loader(path: str) -> Dict[str, object]:
|
||||
with open(path, "r", encoding="utf-8") as handle:
|
||||
return json.load(handle)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rename_model_centralized(library_root: Path, centralized: Path):
|
||||
model = library_root / "model.safetensors"
|
||||
model.write_bytes(b"weights")
|
||||
mirror = _mirror_dir(centralized)
|
||||
_write_sidecar(mirror, "model", file_path=model, preview_name="model.preview.webp")
|
||||
|
||||
service = ModelLifecycleService(
|
||||
scanner=_RecordingScanner(),
|
||||
metadata_manager=_PassthroughMetadataManager(),
|
||||
metadata_loader=_json_metadata_loader,
|
||||
)
|
||||
|
||||
result = await service.rename_model(
|
||||
file_path=_normalize(model), new_file_name="renamed"
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
renamed = library_root / "renamed.safetensors"
|
||||
assert renamed.exists()
|
||||
assert not model.exists()
|
||||
|
||||
# Mirror sidecar/preview renamed with the model; nothing left alongside.
|
||||
assert (mirror / "renamed.metadata.json").exists()
|
||||
assert (mirror / "renamed.preview.webp").exists()
|
||||
assert not (mirror / "model.metadata.json").exists()
|
||||
assert not (mirror / "model.preview.webp").exists()
|
||||
assert not (library_root / "renamed.metadata.json").exists()
|
||||
assert not (library_root / "renamed.preview.webp").exists()
|
||||
|
||||
saved = json.loads((mirror / "renamed.metadata.json").read_text())
|
||||
assert saved["file_name"] == "renamed"
|
||||
assert saved["file_path"].endswith("renamed.safetensors")
|
||||
assert saved["preview_url"].endswith("renamed.preview.webp")
|
||||
assert str(mirror).replace(os.sep, "/") in saved["preview_url"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_move_model_centralized(
|
||||
library_root: Path, centralized: Path, monkeypatch: pytest.MonkeyPatch
|
||||
):
|
||||
source_dir = library_root / "old"
|
||||
source_dir.mkdir()
|
||||
model = source_dir / "model.safetensors"
|
||||
model.write_bytes(b"weights")
|
||||
target_dir = library_root / "new"
|
||||
|
||||
old_mirror = _mirror_dir(centralized, "old")
|
||||
_write_sidecar(
|
||||
old_mirror, "model", file_path=model, preview_name="model.preview.webp"
|
||||
)
|
||||
|
||||
scanner = CheckpointScanner()
|
||||
monkeypatch.setattr(
|
||||
scanner, "update_single_model_cache", AsyncMock(return_value=True)
|
||||
)
|
||||
|
||||
result = await scanner.move_model(_normalize(model), _normalize(target_dir))
|
||||
|
||||
assert result is not None
|
||||
moved_model = target_dir / "model.safetensors"
|
||||
assert moved_model.exists()
|
||||
assert not model.exists()
|
||||
|
||||
new_mirror = _mirror_dir(centralized, "new")
|
||||
assert (new_mirror / "model.metadata.json").exists()
|
||||
assert (new_mirror / "model.preview.webp").exists()
|
||||
assert not (old_mirror / "model.metadata.json").exists()
|
||||
assert not (old_mirror / "model.preview.webp").exists()
|
||||
|
||||
saved = json.loads((new_mirror / "model.metadata.json").read_text())
|
||||
assert saved["file_path"].endswith("new/model.safetensors")
|
||||
assert saved["preview_url"].endswith("model.preview.webp")
|
||||
assert str(new_mirror).replace(os.sep, "/") in saved["preview_url"]
|
||||
|
||||
|
||||
class _FakeCache:
|
||||
def __init__(self, raw_data: List[Dict[str, Any]]):
|
||||
self.raw_data = raw_data
|
||||
self.all_folders: List[str] = []
|
||||
self.folders: List[str] = []
|
||||
|
||||
def remove_from_version_index(self, _item) -> None:
|
||||
pass
|
||||
|
||||
def rebuild_version_index(self) -> None:
|
||||
pass
|
||||
|
||||
async def resort(self) -> None:
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rename_known_folder_centralized(
|
||||
library_root: Path, centralized: Path, monkeypatch: pytest.MonkeyPatch
|
||||
):
|
||||
# The caller (ModelFileService.rename_folder) has already renamed the
|
||||
# model directory on disk when the scanner is asked to re-key records.
|
||||
old_dir = library_root / "oldfolder"
|
||||
new_dir = library_root / "newfolder"
|
||||
new_dir.mkdir()
|
||||
model = new_dir / "model.safetensors"
|
||||
model.write_bytes(b"weights")
|
||||
|
||||
old_model_path = old_dir / "model.safetensors"
|
||||
old_mirror = _mirror_dir(centralized, "oldfolder")
|
||||
_write_sidecar(
|
||||
old_mirror, "model", file_path=old_model_path, preview_name="model.preview.webp"
|
||||
)
|
||||
|
||||
cache_entry: Dict[str, Any] = {
|
||||
"file_path": _normalize(old_model_path),
|
||||
"folder": "oldfolder",
|
||||
"preview_url": _normalize(old_mirror / "model.preview.webp"),
|
||||
"sha256": "",
|
||||
}
|
||||
scanner = CheckpointScanner()
|
||||
scanner._cache = _FakeCache([cache_entry])
|
||||
monkeypatch.setattr(scanner, "_persist_current_cache", AsyncMock())
|
||||
|
||||
changed = await scanner.rename_known_folder(
|
||||
"oldfolder",
|
||||
"newfolder",
|
||||
previous_path=str(old_dir),
|
||||
new_path=str(new_dir),
|
||||
)
|
||||
|
||||
assert changed is True
|
||||
|
||||
new_mirror = _mirror_dir(centralized, "newfolder")
|
||||
assert (new_mirror / "model.metadata.json").exists()
|
||||
assert (new_mirror / "model.preview.webp").exists()
|
||||
assert not old_mirror.exists()
|
||||
|
||||
saved = json.loads((new_mirror / "model.metadata.json").read_text())
|
||||
assert saved["file_path"] == _normalize(model)
|
||||
assert saved["preview_url"] == _normalize(new_mirror / "model.preview.webp")
|
||||
|
||||
assert cache_entry["file_path"] == _normalize(model)
|
||||
assert cache_entry["folder"] == "newfolder"
|
||||
assert cache_entry["preview_url"] == _normalize(
|
||||
new_mirror / "model.preview.webp"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pending_models_mirror_walk(library_root: Path, centralized: Path):
|
||||
model = library_root / "model.safetensors"
|
||||
model.write_bytes(b"weights")
|
||||
mirror = _mirror_dir(centralized)
|
||||
_write_sidecar(
|
||||
mirror,
|
||||
"model",
|
||||
file_path=model,
|
||||
extra={"hash_status": "pending", "sha256": ""},
|
||||
)
|
||||
# Orphan sidecar: recorded model path is gone and no stem match exists.
|
||||
_write_sidecar(
|
||||
mirror,
|
||||
"ghost",
|
||||
file_path=library_root / "ghost.safetensors",
|
||||
extra={"hash_status": "pending", "sha256": ""},
|
||||
)
|
||||
|
||||
scanner = CheckpointScanner()
|
||||
pending = await scanner._find_pending_models_from_filesystem()
|
||||
|
||||
assert len(pending) == 1
|
||||
assert pending[0]["file_path"] == _normalize(model)
|
||||
assert pending[0]["hash_status"] == "pending"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pending_models_mirror_walk_uses_stem_fallback(
|
||||
library_root: Path, centralized: Path
|
||||
):
|
||||
"""A stale recorded file_path falls back to probing by stem + extension."""
|
||||
|
||||
model = library_root / "model.safetensors"
|
||||
model.write_bytes(b"weights")
|
||||
mirror = _mirror_dir(centralized)
|
||||
_write_sidecar(
|
||||
mirror,
|
||||
"model",
|
||||
file_path=library_root / "renamed-away.safetensors",
|
||||
extra={"hash_status": "pending", "sha256": ""},
|
||||
)
|
||||
|
||||
scanner = CheckpointScanner()
|
||||
pending = await scanner._find_pending_models_from_filesystem()
|
||||
|
||||
assert len(pending) == 1
|
||||
assert pending[0]["file_path"] == _normalize(model)
|
||||
@@ -0,0 +1,294 @@
|
||||
"""Sidecar migration use case: layout moves, conflicts, guards."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List
|
||||
|
||||
import pytest
|
||||
|
||||
from py.config import config
|
||||
from py.services.settings_manager import get_settings_manager
|
||||
from py.services.use_cases.sidecar_migration_use_case import SidecarMigrationUseCase
|
||||
|
||||
|
||||
def _normalize(path) -> str:
|
||||
return str(path).replace(os.sep, "/")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def library_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
"""Single lora root under tmp_path; every other root emptied."""
|
||||
|
||||
root = tmp_path / "loras"
|
||||
root.mkdir()
|
||||
for attr, value in (
|
||||
("loras_roots", [str(root)]),
|
||||
("base_models_roots", []),
|
||||
("checkpoints_roots", []),
|
||||
("embeddings_roots", []),
|
||||
("other_roots", []),
|
||||
("extra_loras_roots", []),
|
||||
("extra_checkpoints_roots", []),
|
||||
("extra_unet_roots", []),
|
||||
("extra_embeddings_roots", []),
|
||||
):
|
||||
monkeypatch.setattr(config, attr, value, raising=False)
|
||||
return root
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sidecar_root(tmp_path: Path) -> Path:
|
||||
"""Configured centralized root (mode-independent for migration)."""
|
||||
|
||||
root = tmp_path / "sidecars"
|
||||
get_settings_manager().set("sidecar_storage_path", str(root))
|
||||
return root
|
||||
|
||||
|
||||
def _set_mode(mode: str) -> None:
|
||||
get_settings_manager().set("sidecar_storage_mode", mode)
|
||||
|
||||
|
||||
def _mirror_dir(sidecar_root: Path, *rel: str) -> Path:
|
||||
"""Expected mirror directory for a library-relative path."""
|
||||
|
||||
library = get_settings_manager().get_active_library_name()
|
||||
return sidecar_root.joinpath(library, "loras", *rel)
|
||||
|
||||
|
||||
def _write_model(directory: Path, stem: str) -> Path:
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
model = directory / f"{stem}.safetensors"
|
||||
model.write_bytes(b"weights")
|
||||
return model
|
||||
|
||||
|
||||
def _write_sidecar(directory: Path, stem: str, model: Path, *, preview_ext: str | None = ".preview.webp") -> Path:
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
payload: Dict[str, Any] = {
|
||||
"file_name": stem,
|
||||
"file_path": _normalize(model),
|
||||
}
|
||||
if preview_ext:
|
||||
payload["preview_url"] = _normalize(directory / f"{stem}{preview_ext}")
|
||||
sidecar = directory / f"{stem}.metadata.json"
|
||||
sidecar.write_text(json.dumps(payload), encoding="utf-8")
|
||||
return sidecar
|
||||
|
||||
|
||||
class _FakeCache:
|
||||
def __init__(self, raw_data: List[Dict[str, Any]]) -> None:
|
||||
self.raw_data = raw_data
|
||||
|
||||
|
||||
class _FakeScanner:
|
||||
def __init__(self, raw_data: List[Dict[str, Any]]) -> None:
|
||||
self._cache = _FakeCache(raw_data)
|
||||
|
||||
async def get_cached_data(self) -> _FakeCache:
|
||||
return self._cache
|
||||
|
||||
|
||||
def _make_use_case(model_paths: List[str]) -> SidecarMigrationUseCase:
|
||||
scanner = _FakeScanner([{"file_path": path} for path in model_paths])
|
||||
|
||||
async def scanner_factory() -> _FakeScanner:
|
||||
return scanner
|
||||
|
||||
return SidecarMigrationUseCase(
|
||||
scanner_factories=(("lora", scanner_factory),),
|
||||
settings_service=get_settings_manager(),
|
||||
)
|
||||
|
||||
|
||||
class _ProgressRecorder:
|
||||
def __init__(self) -> None:
|
||||
self.payloads: List[Dict[str, Any]] = []
|
||||
|
||||
async def on_progress(self, payload: Dict[str, Any]) -> None:
|
||||
self.payloads.append(payload)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_migrate_to_centralized_moves_sidecar_and_previews(
|
||||
library_root: Path, sidecar_root: Path
|
||||
):
|
||||
_set_mode("centralized")
|
||||
model = _write_model(library_root / "sub", "model")
|
||||
sidecar = _write_sidecar(library_root / "sub", "model", model)
|
||||
preview = library_root / "sub" / "model.preview.webp"
|
||||
preview.write_bytes(b"preview")
|
||||
extra_preview = library_root / "sub" / "model.png"
|
||||
extra_preview.write_bytes(b"extra")
|
||||
|
||||
recorder = _ProgressRecorder()
|
||||
use_case = _make_use_case([str(model)])
|
||||
summary = await use_case.migrate_to_centralized(recorder, force=True)
|
||||
|
||||
assert summary["success"] is True
|
||||
assert summary["direction"] == "to_centralized"
|
||||
assert summary["moved"] == 3
|
||||
assert summary["models_moved"] == 1
|
||||
assert summary["skipped"] == 0
|
||||
assert summary["conflicts"] == 0
|
||||
assert summary["errors"] == []
|
||||
|
||||
mirror = _mirror_dir(sidecar_root, "sub")
|
||||
assert not sidecar.exists()
|
||||
assert not preview.exists()
|
||||
assert not extra_preview.exists()
|
||||
moved_sidecar = mirror / "model.metadata.json"
|
||||
assert moved_sidecar.exists()
|
||||
assert (mirror / "model.preview.webp").exists()
|
||||
assert (mirror / "model.png").exists()
|
||||
# Model files never move.
|
||||
assert model.exists()
|
||||
|
||||
metadata = json.loads(moved_sidecar.read_text(encoding="utf-8"))
|
||||
assert metadata["file_path"] == _normalize(model)
|
||||
assert metadata["file_name"] == "model"
|
||||
# Recorded extension wins when rewriting preview_url.
|
||||
assert metadata["preview_url"] == _normalize(mirror / "model.preview.webp")
|
||||
|
||||
statuses = [payload["status"] for payload in recorder.payloads]
|
||||
assert statuses[0] == "started"
|
||||
assert statuses[-1] == "completed"
|
||||
assert all(p["type"] == "sidecar_migration_progress" for p in recorder.payloads)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_migrate_to_alongside_reverses_layout(
|
||||
library_root: Path, sidecar_root: Path
|
||||
):
|
||||
_set_mode("alongside")
|
||||
model = _write_model(library_root / "sub", "model")
|
||||
mirror = _mirror_dir(sidecar_root, "sub")
|
||||
sidecar = _write_sidecar(mirror, "model", model)
|
||||
preview = mirror / "model.preview.webp"
|
||||
preview.write_bytes(b"preview")
|
||||
|
||||
use_case = _make_use_case([str(model)])
|
||||
summary = await use_case.migrate_to_alongside(force=True)
|
||||
|
||||
assert summary["success"] is True
|
||||
assert summary["moved"] == 2
|
||||
|
||||
assert not sidecar.exists()
|
||||
assert not preview.exists()
|
||||
moved_sidecar = library_root / "sub" / "model.metadata.json"
|
||||
assert moved_sidecar.exists()
|
||||
assert (library_root / "sub" / "model.preview.webp").exists()
|
||||
|
||||
metadata = json.loads(moved_sidecar.read_text(encoding="utf-8"))
|
||||
assert metadata["file_path"] == _normalize(model)
|
||||
assert metadata["preview_url"] == _normalize(
|
||||
library_root / "sub" / "model.preview.webp"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_migrate_conflict_keeps_newer_file(
|
||||
library_root: Path, sidecar_root: Path
|
||||
):
|
||||
_set_mode("centralized")
|
||||
mirror = _mirror_dir(sidecar_root)
|
||||
|
||||
# Model A: destination (mirror) sidecar is newer -> destination wins.
|
||||
model_a = _write_model(library_root, "model_a")
|
||||
src_a = _write_sidecar(library_root, "model_a", model_a, preview_ext=None)
|
||||
dst_a = _write_sidecar(mirror, "model_a", model_a, preview_ext=None)
|
||||
os.utime(src_a, (1000, 1000))
|
||||
os.utime(dst_a, (2000, 2000))
|
||||
|
||||
# Model B: source (alongside) sidecar is newer -> source replaces.
|
||||
model_b = _write_model(library_root, "model_b")
|
||||
src_b = _write_sidecar(library_root, "model_b", model_b, preview_ext=None)
|
||||
dst_b = _write_sidecar(mirror, "model_b", model_b, preview_ext=None)
|
||||
(mirror / "model_b.metadata.json").write_text(
|
||||
json.dumps({"stale": True}), encoding="utf-8"
|
||||
)
|
||||
os.utime(src_b, (3000, 3000))
|
||||
os.utime(dst_b, (2000, 2000))
|
||||
|
||||
use_case = _make_use_case([str(model_a), str(model_b)])
|
||||
summary = await use_case.migrate_to_centralized(force=True)
|
||||
|
||||
assert summary["conflicts"] == 2
|
||||
assert summary["moved"] == 1
|
||||
|
||||
# A: destination kept, source deleted, content untouched.
|
||||
assert not src_a.exists()
|
||||
metadata_a = json.loads(dst_a.read_text(encoding="utf-8"))
|
||||
assert metadata_a["file_name"] == "model_a"
|
||||
|
||||
# B: newer source replaced the stale destination.
|
||||
assert not src_b.exists()
|
||||
metadata_b = json.loads(dst_b.read_text(encoding="utf-8"))
|
||||
assert metadata_b.get("stale") is None
|
||||
assert metadata_b["file_name"] == "model_b"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_migrate_missing_model_file_is_skipped(
|
||||
library_root: Path, sidecar_root: Path
|
||||
):
|
||||
_set_mode("centralized")
|
||||
missing_model = library_root / "ghost.safetensors"
|
||||
sidecar = _write_sidecar(library_root, "ghost", missing_model, preview_ext=None)
|
||||
|
||||
use_case = _make_use_case([str(missing_model)])
|
||||
summary = await use_case.migrate_to_centralized(force=True)
|
||||
|
||||
assert summary["success"] is True
|
||||
assert summary["skipped"] == 1
|
||||
assert summary["moved"] == 0
|
||||
# Sidecar stays put when the model file is gone.
|
||||
assert sidecar.exists()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_migrate_empty_library_is_noop(
|
||||
library_root: Path, sidecar_root: Path
|
||||
):
|
||||
_set_mode("centralized")
|
||||
recorder = _ProgressRecorder()
|
||||
use_case = _make_use_case([])
|
||||
|
||||
summary = await use_case.migrate_to_centralized(recorder, force=True)
|
||||
|
||||
assert summary["success"] is True
|
||||
assert summary["models_total"] == 0
|
||||
assert summary["moved"] == 0
|
||||
statuses = [payload["status"] for payload in recorder.payloads]
|
||||
assert statuses == ["started", "completed"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_migrate_to_centralized_refuses_when_already_centralized(
|
||||
library_root: Path, sidecar_root: Path
|
||||
):
|
||||
_set_mode("centralized")
|
||||
use_case = _make_use_case([])
|
||||
|
||||
summary = await use_case.migrate_to_centralized()
|
||||
|
||||
assert summary["success"] is False
|
||||
assert "already centralized" in summary["error"]
|
||||
assert summary["moved"] == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_migrate_to_alongside_refuses_when_already_alongside(
|
||||
library_root: Path, sidecar_root: Path
|
||||
):
|
||||
_set_mode("alongside")
|
||||
use_case = _make_use_case([])
|
||||
|
||||
summary = await use_case.migrate_to_alongside()
|
||||
|
||||
assert summary["success"] is False
|
||||
assert "already alongside" in summary["error"]
|
||||
assert summary["moved"] == 0
|
||||
@@ -0,0 +1,250 @@
|
||||
"""Tests for py.utils.sidecar_paths (alongside + centralized storage modes)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from py.services.settings_manager import get_settings_manager
|
||||
from py.utils import sidecar_paths
|
||||
from py.utils.sidecar_paths import (
|
||||
METADATA_SUFFIX,
|
||||
get_configured_sidecar_root,
|
||||
get_metadata_path,
|
||||
get_preview_dir,
|
||||
get_sidecar_dir,
|
||||
get_sidecar_root,
|
||||
get_storage_mode,
|
||||
is_centralized,
|
||||
is_metadata_path,
|
||||
resolve_centralized_dir,
|
||||
resolve_centralized_dir_for_dir,
|
||||
resolve_metadata_path,
|
||||
sanitize_path_component,
|
||||
)
|
||||
|
||||
|
||||
def _normalize(path: Path) -> str:
|
||||
return str(path).replace(os.sep, "/")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def model_roots(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict:
|
||||
"""Point every config model root the sidecar module reads at tmp_path."""
|
||||
|
||||
from py.config import config
|
||||
|
||||
loras = tmp_path / "loras"
|
||||
checkpoints = tmp_path / "checkpoints"
|
||||
loras.mkdir()
|
||||
checkpoints.mkdir()
|
||||
|
||||
for attr, value in (
|
||||
("loras_roots", [str(loras)]),
|
||||
("base_models_roots", [str(checkpoints)]),
|
||||
("embeddings_roots", []),
|
||||
("other_roots", []),
|
||||
("extra_loras_roots", []),
|
||||
("extra_checkpoints_roots", []),
|
||||
("extra_unet_roots", []),
|
||||
("extra_embeddings_roots", []),
|
||||
):
|
||||
monkeypatch.setattr(config, attr, value, raising=False)
|
||||
|
||||
return {"loras": loras, "checkpoints": checkpoints}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def centralized(model_roots: dict, tmp_path: Path) -> Path:
|
||||
"""Enable centralized mode rooted at tmp_path/sidecars; returns the root."""
|
||||
|
||||
sidecar_root = tmp_path / "sidecars"
|
||||
settings = get_settings_manager()
|
||||
settings.set("sidecar_storage_mode", "centralized")
|
||||
settings.set("sidecar_storage_path", str(sidecar_root))
|
||||
return sidecar_root
|
||||
|
||||
|
||||
class TestAlongsideMode:
|
||||
def test_default_mode_is_alongside(self):
|
||||
assert get_storage_mode() == "alongside"
|
||||
assert not is_centralized()
|
||||
assert get_sidecar_root() == ""
|
||||
|
||||
def test_metadata_path_next_to_model(self, tmp_path: Path):
|
||||
model = tmp_path / "sub" / "model.safetensors"
|
||||
assert get_metadata_path(str(model)) == os.path.join(
|
||||
str(tmp_path), "sub", "model" + METADATA_SUFFIX
|
||||
)
|
||||
|
||||
def test_preview_and_sidecar_dir_are_model_dir(self, tmp_path: Path):
|
||||
model = tmp_path / "sub" / "model.safetensors"
|
||||
expected = os.path.dirname(os.path.abspath(str(model)))
|
||||
assert get_sidecar_dir(str(model)) == expected
|
||||
assert get_preview_dir(str(model)) == expected
|
||||
|
||||
|
||||
class TestPathPredicates:
|
||||
def test_is_metadata_path(self):
|
||||
assert is_metadata_path("/x/model.metadata.json")
|
||||
assert not is_metadata_path("/x/model.safetensors")
|
||||
assert not is_metadata_path("/x/model.metadata.json.bak")
|
||||
|
||||
def test_resolve_metadata_path_passthrough_for_sidecar(self):
|
||||
sidecar = "/x/model.metadata.json"
|
||||
assert resolve_metadata_path(sidecar) == sidecar
|
||||
|
||||
def test_resolve_metadata_path_derives_for_model(self, tmp_path: Path):
|
||||
model = tmp_path / "model.safetensors"
|
||||
assert resolve_metadata_path(str(model)) == get_metadata_path(str(model))
|
||||
|
||||
|
||||
class TestSanitizePathComponent:
|
||||
def test_special_characters_replaced(self):
|
||||
assert sanitize_path_component("foo bar/baz:qux") == "foo_bar_baz_qux"
|
||||
|
||||
def test_safe_characters_kept(self):
|
||||
assert sanitize_path_component("Flux-1.dev_v2") == "Flux-1.dev_v2"
|
||||
|
||||
def test_empty_falls_back_to_underscore(self):
|
||||
assert sanitize_path_component("") == "_"
|
||||
assert sanitize_path_component(None) == "_"
|
||||
|
||||
|
||||
class TestCentralizedMode:
|
||||
def test_mirror_layout(self, model_roots: dict, centralized: Path):
|
||||
model = model_roots["loras"] / "styles" / "anime" / "model.safetensors"
|
||||
library = get_settings_manager().get_active_library_name()
|
||||
|
||||
metadata_path = get_metadata_path(str(model))
|
||||
|
||||
expected = os.path.join(
|
||||
str(centralized), library, "loras", "styles", "anime", "model" + METADATA_SUFFIX
|
||||
)
|
||||
assert metadata_path == expected
|
||||
assert get_preview_dir(str(model)) == os.path.dirname(expected)
|
||||
assert is_centralized()
|
||||
|
||||
def test_longest_root_wins(self, model_roots: dict, centralized: Path, monkeypatch: pytest.MonkeyPatch):
|
||||
from py.config import config
|
||||
|
||||
nested = model_roots["loras"] / "nested"
|
||||
nested.mkdir()
|
||||
monkeypatch.setattr(
|
||||
config,
|
||||
"loras_roots",
|
||||
[str(model_roots["loras"]), str(nested)],
|
||||
raising=False,
|
||||
)
|
||||
library = get_settings_manager().get_active_library_name()
|
||||
|
||||
model = nested / "model.safetensors"
|
||||
assert get_metadata_path(str(model)) == os.path.join(
|
||||
str(centralized), library, "nested", "model" + METADATA_SUFFIX
|
||||
)
|
||||
|
||||
def test_outside_roots_falls_back_to_alongside(
|
||||
self, model_roots: dict, centralized: Path, tmp_path: Path
|
||||
):
|
||||
outside = tmp_path / "elsewhere" / "model.safetensors"
|
||||
|
||||
assert resolve_centralized_dir(str(outside)) is None
|
||||
assert get_sidecar_dir(str(outside)) == os.path.dirname(
|
||||
os.path.abspath(str(outside))
|
||||
)
|
||||
assert get_metadata_path(str(outside)) == os.path.join(
|
||||
str(tmp_path), "elsewhere", "model" + METADATA_SUFFIX
|
||||
)
|
||||
|
||||
def test_resolve_centralized_dir_for_dir_matches_model_resolution(
|
||||
self, model_roots: dict, centralized: Path
|
||||
):
|
||||
model_dir = model_roots["checkpoints"] / "sub"
|
||||
model = model_dir / "model.safetensors"
|
||||
|
||||
assert resolve_centralized_dir_for_dir(str(model_dir)) == resolve_centralized_dir(
|
||||
str(model)
|
||||
)
|
||||
|
||||
def test_resolve_centralized_dir_for_dir_root_maps_to_mirror_base(
|
||||
self, model_roots: dict, centralized: Path
|
||||
):
|
||||
library = get_settings_manager().get_active_library_name()
|
||||
|
||||
assert resolve_centralized_dir_for_dir(str(model_roots["loras"])) == os.path.join(
|
||||
str(centralized), library, "loras"
|
||||
)
|
||||
|
||||
def test_empty_path_uses_default_sidecar_root(self, model_roots: dict, tmp_path: Path):
|
||||
settings = get_settings_manager()
|
||||
settings.set("sidecar_storage_mode", "centralized")
|
||||
settings.set("sidecar_storage_path", "")
|
||||
|
||||
root = get_sidecar_root()
|
||||
assert root
|
||||
assert root.endswith(os.sep + "sidecars")
|
||||
assert is_centralized()
|
||||
|
||||
|
||||
class TestModeIndependentResolution:
|
||||
"""Migration tooling resolves the mirror layout regardless of active mode."""
|
||||
|
||||
def test_configured_root_resolves_in_alongside_mode(self, tmp_path: Path):
|
||||
sidecar_root = tmp_path / "sidecars"
|
||||
settings = get_settings_manager()
|
||||
settings.set("sidecar_storage_mode", "alongside")
|
||||
settings.set("sidecar_storage_path", str(sidecar_root))
|
||||
|
||||
assert get_sidecar_root() == ""
|
||||
assert get_configured_sidecar_root() == os.path.abspath(str(sidecar_root))
|
||||
|
||||
def test_configured_root_defaults_to_settings_dir(self):
|
||||
settings = get_settings_manager()
|
||||
settings.set("sidecar_storage_mode", "alongside")
|
||||
settings.set("sidecar_storage_path", "")
|
||||
|
||||
root = get_configured_sidecar_root()
|
||||
assert root
|
||||
assert root.endswith(os.sep + "sidecars")
|
||||
|
||||
def test_resolve_centralized_dir_for_dir_with_explicit_root(
|
||||
self, model_roots: dict, tmp_path: Path
|
||||
):
|
||||
sidecar_root = tmp_path / "sidecars"
|
||||
settings = get_settings_manager()
|
||||
settings.set("sidecar_storage_mode", "alongside")
|
||||
library = settings.get_active_library_name()
|
||||
|
||||
model_dir = model_roots["loras"] / "sub"
|
||||
|
||||
# Alongside mode: no root resolves without the override.
|
||||
assert resolve_centralized_dir_for_dir(str(model_dir)) is None
|
||||
assert resolve_centralized_dir_for_dir(
|
||||
str(model_dir), sidecar_root=str(sidecar_root)
|
||||
) == os.path.join(str(sidecar_root), library, "loras", "sub")
|
||||
|
||||
|
||||
class TestSettingsValidation:
|
||||
def test_invalid_mode_falls_back_to_alongside(self):
|
||||
settings = get_settings_manager()
|
||||
settings.set("sidecar_storage_mode", "bogus")
|
||||
assert settings.get("sidecar_storage_mode") == "alongside"
|
||||
|
||||
def test_mode_is_normalized(self):
|
||||
settings = get_settings_manager()
|
||||
settings.set("sidecar_storage_mode", " Centralized ")
|
||||
assert settings.get("sidecar_storage_mode") == "centralized"
|
||||
|
||||
def test_path_is_normalized_to_absolute(self, tmp_path: Path):
|
||||
settings = get_settings_manager()
|
||||
settings.set("sidecar_storage_path", str(tmp_path / "sidecars"))
|
||||
assert settings.get("sidecar_storage_path") == os.path.abspath(
|
||||
str(tmp_path / "sidecars")
|
||||
)
|
||||
|
||||
def test_non_string_path_becomes_empty(self):
|
||||
settings = get_settings_manager()
|
||||
settings.set("sidecar_storage_path", None)
|
||||
assert settings.get("sidecar_storage_path") == ""
|
||||
Reference in New Issue
Block a user