fix(lora): read LoraManager's settings and storage layout the way it does (#355)
Add py/lora_manager_layout.py: the settings file in LoraManager's lookup order (environment folder, portable plugin file, user config folder), only the seven keys the import needs, the LoRA roots, the centralized sidecar root, the example-images root and the active library, each folder through one guard. Sidecar discovery for both storage modes and the local preview and example-image lookup live there too. The test runner and the browser test server point LoraManager's settings lookup at a disposable folder, so no test reads the real settings.
This commit is contained in:
Vito Sansevero
@@ -0,0 +1,599 @@
|
||||
"""Where LoraManager keeps its files (#355).
|
||||
|
||||
LoraManager (``custom_nodes/comfyui-lora-manager``) stores one
|
||||
``<name>.metadata.json`` sidecar per model, a preview image or video beside
|
||||
it, and each model's downloaded CivitAI example images. Where it stores them
|
||||
is a user setting, and its settings file itself can live in three places.
|
||||
This module repeats LoraManager's own rules, reading its files only (never
|
||||
importing it), so the LoRA import, the trigger-word cache and the image
|
||||
server share one definition:
|
||||
|
||||
- ``load_layout`` finds the settings file (environment folder, portable
|
||||
plugin file, or the user config folder, in LoraManager's order), keeps only
|
||||
the seven keys the import needs, and resolves the LoRA roots, the sidecar
|
||||
root (centralized storage), the example-images root and the active
|
||||
LoraManager library. Every folder from the settings passes the folder
|
||||
guard (exists, not a filesystem root, not the home folder itself).
|
||||
- ``discover_sidecars`` finds every sidecar whose model file exists inside a
|
||||
LoRA root: beside the models in ``alongside`` mode, under the sidecar root
|
||||
(matched to its model by ``file_path``) in ``centralized`` mode.
|
||||
- ``local_media`` lists a LoRA's preview and example images that the image
|
||||
server may serve.
|
||||
|
||||
Standard library plus ``platformdirs`` only (LoraManager depends on it), so
|
||||
it is testable outside ComfyUI.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
from dataclasses import dataclass, replace
|
||||
from pathlib import Path
|
||||
from typing import (
|
||||
AbstractSet,
|
||||
Callable,
|
||||
Dict,
|
||||
Iterator,
|
||||
List,
|
||||
Mapping,
|
||||
Optional,
|
||||
Sequence,
|
||||
Tuple,
|
||||
)
|
||||
|
||||
# py/utils/settings_paths.py: SETTINGS_DIR_ENV, _LM_PORTABLE_ENV, APP_NAME.
|
||||
SETTINGS_DIR_ENV = "LORA_MANAGER_SETTINGS_DIR"
|
||||
PORTABLE_ENV = "LORA_MANAGER_PORTABLE"
|
||||
APP_NAME = "ComfyUI-LoRA-Manager"
|
||||
# py/utils/constants.py PREVIEW_EXTENSIONS, in LoraManager's order.
|
||||
PREVIEW_EXTENSIONS = (
|
||||
".webp",
|
||||
".preview.webp",
|
||||
".preview.png",
|
||||
".preview.jpeg",
|
||||
".preview.jpg",
|
||||
".preview.mp4",
|
||||
".png",
|
||||
".jpeg",
|
||||
".jpg",
|
||||
".mp4",
|
||||
".gif",
|
||||
".webm",
|
||||
".avif",
|
||||
".jxl",
|
||||
)
|
||||
# LoraManager's LoRA scanner accepts .safetensors only (py/services/lora_scanner.py).
|
||||
MODEL_SUFFIX = ".safetensors"
|
||||
# py/utils/sidecar_paths.py METADATA_SUFFIX.
|
||||
METADATA_SUFFIX = ".metadata.json"
|
||||
MAX_SETTINGS_BYTES = 5 * 1024 * 1024
|
||||
# The only settings the import reads; every other key (the file can hold API
|
||||
# keys) is dropped right after parsing.
|
||||
SETTINGS_KEYS = frozenset(
|
||||
{
|
||||
"sidecar_storage_mode",
|
||||
"sidecar_storage_path",
|
||||
"example_images_path",
|
||||
"folder_paths",
|
||||
"extra_folder_paths",
|
||||
"libraries",
|
||||
"active_library",
|
||||
}
|
||||
)
|
||||
# py/utils/sidecar_paths.py STORAGE_MODE_ALONGSIDE / STORAGE_MODE_CENTRALIZED.
|
||||
MODE_ALONGSIDE = "alongside"
|
||||
MODE_CENTRALIZED = "centralized"
|
||||
# utils/video_metadata.py VIDEO_EXTENSIONS (not imported: that package needs
|
||||
# the node's relative imports).
|
||||
VIDEO_SUFFIXES = frozenset({".mp4", ".mov", ".m4v", ".webm", ".mkv", ".avi", ".wmv"})
|
||||
|
||||
SOURCE_ENVIRONMENT = "environment"
|
||||
SOURCE_PORTABLE = "portable"
|
||||
SOURCE_USER_CONFIG = "user config"
|
||||
|
||||
NO_INSTALL = "no LoraManager install found"
|
||||
UNREADABLE = "the settings file could not be read"
|
||||
NOT_JSON = "the settings file is not valid JSON"
|
||||
NOT_OBJECT = "the settings file is not a JSON object"
|
||||
TOO_LARGE = "the settings file is larger than 5 MB"
|
||||
NO_PLATFORMDIRS = "platformdirs is not available"
|
||||
NO_SIDECAR_ROOT = "the sidecar folder does not exist"
|
||||
BAD_SIDECAR_ROOT = "the sidecar folder is a filesystem root or the home folder"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LoraManagerLayout:
|
||||
"""What LoraManager's settings say about where its files are."""
|
||||
|
||||
settings_path: Optional[Path]
|
||||
source: Optional[str]
|
||||
error: Optional[str]
|
||||
mode: str
|
||||
lora_roots: Tuple[str, ...]
|
||||
sidecar_root: Optional[str]
|
||||
example_images_root: Optional[str]
|
||||
library: str
|
||||
media_roots: Tuple[str, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Sidecar:
|
||||
"""One LoRA's metadata file and the model file it describes."""
|
||||
|
||||
metadata_path: Path
|
||||
model_path: Path
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Discovery:
|
||||
"""The sidecars kept, how many metadata files were seen, and those unread."""
|
||||
|
||||
layout: LoraManagerLayout
|
||||
sidecars: List[Sidecar]
|
||||
scanned: int
|
||||
unreadable: List[str]
|
||||
|
||||
|
||||
class _SettingsError(Exception):
|
||||
"""The settings file cannot be used, for the reason it carries."""
|
||||
|
||||
|
||||
# ── paths ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _canonical(path: str, pathmod=os.path) -> str:
|
||||
# py/config.py _canonical_path (not imported: py.config imports ComfyUI's
|
||||
# ``server`` at import time).
|
||||
return pathmod.normcase(pathmod.realpath(path))
|
||||
|
||||
|
||||
def _is_filesystem_root(path: str, pathmod=os.path) -> bool:
|
||||
# py/config.py _is_filesystem_root: '/', 'C:\\' and UNC share roots.
|
||||
return pathmod.dirname(path) == path
|
||||
|
||||
|
||||
def _guard_folder(
|
||||
path: str,
|
||||
home: str,
|
||||
*,
|
||||
pathmod=os.path,
|
||||
isdir: Optional[Callable[[str], bool]] = None,
|
||||
) -> Optional[str]:
|
||||
"""The canonical form of ``path`` when it may be read and served, else None.
|
||||
|
||||
Kept only when it is a folder, not a filesystem root and not the home
|
||||
folder itself.
|
||||
"""
|
||||
isdir = isdir or os.path.isdir
|
||||
try:
|
||||
canonical = _canonical(path, pathmod)
|
||||
if not isdir(canonical) or _is_filesystem_root(canonical, pathmod):
|
||||
return None
|
||||
if canonical == _canonical(home, pathmod):
|
||||
return None
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
return canonical
|
||||
|
||||
|
||||
def _inside(path, roots: Sequence[str], pathmod=os.path) -> bool:
|
||||
"""True when ``path`` (canonical) is one of ``roots`` or below one."""
|
||||
try:
|
||||
candidate = _canonical(str(path), pathmod)
|
||||
except (OSError, ValueError):
|
||||
return False
|
||||
for root in roots:
|
||||
prefix = root if root.endswith(pathmod.sep) else root + pathmod.sep
|
||||
if candidate == root or candidate.startswith(prefix):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _absolute(value) -> Optional[str]:
|
||||
"""A settings path, stripped, ``~``-expanded and absolute; None when unset."""
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
return None
|
||||
return os.path.abspath(os.path.expanduser(value.strip()))
|
||||
|
||||
|
||||
# ── the settings file ───────────────────────────────────────────────
|
||||
|
||||
|
||||
def _platform_user_config_dir() -> str:
|
||||
from platformdirs import user_config_dir
|
||||
|
||||
return user_config_dir(APP_NAME, appauthor=False)
|
||||
|
||||
|
||||
def _read_bytes(path: Path) -> bytes:
|
||||
with open(path, "rb") as handle:
|
||||
return handle.read(MAX_SETTINGS_BYTES + 1)
|
||||
|
||||
|
||||
def _read_settings(path: Path) -> Optional[Dict]:
|
||||
"""The parsed object in ``path``; None when it does not exist."""
|
||||
try:
|
||||
raw = _read_bytes(path)
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
except OSError:
|
||||
raise _SettingsError(UNREADABLE) from None
|
||||
if len(raw) > MAX_SETTINGS_BYTES:
|
||||
raise _SettingsError(TOO_LARGE)
|
||||
try:
|
||||
data = json.loads(raw.decode("utf-8"))
|
||||
except (UnicodeDecodeError, ValueError):
|
||||
raise _SettingsError(NOT_JSON) from None
|
||||
if not isinstance(data, dict):
|
||||
raise _SettingsError(NOT_OBJECT)
|
||||
return data
|
||||
|
||||
|
||||
def _settings_subset(data: Mapping) -> Dict:
|
||||
"""Only the keys the import reads."""
|
||||
return {key: data[key] for key in SETTINGS_KEYS if key in data}
|
||||
|
||||
|
||||
def _portable_flag(environ: Mapping[str, str], plugin_settings: Path) -> bool:
|
||||
"""LoraManager's portable mode (settings_paths._should_use_portable_settings)."""
|
||||
forced = environ.get(PORTABLE_ENV)
|
||||
if forced == "1":
|
||||
return True
|
||||
if forced == "0":
|
||||
return False
|
||||
try:
|
||||
data = _read_settings(plugin_settings)
|
||||
except _SettingsError:
|
||||
return False
|
||||
return bool(data) and data.get("use_portable_settings") is True
|
||||
|
||||
|
||||
def _settings_location(
|
||||
lora_manager_path: str,
|
||||
environ: Mapping[str, str],
|
||||
user_config_dir: Callable[[], str],
|
||||
) -> Tuple[Path, str]:
|
||||
"""``(settings file, source)`` in LoraManager's lookup order."""
|
||||
override = environ.get(SETTINGS_DIR_ENV)
|
||||
if override:
|
||||
folder = os.path.abspath(os.path.expanduser(override))
|
||||
return Path(folder) / "settings.json", SOURCE_ENVIRONMENT
|
||||
plugin_settings = Path(lora_manager_path) / "settings.json"
|
||||
if _portable_flag(environ, plugin_settings):
|
||||
return plugin_settings, SOURCE_PORTABLE
|
||||
try:
|
||||
folder = user_config_dir()
|
||||
except ImportError:
|
||||
raise _SettingsError(NO_PLATFORMDIRS) from None
|
||||
return Path(folder) / "settings.json", SOURCE_USER_CONFIG
|
||||
|
||||
|
||||
# ── libraries and folders ───────────────────────────────────────────
|
||||
|
||||
|
||||
def _sanitize_library_name(name: str) -> str:
|
||||
# py/utils/example_images_paths.py sanitize_library_name.
|
||||
safe = re.sub(r"[^A-Za-z0-9_.-]", "_", name or "")
|
||||
return safe or "default"
|
||||
|
||||
|
||||
def _active_library(settings: Mapping) -> Tuple[str, Optional[Dict], int]:
|
||||
"""``(name, library dict or None, number of libraries)``."""
|
||||
libraries = settings.get("libraries")
|
||||
if not isinstance(libraries, dict) or not libraries:
|
||||
return "default", None, 0
|
||||
active = settings.get("active_library")
|
||||
if not isinstance(active, str) or active not in libraries:
|
||||
active = next(iter(libraries))
|
||||
library = libraries[active]
|
||||
return str(active), library if isinstance(library, dict) else None, len(libraries)
|
||||
|
||||
|
||||
def _lora_list(settings: Mapping, library: Optional[Dict], key: str) -> List[str]:
|
||||
"""The ``loras`` folders of ``key`` (the library's, else the top level)."""
|
||||
source = library if library is not None and key in library else settings
|
||||
section = source.get(key)
|
||||
if not isinstance(section, dict):
|
||||
return []
|
||||
value = section.get("loras")
|
||||
if isinstance(value, str):
|
||||
value = [value]
|
||||
if not isinstance(value, list):
|
||||
return []
|
||||
return [entry for entry in value if isinstance(entry, str) and entry.strip()]
|
||||
|
||||
|
||||
def _guarded(paths, home: str) -> Tuple[str, ...]:
|
||||
kept = []
|
||||
for path in paths:
|
||||
absolute = _absolute(path)
|
||||
guarded = _guard_folder(absolute, home) if absolute else None
|
||||
if guarded and guarded not in kept:
|
||||
kept.append(guarded)
|
||||
return tuple(kept)
|
||||
|
||||
|
||||
def _example_images_root(
|
||||
settings: Mapping, library: str, libraries: int, home: str
|
||||
) -> Optional[str]:
|
||||
"""py/utils/example_images_paths.py get_library_root, when the folder exists."""
|
||||
configured = _absolute(settings.get("example_images_path"))
|
||||
root = _guard_folder(configured, home) if configured else None
|
||||
if root is None or libraries <= 1:
|
||||
return root
|
||||
scoped = os.path.join(root, _sanitize_library_name(library))
|
||||
return scoped if os.path.isdir(scoped) else None
|
||||
|
||||
|
||||
def _media_roots(*groups) -> Tuple[str, ...]:
|
||||
roots = []
|
||||
for group in groups:
|
||||
for root in group:
|
||||
if root and root not in roots:
|
||||
roots.append(root)
|
||||
return tuple(roots)
|
||||
|
||||
|
||||
def load_layout(
|
||||
lora_manager_path: Optional[str],
|
||||
*,
|
||||
default_roots: Callable[[], List[str]],
|
||||
environ: Mapping[str, str] = os.environ,
|
||||
user_config_dir: Callable[[], str] = _platform_user_config_dir,
|
||||
) -> LoraManagerLayout:
|
||||
"""Read LoraManager's settings the way LoraManager does (read only).
|
||||
|
||||
``lora_manager_path`` is the resolved install (``None`` when none was
|
||||
found). An unusable layout carries ``error``, alongside mode and the
|
||||
default LoRA roots, so serving and trigger words keep working.
|
||||
"""
|
||||
home = str(Path.home())
|
||||
defaults = _guarded(default_roots(), home)
|
||||
fallback = LoraManagerLayout(
|
||||
settings_path=None,
|
||||
source=None,
|
||||
error=NO_INSTALL,
|
||||
mode=MODE_ALONGSIDE,
|
||||
lora_roots=defaults,
|
||||
sidecar_root=None,
|
||||
example_images_root=None,
|
||||
library="default",
|
||||
media_roots=defaults,
|
||||
)
|
||||
if not lora_manager_path:
|
||||
return fallback
|
||||
try:
|
||||
settings_path, source = _settings_location(
|
||||
lora_manager_path, environ, user_config_dir
|
||||
)
|
||||
except _SettingsError as error:
|
||||
return replace(fallback, error=str(error))
|
||||
fallback = replace(fallback, settings_path=settings_path, source=source)
|
||||
try:
|
||||
raw = _read_settings(settings_path)
|
||||
except _SettingsError as error:
|
||||
return replace(fallback, error=str(error))
|
||||
settings = _settings_subset(raw or {})
|
||||
return _layout_from_settings(settings, settings_path, source, defaults, home)
|
||||
|
||||
|
||||
def _layout_from_settings(settings, settings_path, source, defaults, home):
|
||||
library, entry, libraries = _active_library(settings)
|
||||
lora_roots = _media_roots(
|
||||
defaults,
|
||||
_guarded(_lora_list(settings, entry, "folder_paths"), home),
|
||||
_guarded(_lora_list(settings, entry, "extra_folder_paths"), home),
|
||||
)
|
||||
mode = settings.get("sidecar_storage_mode")
|
||||
mode = MODE_CENTRALIZED if mode == MODE_CENTRALIZED else MODE_ALONGSIDE
|
||||
example_root = _example_images_root(settings, library, libraries, home)
|
||||
layout = LoraManagerLayout(
|
||||
settings_path=settings_path,
|
||||
source=source,
|
||||
error=None,
|
||||
mode=mode,
|
||||
lora_roots=lora_roots,
|
||||
sidecar_root=None,
|
||||
example_images_root=example_root,
|
||||
library=library,
|
||||
media_roots=_media_roots(lora_roots, [example_root]),
|
||||
)
|
||||
if mode != MODE_CENTRALIZED:
|
||||
return layout
|
||||
configured = _absolute(settings.get("sidecar_storage_path"))
|
||||
configured = configured or os.path.join(str(settings_path.parent), "sidecars")
|
||||
sidecar_root = _guard_folder(configured, home)
|
||||
if sidecar_root is None:
|
||||
reason = NO_SIDECAR_ROOT if not os.path.isdir(configured) else BAD_SIDECAR_ROOT
|
||||
return replace(
|
||||
layout,
|
||||
error=reason,
|
||||
lora_roots=defaults,
|
||||
example_images_root=None,
|
||||
media_roots=defaults,
|
||||
)
|
||||
return replace(
|
||||
layout,
|
||||
sidecar_root=sidecar_root,
|
||||
media_roots=_media_roots(lora_roots, [sidecar_root], [example_root]),
|
||||
)
|
||||
|
||||
|
||||
# ── sidecars ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _metadata_files(folder: str) -> Iterator[Path]:
|
||||
"""Every ``*.metadata.json`` under ``folder``; an unlistable folder raises.
|
||||
|
||||
Symlinked folders are followed, each real folder once (no loops).
|
||||
"""
|
||||
|
||||
def fail(error):
|
||||
raise error
|
||||
|
||||
visited = set()
|
||||
for current, dirs, files in os.walk(folder, onerror=fail, followlinks=True):
|
||||
real = _canonical(current)
|
||||
if real in visited:
|
||||
dirs[:] = []
|
||||
continue
|
||||
visited.add(real)
|
||||
dirs.sort()
|
||||
for name in sorted(files):
|
||||
if name.endswith(METADATA_SUFFIX):
|
||||
yield Path(current) / name
|
||||
|
||||
|
||||
def _read_sidecar(path: Path) -> Optional[Dict]:
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as handle:
|
||||
data = json.load(handle)
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
return data if isinstance(data, dict) else None
|
||||
|
||||
|
||||
def _model_for(path: Path, layout: LoraManagerLayout, unreadable: List[str]):
|
||||
"""The model file a sidecar describes, when it exists inside a LoRA root."""
|
||||
if layout.mode == MODE_ALONGSIDE:
|
||||
base = path.name[: -len(METADATA_SUFFIX)]
|
||||
model = path.parent / f"{base}{MODEL_SUFFIX}"
|
||||
return model if model.is_file() else None
|
||||
metadata = _read_sidecar(path)
|
||||
if metadata is None:
|
||||
unreadable.append(path.name)
|
||||
return None
|
||||
file_path = metadata.get("file_path")
|
||||
if not isinstance(file_path, str) or not file_path.strip():
|
||||
return None
|
||||
model = Path(file_path)
|
||||
if not model.is_file() or not _inside(model, layout.lora_roots):
|
||||
return None
|
||||
return model
|
||||
|
||||
|
||||
def discover_sidecars(layout: LoraManagerLayout) -> Discovery:
|
||||
"""Every sidecar whose model exists inside a LoRA root, sorted, once each.
|
||||
|
||||
Raises ``OSError`` when a folder cannot be listed: a discovery that
|
||||
cannot see every folder must not replace the collection with part of it.
|
||||
"""
|
||||
if layout.error is not None:
|
||||
return Discovery(layout, [], 0, [])
|
||||
if layout.mode == MODE_CENTRALIZED:
|
||||
folders = [layout.sidecar_root]
|
||||
else:
|
||||
folders = list(layout.lora_roots)
|
||||
seen = set()
|
||||
scanned = 0
|
||||
unreadable: List[str] = []
|
||||
sidecars = []
|
||||
for folder in folders:
|
||||
for path in _metadata_files(folder):
|
||||
resolved = Path(os.path.realpath(path))
|
||||
key = os.path.normcase(str(resolved))
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
scanned += 1
|
||||
model = _model_for(resolved, layout, unreadable)
|
||||
if model is not None:
|
||||
sidecars.append(Sidecar(resolved, Path(os.path.realpath(model))))
|
||||
sidecars.sort(key=lambda sidecar: str(sidecar.metadata_path))
|
||||
return Discovery(layout, sidecars, scanned, unreadable)
|
||||
|
||||
|
||||
# ── images ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _natural_key(name: str):
|
||||
parts = re.split(r"(\d+)", name)
|
||||
return [int(part) if part.isdigit() else part.lower() for part in parts], name
|
||||
|
||||
|
||||
def _example_files(folder: str) -> List[str]:
|
||||
try:
|
||||
with os.scandir(folder) as entries:
|
||||
names = [entry.name for entry in entries if entry.is_file()]
|
||||
except OSError:
|
||||
return []
|
||||
return [os.path.join(folder, name) for name in sorted(names, key=_natural_key)]
|
||||
|
||||
|
||||
def _preview_candidates(sidecar: Sidecar, metadata: Dict) -> Iterator[str]:
|
||||
"""The paths a preview is looked for at, in LoraManager's order."""
|
||||
url = metadata.get("preview_url")
|
||||
if isinstance(url, str) and url.strip() and os.path.isabs(url.strip()):
|
||||
yield url.strip()
|
||||
base = sidecar.metadata_path.name[: -len(METADATA_SUFFIX)]
|
||||
for suffix in PREVIEW_EXTENSIONS:
|
||||
yield str(sidecar.metadata_path.parent / f"{base}{suffix}")
|
||||
|
||||
|
||||
def local_media(
|
||||
sidecar: Sidecar,
|
||||
metadata: Dict,
|
||||
layout: LoraManagerLayout,
|
||||
media_extensions: AbstractSet[str],
|
||||
) -> List[str]:
|
||||
"""The LoRA's preview, then its example images: absolute, resolved paths.
|
||||
|
||||
A file counts only when its extension is in ``media_extensions`` (what the
|
||||
image server serves) and it lies inside ``layout.media_roots``.
|
||||
"""
|
||||
|
||||
def usable(path: str) -> bool:
|
||||
return (
|
||||
os.path.splitext(path)[1].lower() in media_extensions
|
||||
and os.path.isfile(path)
|
||||
and _inside(path, layout.media_roots)
|
||||
)
|
||||
|
||||
found: List[str] = []
|
||||
for path in _preview_candidates(sidecar, metadata):
|
||||
if usable(path):
|
||||
found.append(path)
|
||||
break
|
||||
sha256 = metadata.get("sha256")
|
||||
if layout.example_images_root and isinstance(sha256, str) and sha256.strip():
|
||||
folder = os.path.join(layout.example_images_root, sha256.strip().lower())
|
||||
found.extend(path for path in _example_files(folder) if usable(path))
|
||||
media: List[str] = []
|
||||
for path in found:
|
||||
resolved = os.path.realpath(path)
|
||||
if resolved not in media:
|
||||
media.append(resolved)
|
||||
return _cover_first(media)
|
||||
|
||||
|
||||
def _cover_first(media: List[str]) -> List[str]:
|
||||
"""A still image in front of a leading video (no thumbnails for these yet)."""
|
||||
if not media or os.path.splitext(media[0])[1].lower() not in VIDEO_SUFFIXES:
|
||||
return media
|
||||
for index, path in enumerate(media):
|
||||
if os.path.splitext(path)[1].lower() not in VIDEO_SUFFIXES:
|
||||
return [path] + media[:index] + media[index + 1 :]
|
||||
return media
|
||||
|
||||
|
||||
# ── the review ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def layout_summary(layout: LoraManagerLayout, home: Path) -> Dict:
|
||||
"""What the review shows about the layout (no settings values)."""
|
||||
settings_file = None
|
||||
if layout.settings_path is not None:
|
||||
settings_file = str(layout.settings_path)
|
||||
try:
|
||||
relative = Path(settings_file).relative_to(home)
|
||||
settings_file = os.path.join("~", str(relative))
|
||||
except ValueError:
|
||||
pass
|
||||
return {
|
||||
"source": layout.source,
|
||||
"settings_file": settings_file,
|
||||
"error": layout.error,
|
||||
"mode": layout.mode,
|
||||
"library": layout.library,
|
||||
"example_images": layout.example_images_root is not None,
|
||||
}
|
||||
@@ -166,7 +166,20 @@ def register_reveal_stub(routes, api, root: Path):
|
||||
return types.SimpleNamespace(reset=reset)
|
||||
|
||||
|
||||
def isolate_lora_manager_settings(root: Path) -> Path:
|
||||
"""LoraManager's settings lookup reads an empty folder under ``root`` (#355).
|
||||
|
||||
The server's temporary root is removed when it exits, and with it this
|
||||
folder: no page test reads the developer's real LoraManager settings.
|
||||
"""
|
||||
folder = root / "lora-manager-settings"
|
||||
folder.mkdir(exist_ok=True)
|
||||
os.environ["LORA_MANAGER_SETTINGS_DIR"] = str(folder)
|
||||
return folder
|
||||
|
||||
|
||||
def create_app(root: Path) -> web.Application:
|
||||
isolate_lora_manager_settings(root)
|
||||
os.environ["PROMPT_MANAGER_CONFIG_PATH"] = str(root / "config.json")
|
||||
os.chdir(root)
|
||||
app = web.Application()
|
||||
|
||||
+29
-10
@@ -33,7 +33,9 @@ written next to it (TESTING.md "Time limits and hangs").
|
||||
import argparse
|
||||
import faulthandler
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
@@ -41,6 +43,9 @@ DEFAULT_PER_TEST_LIMIT = 120
|
||||
DEFAULT_SHUTDOWN_LIMIT = 120
|
||||
NO_TESTS_EXIT = 5
|
||||
USAGE_EXIT = 2
|
||||
# LoraManager's settings-folder override (#355): every test reads an empty,
|
||||
# disposable LoraManager settings folder, never the developer's real one.
|
||||
LORA_MANAGER_SETTINGS_ENV = "LORA_MANAGER_SETTINGS_DIR"
|
||||
|
||||
|
||||
class LimitError(ValueError):
|
||||
@@ -115,6 +120,16 @@ def parse_args(argv):
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def isolate_lora_manager_settings():
|
||||
"""Point LoraManager's settings lookup at a fresh temporary folder.
|
||||
|
||||
Returns the folder; the caller removes it when the run ends.
|
||||
"""
|
||||
folder = tempfile.mkdtemp(prefix="pm-lora-settings-")
|
||||
os.environ[LORA_MANAGER_SETTINGS_ENV] = folder
|
||||
return folder
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
args = parse_args(sys.argv[1:] if argv is None else argv)
|
||||
# Keep the real stderr: a test may replace sys.stderr, and the watchdog
|
||||
@@ -128,16 +143,20 @@ def main(argv=None):
|
||||
return USAGE_EXIT
|
||||
|
||||
faulthandler.dump_traceback_later(per_test_limit, exit=True, file=stream)
|
||||
start = os.path.abspath(args.start_directory)
|
||||
suite = unittest.TestLoader().discover(start, args.pattern, top_level_dir=start)
|
||||
runner = unittest.TextTestRunner(
|
||||
stream=stream,
|
||||
verbosity=args.verbosity,
|
||||
# As unittest.main does: show warnings unless -W/PYTHONWARNINGS set them.
|
||||
warnings=None if sys.warnoptions else "default",
|
||||
resultclass=make_result_class(stream, per_test_limit, shutdown_limit),
|
||||
)
|
||||
result = runner.run(suite)
|
||||
settings_folder = isolate_lora_manager_settings()
|
||||
try:
|
||||
start = os.path.abspath(args.start_directory)
|
||||
suite = unittest.TestLoader().discover(start, args.pattern, top_level_dir=start)
|
||||
runner = unittest.TextTestRunner(
|
||||
stream=stream,
|
||||
verbosity=args.verbosity,
|
||||
# As unittest.main does: show warnings unless -W/PYTHONWARNINGS set them.
|
||||
warnings=None if sys.warnoptions else "default",
|
||||
resultclass=make_result_class(stream, per_test_limit, shutdown_limit),
|
||||
)
|
||||
result = runner.run(suite)
|
||||
finally:
|
||||
shutil.rmtree(settings_folder, ignore_errors=True)
|
||||
if result.testsRun == 0 and not result.skipped:
|
||||
return NO_TESTS_EXIT
|
||||
return 0 if result.wasSuccessful() else 1
|
||||
|
||||
@@ -79,5 +79,21 @@ class TestBrowserServerLifetime(unittest.TestCase):
|
||||
process.wait(timeout=1)
|
||||
|
||||
|
||||
class TestBrowserServerLoraIsolation(unittest.TestCase):
|
||||
"""#355: page tests never read the developer's LoraManager settings."""
|
||||
|
||||
def test_settings_folder_is_inside_the_server_root(self):
|
||||
from unittest.mock import patch
|
||||
|
||||
from tests.browser.server import isolate_lora_manager_settings
|
||||
|
||||
with tempfile.TemporaryDirectory() as root:
|
||||
with patch.dict(os.environ, {"LORA_MANAGER_SETTINGS_DIR": "/real/one"}):
|
||||
folder = isolate_lora_manager_settings(Path(root))
|
||||
self.assertEqual(os.environ["LORA_MANAGER_SETTINGS_DIR"], str(folder))
|
||||
self.assertEqual(folder.parent, Path(root))
|
||||
self.assertTrue(folder.is_dir())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -0,0 +1,770 @@
|
||||
"""Where LoraManager keeps its files (#355): ``py/lora_manager_layout.py``.
|
||||
|
||||
Every test builds its own LoraManager install, settings file, LoRA roots,
|
||||
sidecar root and example-images folder in a temporary folder, and injects the
|
||||
environment and the user config folder: no test reads the developer's real
|
||||
LoraManager settings or opens the network.
|
||||
"""
|
||||
|
||||
import json
|
||||
import ntpath
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from py import lora_manager_layout as layout_module
|
||||
from py.lora_manager_layout import (
|
||||
METADATA_SUFFIX,
|
||||
PREVIEW_EXTENSIONS,
|
||||
Sidecar,
|
||||
discover_sidecars,
|
||||
layout_summary,
|
||||
load_layout,
|
||||
local_media,
|
||||
)
|
||||
|
||||
MEDIA = frozenset(
|
||||
{".png", ".jpg", ".jpeg", ".webp", ".gif", ".mp4", ".webm", ".mov", ".bmp"}
|
||||
)
|
||||
SECRET = "sk-civitai-must-never-leak-0123"
|
||||
|
||||
|
||||
class LayoutTestCase(unittest.TestCase):
|
||||
"""A disposable LoraManager install and the folders its settings name."""
|
||||
|
||||
def setUp(self):
|
||||
self.tmp = Path(tempfile.mkdtemp()).resolve()
|
||||
self.addCleanup(shutil.rmtree, self.tmp, True)
|
||||
self.plugin = self.tmp / "custom_nodes" / "ComfyUI-Lora-Manager"
|
||||
(self.plugin / "py").mkdir(parents=True)
|
||||
(self.plugin / "__init__.py").write_text("")
|
||||
self.user_config = self.tmp / "user-config" / "ComfyUI-LoRA-Manager"
|
||||
self.env_dir = self.tmp / "env-settings"
|
||||
self.default_loras = self.tmp / "models" / "loras"
|
||||
self.default_loras.mkdir(parents=True)
|
||||
self.environ = {}
|
||||
self.config_calls = []
|
||||
|
||||
def user_config_dir(self):
|
||||
self.config_calls.append(True)
|
||||
return str(self.user_config)
|
||||
|
||||
def write_settings(self, folder, data):
|
||||
folder.mkdir(parents=True, exist_ok=True)
|
||||
path = folder / "settings.json"
|
||||
path.write_text(json.dumps(data), encoding="utf-8")
|
||||
return path
|
||||
|
||||
def load(self, path="__plugin__", **kwargs):
|
||||
lora_manager = str(self.plugin) if path == "__plugin__" else path
|
||||
kwargs.setdefault("default_roots", lambda: [str(self.default_loras)])
|
||||
kwargs.setdefault("environ", self.environ)
|
||||
kwargs.setdefault("user_config_dir", self.user_config_dir)
|
||||
return load_layout(lora_manager, **kwargs)
|
||||
|
||||
def model(self, folder, stem):
|
||||
folder.mkdir(parents=True, exist_ok=True)
|
||||
path = folder / f"{stem}.safetensors"
|
||||
path.write_bytes(b"model")
|
||||
return path
|
||||
|
||||
def sidecar(self, folder, stem, data=None):
|
||||
folder.mkdir(parents=True, exist_ok=True)
|
||||
path = folder / f"{stem}{METADATA_SUFFIX}"
|
||||
path.write_text(json.dumps(data if data is not None else {}), "utf-8")
|
||||
return path
|
||||
|
||||
|
||||
class SettingsLookupTests(LayoutTestCase):
|
||||
|
||||
def test_environment_folder_wins_over_portable_and_user_config(self):
|
||||
env_file = self.write_settings(self.env_dir, {"sidecar_storage_mode": "x"})
|
||||
self.write_settings(self.plugin, {"use_portable_settings": True})
|
||||
self.write_settings(self.user_config, {})
|
||||
self.environ["LORA_MANAGER_SETTINGS_DIR"] = str(self.env_dir)
|
||||
|
||||
layout = self.load()
|
||||
|
||||
self.assertEqual(layout.source, "environment")
|
||||
self.assertEqual(layout.settings_path, env_file)
|
||||
self.assertIsNone(layout.error)
|
||||
self.assertEqual(self.config_calls, [])
|
||||
|
||||
def test_an_empty_environment_folder_is_ignored(self):
|
||||
self.environ["LORA_MANAGER_SETTINGS_DIR"] = ""
|
||||
self.assertEqual(self.load().source, "user config")
|
||||
|
||||
def test_portable_flag_in_the_plugin_settings(self):
|
||||
plugin_file = self.write_settings(self.plugin, {"use_portable_settings": True})
|
||||
layout = self.load()
|
||||
self.assertEqual(layout.source, "portable")
|
||||
self.assertEqual(layout.settings_path, plugin_file)
|
||||
self.assertEqual(self.config_calls, [])
|
||||
|
||||
def test_portable_environment_one_turns_it_on(self):
|
||||
self.environ["LORA_MANAGER_PORTABLE"] = "1"
|
||||
layout = self.load()
|
||||
self.assertEqual(layout.source, "portable")
|
||||
self.assertEqual(layout.settings_path, self.plugin / "settings.json")
|
||||
|
||||
def test_portable_environment_zero_turns_it_off_despite_the_file(self):
|
||||
self.write_settings(self.plugin, {"use_portable_settings": True})
|
||||
self.environ["LORA_MANAGER_PORTABLE"] = "0"
|
||||
self.assertEqual(self.load().source, "user config")
|
||||
|
||||
def test_other_portable_environment_values_are_ignored(self):
|
||||
for value in ("true", "yes"):
|
||||
with self.subTest(value=value):
|
||||
self.environ["LORA_MANAGER_PORTABLE"] = value
|
||||
self.assertEqual(self.load().source, "user config")
|
||||
self.write_settings(self.plugin, {"use_portable_settings": True})
|
||||
self.assertEqual(self.load().source, "portable")
|
||||
(self.plugin / "settings.json").unlink()
|
||||
|
||||
def test_a_string_portable_flag_is_ignored(self):
|
||||
self.write_settings(self.plugin, {"use_portable_settings": "true"})
|
||||
self.assertEqual(self.load().source, "user config")
|
||||
|
||||
def test_an_unreadable_plugin_file_means_not_portable(self):
|
||||
(self.plugin / "settings.json").write_text("{not json")
|
||||
self.assertEqual(self.load().source, "user config")
|
||||
|
||||
def test_user_config_folder_is_the_injected_one(self):
|
||||
user_file = self.write_settings(self.user_config, {})
|
||||
layout = self.load()
|
||||
self.assertEqual(layout.source, "user config")
|
||||
self.assertEqual(layout.settings_path, user_file)
|
||||
self.assertEqual(self.config_calls, [True])
|
||||
|
||||
def test_the_real_user_config_folder_is_never_opened(self):
|
||||
opened = []
|
||||
real_open = open
|
||||
|
||||
def spy(path, *args, **kwargs):
|
||||
opened.append(os.path.realpath(str(path)))
|
||||
return real_open(path, *args, **kwargs)
|
||||
|
||||
with patch("builtins.open", spy):
|
||||
layout = self.load()
|
||||
self.assertEqual(layout.settings_path, self.user_config / "settings.json")
|
||||
for path in opened:
|
||||
self.assertTrue(path.startswith(str(self.tmp)), path)
|
||||
|
||||
def test_a_missing_settings_file_means_defaults(self):
|
||||
layout = self.load()
|
||||
self.assertIsNone(layout.error)
|
||||
self.assertEqual(layout.mode, "alongside")
|
||||
self.assertIsNone(layout.example_images_root)
|
||||
self.assertIsNone(layout.sidecar_root)
|
||||
self.assertEqual(layout.library, "default")
|
||||
self.assertEqual(
|
||||
layout.lora_roots, (os.path.normcase(str(self.default_loras)),)
|
||||
)
|
||||
|
||||
def test_settings_dir_is_expanded_and_made_absolute(self):
|
||||
self.write_settings(self.env_dir, {})
|
||||
self.environ["LORA_MANAGER_SETTINGS_DIR"] = os.path.relpath(self.env_dir)
|
||||
self.assertEqual(self.load().settings_path, self.env_dir / "settings.json")
|
||||
|
||||
|
||||
class UnusableLayoutTests(LayoutTestCase):
|
||||
|
||||
def test_no_install(self):
|
||||
layout = self.load(path=None)
|
||||
self.assertEqual(layout.error, "no LoraManager install found")
|
||||
self.assertIsNone(layout.settings_path)
|
||||
self.assertIsNone(layout.source)
|
||||
# The default LoRA roots stay known for serving and trigger words.
|
||||
self.assertEqual(
|
||||
layout.lora_roots, (os.path.normcase(str(self.default_loras)),)
|
||||
)
|
||||
self.assertEqual(layout.media_roots, layout.lora_roots)
|
||||
self.assertEqual(discover_sidecars(layout).sidecars, [])
|
||||
|
||||
def test_unreadable_settings(self):
|
||||
self.write_settings(self.user_config, {})
|
||||
with patch.object(layout_module, "_read_bytes", side_effect=PermissionError):
|
||||
layout = self.load()
|
||||
self.assertEqual(layout.error, "the settings file could not be read")
|
||||
|
||||
def test_settings_that_are_not_json(self):
|
||||
self.user_config.mkdir(parents=True)
|
||||
(self.user_config / "settings.json").write_text("{nope")
|
||||
self.assertEqual(self.load().error, "the settings file is not valid JSON")
|
||||
|
||||
def test_settings_that_are_not_utf8(self):
|
||||
self.user_config.mkdir(parents=True)
|
||||
(self.user_config / "settings.json").write_bytes(b'{"a": "\xff"}')
|
||||
self.assertEqual(self.load().error, "the settings file is not valid JSON")
|
||||
|
||||
def test_settings_that_are_not_an_object(self):
|
||||
self.write_settings(self.user_config, ["a"])
|
||||
self.assertEqual(self.load().error, "the settings file is not a JSON object")
|
||||
|
||||
def test_oversized_settings(self):
|
||||
self.user_config.mkdir(parents=True)
|
||||
with open(self.user_config / "settings.json", "wb") as handle:
|
||||
handle.write(b" " * (layout_module.MAX_SETTINGS_BYTES + 1))
|
||||
self.assertEqual(self.load().error, "the settings file is larger than 5 MB")
|
||||
|
||||
def test_platformdirs_import_failure(self):
|
||||
def missing():
|
||||
raise ImportError("no platformdirs")
|
||||
|
||||
layout = self.load(user_config_dir=missing)
|
||||
self.assertEqual(layout.error, "platformdirs is not available")
|
||||
self.assertEqual(layout.mode, "alongside")
|
||||
|
||||
def test_default_user_config_dir_needs_platformdirs(self):
|
||||
with patch.dict(sys.modules, {"platformdirs": None}):
|
||||
with self.assertRaises(ImportError):
|
||||
layout_module._platform_user_config_dir()
|
||||
|
||||
def test_default_user_config_dir_uses_the_app_name(self):
|
||||
fake = type(sys)("platformdirs")
|
||||
calls = []
|
||||
fake.user_config_dir = lambda *a, **k: calls.append((a, k)) or "/x"
|
||||
with patch.dict(sys.modules, {"platformdirs": fake}):
|
||||
self.assertEqual(layout_module._platform_user_config_dir(), "/x")
|
||||
self.assertEqual(calls, [(("ComfyUI-LoRA-Manager",), {"appauthor": False})])
|
||||
|
||||
def test_missing_centralized_sidecar_root(self):
|
||||
self.write_settings(
|
||||
self.user_config,
|
||||
{
|
||||
"sidecar_storage_mode": "centralized",
|
||||
"sidecar_storage_path": str(self.tmp / "gone"),
|
||||
},
|
||||
)
|
||||
layout = self.load()
|
||||
self.assertEqual(layout.error, "the sidecar folder does not exist")
|
||||
self.assertEqual(layout.mode, "centralized")
|
||||
self.assertEqual(
|
||||
layout.lora_roots, (os.path.normcase(str(self.default_loras)),)
|
||||
)
|
||||
self.assertEqual(layout.media_roots, layout.lora_roots)
|
||||
self.assertEqual(discover_sidecars(layout).scanned, 0)
|
||||
|
||||
def test_a_centralized_sidecar_root_that_is_a_filesystem_root(self):
|
||||
self.write_settings(
|
||||
self.user_config,
|
||||
{"sidecar_storage_mode": "centralized", "sidecar_storage_path": "/"},
|
||||
)
|
||||
layout = self.load()
|
||||
self.assertEqual(
|
||||
layout.error, "the sidecar folder is a filesystem root or the home folder"
|
||||
)
|
||||
self.assertNotIn(os.path.normcase(os.path.realpath("/")), layout.media_roots)
|
||||
|
||||
|
||||
class SecretTests(LayoutTestCase):
|
||||
|
||||
def test_other_settings_keys_never_leave_the_module(self):
|
||||
self.write_settings(
|
||||
self.user_config,
|
||||
{"civitai_api_key": SECRET, "nested": {"token": SECRET}},
|
||||
)
|
||||
layout = self.load()
|
||||
self.assertNotIn(SECRET, repr(layout))
|
||||
self.assertNotIn(SECRET, json.dumps(layout_summary(layout, Path.home())))
|
||||
|
||||
def test_only_the_seven_keys_are_kept(self):
|
||||
self.assertEqual(
|
||||
layout_module.SETTINGS_KEYS,
|
||||
frozenset(
|
||||
{
|
||||
"sidecar_storage_mode",
|
||||
"sidecar_storage_path",
|
||||
"example_images_path",
|
||||
"folder_paths",
|
||||
"extra_folder_paths",
|
||||
"libraries",
|
||||
"active_library",
|
||||
}
|
||||
),
|
||||
)
|
||||
kept = layout_module._settings_subset(
|
||||
{"civitai_api_key": SECRET, "libraries": {}}
|
||||
)
|
||||
self.assertEqual(kept, {"libraries": {}})
|
||||
|
||||
|
||||
class ActiveLibraryTests(LayoutTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.top = self.tmp / "top-loras"
|
||||
self.lib_a = self.tmp / "a-loras"
|
||||
self.lib_b = self.tmp / "b-loras"
|
||||
self.extra = self.tmp / "b-extra"
|
||||
for folder in (self.top, self.lib_a, self.lib_b, self.extra):
|
||||
folder.mkdir()
|
||||
self.examples = self.tmp / "examples"
|
||||
self.examples.mkdir()
|
||||
|
||||
def settings(self, **extra):
|
||||
data = {
|
||||
"folder_paths": {"loras": [str(self.top)]},
|
||||
"libraries": {
|
||||
"first lib": {"folder_paths": {"loras": [str(self.lib_a)]}},
|
||||
"second": {
|
||||
"folder_paths": {"loras": [str(self.lib_b)]},
|
||||
"extra_folder_paths": {"loras": [str(self.extra)]},
|
||||
},
|
||||
},
|
||||
"example_images_path": str(self.examples),
|
||||
}
|
||||
data.update(extra)
|
||||
self.write_settings(self.user_config, data)
|
||||
|
||||
def test_the_active_library_lists_are_used(self):
|
||||
self.settings(active_library="second")
|
||||
(self.examples / "second").mkdir()
|
||||
layout = self.load()
|
||||
self.assertEqual(layout.library, "second")
|
||||
self.assertIn(os.path.normcase(str(self.lib_b)), layout.lora_roots)
|
||||
self.assertIn(os.path.normcase(str(self.extra)), layout.lora_roots)
|
||||
self.assertNotIn(os.path.normcase(str(self.top)), layout.lora_roots)
|
||||
self.assertNotIn(os.path.normcase(str(self.lib_a)), layout.lora_roots)
|
||||
self.assertIn(os.path.normcase(str(self.default_loras)), layout.lora_roots)
|
||||
self.assertEqual(
|
||||
layout.example_images_root, os.path.normcase(str(self.examples / "second"))
|
||||
)
|
||||
|
||||
def test_an_unknown_active_library_falls_back_to_the_first(self):
|
||||
self.settings(active_library="missing")
|
||||
(self.examples / "first_lib").mkdir()
|
||||
layout = self.load()
|
||||
self.assertEqual(layout.library, "first lib")
|
||||
self.assertIn(os.path.normcase(str(self.lib_a)), layout.lora_roots)
|
||||
# The sanitized library name is the example-images subfolder.
|
||||
self.assertEqual(
|
||||
layout.example_images_root,
|
||||
os.path.normcase(str(self.examples / "first_lib")),
|
||||
)
|
||||
|
||||
def test_no_libraries_means_default_and_the_top_level_lists(self):
|
||||
self.write_settings(
|
||||
self.user_config,
|
||||
{
|
||||
"folder_paths": {"loras": str(self.top)},
|
||||
"extra_folder_paths": {"loras": [str(self.extra), 7]},
|
||||
"example_images_path": str(self.examples),
|
||||
},
|
||||
)
|
||||
layout = self.load()
|
||||
self.assertEqual(layout.library, "default")
|
||||
self.assertIn(os.path.normcase(str(self.top)), layout.lora_roots)
|
||||
self.assertIn(os.path.normcase(str(self.extra)), layout.lora_roots)
|
||||
self.assertEqual(
|
||||
layout.example_images_root, os.path.normcase(str(self.examples))
|
||||
)
|
||||
|
||||
def test_one_library_has_no_library_subfolder(self):
|
||||
self.write_settings(
|
||||
self.user_config,
|
||||
{
|
||||
"libraries": {
|
||||
"comfyui": {"folder_paths": {"loras": [str(self.lib_a)]}}
|
||||
},
|
||||
"active_library": "comfyui",
|
||||
"example_images_path": str(self.examples),
|
||||
},
|
||||
)
|
||||
layout = self.load()
|
||||
self.assertEqual(layout.library, "comfyui")
|
||||
self.assertEqual(
|
||||
layout.example_images_root, os.path.normcase(str(self.examples))
|
||||
)
|
||||
|
||||
def test_a_missing_library_subfolder_means_no_example_images(self):
|
||||
self.settings(active_library="second")
|
||||
self.assertIsNone(self.load().example_images_root)
|
||||
|
||||
def test_an_unset_or_missing_example_folder_means_none(self):
|
||||
self.write_settings(self.user_config, {"example_images_path": " "})
|
||||
self.assertIsNone(self.load().example_images_root)
|
||||
self.write_settings(
|
||||
self.user_config, {"example_images_path": str(self.tmp / "gone")}
|
||||
)
|
||||
self.assertIsNone(self.load().example_images_root)
|
||||
|
||||
def test_sanitize_library_name(self):
|
||||
sanitize = layout_module._sanitize_library_name
|
||||
self.assertEqual(sanitize("My Lib/2"), "My_Lib_2")
|
||||
self.assertEqual(sanitize("ok-name_1.0"), "ok-name_1.0")
|
||||
self.assertEqual(sanitize(""), "default")
|
||||
|
||||
def test_odd_library_shapes_are_ignored(self):
|
||||
self.write_settings(
|
||||
self.user_config,
|
||||
{"libraries": {"x": "not a dict"}, "folder_paths": ["not", "a dict"]},
|
||||
)
|
||||
layout = self.load()
|
||||
self.assertEqual(layout.library, "x")
|
||||
self.assertEqual(
|
||||
layout.lora_roots, (os.path.normcase(str(self.default_loras)),)
|
||||
)
|
||||
|
||||
|
||||
class CentralizedDiscoveryTests(LayoutTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.loras = self.tmp / "loras"
|
||||
self.checkpoints = self.tmp / "checkpoints"
|
||||
self.checkpoints.mkdir()
|
||||
self.sidecars = self.tmp / "sidecars"
|
||||
self.sidecars.mkdir()
|
||||
self.write_settings(
|
||||
self.user_config,
|
||||
{
|
||||
"sidecar_storage_mode": "centralized",
|
||||
"sidecar_storage_path": str(self.sidecars),
|
||||
"folder_paths": {"loras": [str(self.loras)]},
|
||||
},
|
||||
)
|
||||
self.loras.mkdir()
|
||||
|
||||
def test_component_folders_and_the_legacy_nested_form(self):
|
||||
a = self.model(self.loras, "alpha")
|
||||
b = self.model(self.loras / "sub", "beta")
|
||||
self.sidecar(self.sidecars / "loras-1a2b3c4d", "alpha", {"file_path": str(a)})
|
||||
self.sidecar(
|
||||
self.sidecars / "comfyui" / "loras-1a2b3c4d" / "sub",
|
||||
"beta",
|
||||
{"file_path": b.as_posix()},
|
||||
)
|
||||
discovery = discover_sidecars(self.load())
|
||||
self.assertEqual(discovery.scanned, 2)
|
||||
self.assertEqual(
|
||||
sorted(s.model_path.name for s in discovery.sidecars),
|
||||
["alpha.safetensors", "beta.safetensors"],
|
||||
)
|
||||
self.assertEqual(discovery.unreadable, [])
|
||||
self.assertEqual(
|
||||
[str(s.metadata_path) for s in discovery.sidecars],
|
||||
sorted(str(s.metadata_path) for s in discovery.sidecars),
|
||||
)
|
||||
|
||||
def test_missing_models_and_checkpoints_are_skipped(self):
|
||||
ckpt = self.model(self.checkpoints, "base")
|
||||
self.sidecar(self.sidecars / "c", "base", {"file_path": str(ckpt)})
|
||||
self.sidecar(
|
||||
self.sidecars / "l",
|
||||
"gone",
|
||||
{"file_path": str(self.loras / "gone.safetensors")},
|
||||
)
|
||||
self.sidecar(self.sidecars / "l", "nopath", {"model_name": "x"})
|
||||
discovery = discover_sidecars(self.load())
|
||||
self.assertEqual(discovery.scanned, 3)
|
||||
self.assertEqual(discovery.sidecars, [])
|
||||
|
||||
def test_leftover_sidecars_beside_models_are_not_scanned(self):
|
||||
model = self.model(self.loras, "alpha")
|
||||
self.sidecar(self.loras, "alpha", {"file_path": str(model)})
|
||||
discovery = discover_sidecars(self.load())
|
||||
self.assertEqual((discovery.scanned, discovery.sidecars), (0, []))
|
||||
|
||||
def test_unparseable_sidecars_are_recorded(self):
|
||||
(self.sidecars / f"broken{METADATA_SUFFIX}").write_text("{{{")
|
||||
(self.sidecars / f"list{METADATA_SUFFIX}").write_text("[1]")
|
||||
discovery = discover_sidecars(self.load())
|
||||
self.assertEqual(discovery.scanned, 2)
|
||||
self.assertEqual(
|
||||
sorted(discovery.unreadable),
|
||||
[f"broken{METADATA_SUFFIX}", f"list{METADATA_SUFFIX}"],
|
||||
)
|
||||
|
||||
def test_default_sidecar_root_is_beside_the_settings(self):
|
||||
self.write_settings(
|
||||
self.user_config,
|
||||
{
|
||||
"sidecar_storage_mode": "centralized",
|
||||
"folder_paths": {"loras": [str(self.loras)]},
|
||||
},
|
||||
)
|
||||
(self.user_config / "sidecars").mkdir()
|
||||
layout = self.load()
|
||||
self.assertEqual(
|
||||
layout.sidecar_root, os.path.normcase(str(self.user_config / "sidecars"))
|
||||
)
|
||||
self.assertIn(layout.sidecar_root, layout.media_roots)
|
||||
|
||||
def test_an_unknown_mode_is_alongside(self):
|
||||
self.write_settings(self.user_config, {"sidecar_storage_mode": "elsewhere"})
|
||||
self.assertEqual(self.load().mode, "alongside")
|
||||
|
||||
def test_an_unlistable_folder_raises(self):
|
||||
layout = self.load()
|
||||
|
||||
def refuse(error):
|
||||
raise error
|
||||
|
||||
def failing_walk(top, onerror=None, followlinks=False):
|
||||
onerror(PermissionError(13, "denied", top))
|
||||
return iter(())
|
||||
|
||||
with patch.object(layout_module.os, "walk", failing_walk):
|
||||
with self.assertRaises(OSError):
|
||||
discover_sidecars(layout)
|
||||
|
||||
|
||||
class AlongsideDiscoveryTests(LayoutTestCase):
|
||||
|
||||
def test_a_sidecar_with_its_model_is_kept(self):
|
||||
model = self.model(self.default_loras, "alpha")
|
||||
meta = self.sidecar(self.default_loras, "alpha")
|
||||
self.sidecar(self.default_loras, "orphan")
|
||||
discovery = discover_sidecars(self.load())
|
||||
self.assertEqual(discovery.scanned, 2)
|
||||
self.assertEqual(
|
||||
discovery.sidecars,
|
||||
[Sidecar(Path(os.path.realpath(meta)), Path(os.path.realpath(model)))],
|
||||
)
|
||||
|
||||
def test_a_folder_reached_through_two_nested_roots_is_listed_once(self):
|
||||
nested = self.default_loras / "nested"
|
||||
self.model(nested, "alpha")
|
||||
self.sidecar(nested, "alpha")
|
||||
layout = self.load(default_roots=lambda: [str(self.default_loras), str(nested)])
|
||||
discovery = discover_sidecars(layout)
|
||||
self.assertEqual((discovery.scanned, len(discovery.sidecars)), (1, 1))
|
||||
|
||||
@unittest.skipUnless(hasattr(os, "symlink"), "symlinks")
|
||||
def test_a_symlink_loop_is_walked_once(self):
|
||||
self.model(self.default_loras, "alpha")
|
||||
self.sidecar(self.default_loras, "alpha")
|
||||
try:
|
||||
os.symlink(self.default_loras, self.default_loras / "loop")
|
||||
except (OSError, NotImplementedError):
|
||||
self.skipTest("symlinks need privileges here")
|
||||
discovery = discover_sidecars(self.load())
|
||||
self.assertEqual((discovery.scanned, len(discovery.sidecars)), (1, 1))
|
||||
|
||||
|
||||
class LocalMediaTests(LayoutTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.examples = self.tmp / "examples"
|
||||
self.examples.mkdir()
|
||||
self.write_settings(
|
||||
self.user_config, {"example_images_path": str(self.examples)}
|
||||
)
|
||||
self.model(self.default_loras, "alpha")
|
||||
self.meta = self.sidecar(self.default_loras, "alpha")
|
||||
self.layout = self.load()
|
||||
self.entry = discover_sidecars(self.layout).sidecars[0]
|
||||
|
||||
def media(self, metadata=None):
|
||||
return local_media(self.entry, metadata or {}, self.layout, MEDIA)
|
||||
|
||||
def touch(self, path):
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(b"x")
|
||||
return os.path.realpath(path)
|
||||
|
||||
def test_preview_url_comes_first(self):
|
||||
preview = self.touch(self.default_loras / "custom.jpeg")
|
||||
self.touch(self.default_loras / "alpha.webp")
|
||||
self.assertEqual(self.media({"preview_url": preview}), [preview])
|
||||
|
||||
def test_preview_url_outside_media_roots_or_not_servable_is_ignored(self):
|
||||
outside = self.touch(self.tmp / "elsewhere" / "p.png")
|
||||
avif = self.touch(self.default_loras / "p.avif")
|
||||
fallback = self.touch(self.default_loras / "alpha.png")
|
||||
for url in (
|
||||
outside,
|
||||
avif,
|
||||
"relative.png",
|
||||
7,
|
||||
str(self.default_loras / "x.png"),
|
||||
):
|
||||
with self.subTest(url=url):
|
||||
self.assertEqual(self.media({"preview_url": url}), [fallback])
|
||||
|
||||
def test_suffix_order_follows_lora_manager(self):
|
||||
self.assertEqual(
|
||||
PREVIEW_EXTENSIONS,
|
||||
(
|
||||
".webp",
|
||||
".preview.webp",
|
||||
".preview.png",
|
||||
".preview.jpeg",
|
||||
".preview.jpg",
|
||||
".preview.mp4",
|
||||
".png",
|
||||
".jpeg",
|
||||
".jpg",
|
||||
".mp4",
|
||||
".gif",
|
||||
".webm",
|
||||
".avif",
|
||||
".jxl",
|
||||
),
|
||||
)
|
||||
mp4 = self.touch(self.default_loras / "alpha.mp4")
|
||||
self.assertEqual(self.media(), [mp4])
|
||||
webp = self.touch(self.default_loras / "alpha.preview.webp")
|
||||
self.assertEqual(self.media(), [webp])
|
||||
|
||||
def test_an_avif_only_preview_gives_nothing(self):
|
||||
self.touch(self.default_loras / "alpha.avif")
|
||||
self.assertEqual(self.media(), [])
|
||||
|
||||
def test_example_images_in_numeric_order(self):
|
||||
folder = self.examples / "abcdef"
|
||||
names = ["image_10.png", "image_2.png", "image_1.jpg", "notes.txt"]
|
||||
for name in names:
|
||||
self.touch(folder / name)
|
||||
(folder / "sub").mkdir()
|
||||
preview = self.touch(self.default_loras / "alpha.png")
|
||||
found = self.media({"sha256": "ABCDEF"})
|
||||
self.assertEqual(
|
||||
[os.path.basename(p) for p in found],
|
||||
["alpha.png", "image_1.jpg", "image_2.png", "image_10.png"],
|
||||
)
|
||||
self.assertEqual(found[0], preview)
|
||||
|
||||
def test_missing_sha256_or_folder_gives_no_example_images(self):
|
||||
self.touch(self.examples / "abc" / "1.png")
|
||||
self.assertEqual(self.media(), [])
|
||||
self.assertEqual(self.media({"sha256": "zzz"}), [])
|
||||
self.assertEqual(self.media({"sha256": 5}), [])
|
||||
|
||||
def test_an_unlistable_example_folder_gives_what_was_found(self):
|
||||
self.touch(self.examples / "abc" / "1.png")
|
||||
with patch.object(layout_module.os, "scandir", side_effect=PermissionError):
|
||||
self.assertEqual(self.media({"sha256": "abc"}), [])
|
||||
|
||||
def test_cover_moves_the_first_still_in_front_of_a_video(self):
|
||||
video = self.touch(self.default_loras / "alpha.mp4")
|
||||
still = self.touch(self.examples / "abc" / "2.png")
|
||||
other = self.touch(self.examples / "abc" / "1.webm")
|
||||
self.assertEqual(self.media({"sha256": "abc"}), [still, video, other])
|
||||
|
||||
def test_videos_only_keep_their_order(self):
|
||||
video = self.touch(self.default_loras / "alpha.mp4")
|
||||
self.assertEqual(self.media(), [video])
|
||||
|
||||
def test_repeats_are_dropped(self):
|
||||
preview = self.touch(self.default_loras / "alpha.png")
|
||||
self.assertEqual(self.media({"preview_url": preview}), [preview])
|
||||
|
||||
def test_video_suffixes_match_the_video_module(self):
|
||||
from utils.video_metadata import VIDEO_EXTENSIONS
|
||||
|
||||
self.assertEqual(layout_module.VIDEO_SUFFIXES, frozenset(VIDEO_EXTENSIONS))
|
||||
|
||||
|
||||
class FolderGuardTests(unittest.TestCase):
|
||||
"""Portable path-flavor controls (TESTING.md Windows portability preflight)."""
|
||||
|
||||
def setUp(self):
|
||||
self.tmp = Path(tempfile.mkdtemp()).resolve()
|
||||
self.addCleanup(shutil.rmtree, self.tmp, True)
|
||||
|
||||
def guard_nt(self, path, home="C:\\Users\\Vito"):
|
||||
return layout_module._guard_folder(
|
||||
path, home, pathmod=ntpath, isdir=lambda _p: True
|
||||
)
|
||||
|
||||
def test_filesystem_roots_and_home_are_rejected(self):
|
||||
self.assertIsNone(layout_module._guard_folder("/", str(self.tmp)))
|
||||
self.assertIsNone(self.guard_nt("C:\\"))
|
||||
self.assertIsNone(self.guard_nt("c:/"))
|
||||
self.assertIsNone(self.guard_nt("\\\\server\\share\\"))
|
||||
self.assertIsNone(self.guard_nt("c:\\USERS\\vito"))
|
||||
self.assertIsNone(self.guard_nt("C:\\Users\\Vito\\"))
|
||||
self.assertEqual(
|
||||
self.guard_nt("C:\\Users\\Vito\\Loras"), "c:\\users\\vito\\loras"
|
||||
)
|
||||
self.assertIsNone(layout_module._guard_folder(str(self.tmp), str(self.tmp)))
|
||||
|
||||
def test_missing_folders_add_nothing(self):
|
||||
self.assertIsNone(layout_module._guard_folder(str(self.tmp / "x"), "/home"))
|
||||
self.assertIsNone(
|
||||
layout_module._guard_folder("\\\\server\\share\\loras", str(self.tmp))
|
||||
)
|
||||
self.assertEqual(
|
||||
layout_module._guard_folder(str(self.tmp), "/nonexistent-home"),
|
||||
os.path.normcase(str(self.tmp)),
|
||||
)
|
||||
|
||||
def test_windows_model_path_matches_a_lower_case_root(self):
|
||||
self.assertTrue(
|
||||
layout_module._inside(
|
||||
"C:/Models/Loras/A.safetensors", ("c:\\models\\loras",), ntpath
|
||||
)
|
||||
)
|
||||
self.assertFalse(
|
||||
layout_module._inside(
|
||||
"C:/Models/LorasX/A.safetensors", ("c:\\models\\loras",), ntpath
|
||||
)
|
||||
)
|
||||
self.assertFalse(
|
||||
layout_module._inside(
|
||||
"\\\\other\\share\\A.safetensors", ("c:\\models\\loras",), ntpath
|
||||
)
|
||||
)
|
||||
|
||||
def test_mixed_case_native_containment(self):
|
||||
inside = str(self.tmp / "Loras" / "a.png")
|
||||
root = os.path.normcase(str(self.tmp / "Loras"))
|
||||
self.assertTrue(layout_module._inside(inside, (root,)))
|
||||
self.assertFalse(layout_module._inside(str(self.tmp / "other.png"), (root,)))
|
||||
|
||||
|
||||
class SummaryTests(LayoutTestCase):
|
||||
|
||||
def test_summary_names_what_the_import_read(self):
|
||||
examples = self.tmp / "ex"
|
||||
examples.mkdir()
|
||||
sidecars = self.plugin / "sidecars"
|
||||
sidecars.mkdir()
|
||||
self.write_settings(
|
||||
self.plugin,
|
||||
{
|
||||
"use_portable_settings": True,
|
||||
"sidecar_storage_mode": "centralized",
|
||||
"example_images_path": str(examples),
|
||||
"libraries": {"comfyui": {}},
|
||||
},
|
||||
)
|
||||
layout = self.load()
|
||||
summary = layout_summary(layout, self.tmp)
|
||||
self.assertEqual(
|
||||
summary,
|
||||
{
|
||||
"source": "portable",
|
||||
"settings_file": os.path.join(
|
||||
"~", "custom_nodes", "ComfyUI-Lora-Manager", "settings.json"
|
||||
),
|
||||
"error": None,
|
||||
"mode": "centralized",
|
||||
"library": "comfyui",
|
||||
"example_images": True,
|
||||
},
|
||||
)
|
||||
|
||||
def test_summary_outside_home_and_without_settings(self):
|
||||
summary = layout_summary(self.load(path=None), self.tmp / "home")
|
||||
self.assertEqual(summary["settings_file"], None)
|
||||
self.assertEqual(summary["error"], "no LoraManager install found")
|
||||
self.assertFalse(summary["example_images"])
|
||||
layout = self.load()
|
||||
self.assertEqual(
|
||||
layout_summary(layout, self.tmp / "home")["settings_file"],
|
||||
str(self.user_config / "settings.json"),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -152,6 +152,30 @@ class RunSuiteTests(unittest.TestCase):
|
||||
result = run_suite(self.folder, None, "-p", "check_*.py")
|
||||
self.assertEqual(result.returncode, 1, result.stderr)
|
||||
|
||||
def test_lora_manager_settings_read_a_disposable_folder(self):
|
||||
# #355: no test reads the developer's real LoraManager settings.
|
||||
record = self.folder / "seen.txt"
|
||||
self.write(
|
||||
"test_probe_lora.py",
|
||||
f"""
|
||||
import os
|
||||
import unittest
|
||||
|
||||
class Probe(unittest.TestCase):
|
||||
def test_isolated(self):
|
||||
folder = os.environ["LORA_MANAGER_SETTINGS_DIR"]
|
||||
with open({str(record)!r}, "w", encoding="utf-8") as handle:
|
||||
handle.write(folder)
|
||||
self.assertTrue(os.path.isdir(folder))
|
||||
self.assertEqual(os.listdir(folder), [])
|
||||
""",
|
||||
)
|
||||
result = run_suite(self.folder, {"LORA_MANAGER_SETTINGS_DIR": "/real/one"})
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
folder = record.read_text(encoding="utf-8")
|
||||
self.assertNotEqual(folder, "/real/one")
|
||||
self.assertFalse(os.path.exists(folder), "removed when the run ends")
|
||||
|
||||
def test_empty_folder_exits_five(self):
|
||||
result = run_suite(self.folder)
|
||||
self.assertEqual(result.returncode, 5, result.stderr)
|
||||
|
||||
Reference in New Issue
Block a user