Compare commits
11
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
bac05711b9 | ||
|
|
d7ccd9b4df | ||
|
|
e2525a9d79 | ||
|
|
c47be3731c | ||
|
|
f65de3fea4 | ||
|
|
dead4a7aa3 | ||
|
|
9d1e5a0e22 | ||
|
|
ed4bd97005 | ||
|
|
ebebf0d69c | ||
|
|
9193373eac | ||
|
|
fba5903049 |
@@ -32,6 +32,7 @@ from .saved_views import SavedViewRoutesMixin
|
||||
from .history import HistoryRoutesMixin
|
||||
from .images import ImageRoutesMixin
|
||||
from .admin import AdminRoutesMixin
|
||||
from .settings_routes import SettingsRoutesMixin
|
||||
from .job_routes import JobRoutesMixin
|
||||
from .image_tag_jobs import ImageTagJobsMixin
|
||||
from .jobs import JobRegistry
|
||||
@@ -261,6 +262,7 @@ class PromptManagerAPI(
|
||||
PromptRoutesMixin,
|
||||
ImageRoutesMixin,
|
||||
AdminRoutesMixin,
|
||||
SettingsRoutesMixin,
|
||||
JobRoutesMixin,
|
||||
ImageTagJobsMixin,
|
||||
LoggingRoutesMixin,
|
||||
@@ -361,6 +363,24 @@ class PromptManagerAPI(
|
||||
"""409 envelope for a long job that is already running."""
|
||||
return _json_error(f"{label} already running", 409)
|
||||
|
||||
async def _run_held(self, kind, title, admission, operation):
|
||||
"""Run a legacy-page operation under a registry hold (D80).
|
||||
|
||||
While ``operation()`` runs, the registry lists it as a running,
|
||||
non-cancellable job, so workspace jobs its admission class conflicts
|
||||
with are refused, naming it. When the registry refuses the hold, the
|
||||
legacy 409 envelope names the running job and nothing is done. A
|
||||
response of 400 or above ends the hold ``failed``.
|
||||
"""
|
||||
busy = self._jobs.blocker(kind, admission=admission)
|
||||
if busy is not None:
|
||||
return self._long_job_busy_response(busy.title)
|
||||
async with self._jobs.hold(kind, title, admission=admission) as job:
|
||||
response = await operation()
|
||||
if getattr(response, "status", 200) >= 400:
|
||||
job.finish("failed")
|
||||
return response
|
||||
|
||||
async def _run_in_executor(self, func, *args, **kwargs):
|
||||
"""Run a blocking function in the default thread pool executor.
|
||||
|
||||
@@ -456,6 +476,7 @@ class PromptManagerAPI(
|
||||
self._register_prompt_routes(routes)
|
||||
self._register_image_routes(routes)
|
||||
self._register_admin_routes(routes)
|
||||
self._register_settings_routes(routes)
|
||||
self._register_job_routes(routes)
|
||||
self._register_logging_routes(routes)
|
||||
self._register_autotag_routes(routes)
|
||||
|
||||
+78
-231
@@ -22,16 +22,17 @@ from .jobs import (
|
||||
SCAN_GROUP_NAMES,
|
||||
SCAN_GROUPS,
|
||||
SCAN_SCOPES,
|
||||
DUPLICATE_CLEANUP_JOB_KIND,
|
||||
MAINTENANCE_JOB_KIND,
|
||||
RESTORE_JOB_KIND,
|
||||
Job,
|
||||
JobBusy,
|
||||
)
|
||||
|
||||
try:
|
||||
from ...utils.parallel import map_parallel
|
||||
from ...utils.validators import validate_result_timeout, validate_worker_threads
|
||||
except ImportError:
|
||||
from utils.parallel import map_parallel
|
||||
from utils.validators import validate_result_timeout, validate_worker_threads
|
||||
|
||||
try:
|
||||
from ...database.operations import PromptDatabase
|
||||
@@ -502,14 +503,6 @@ class AdminRoutesMixin:
|
||||
async def get_stats_route(request):
|
||||
return await self.get_statistics(request)
|
||||
|
||||
@routes.get("/prompt_manager/settings")
|
||||
async def get_settings_route(request):
|
||||
return await self.get_settings(request)
|
||||
|
||||
@routes.post("/prompt_manager/settings")
|
||||
async def save_settings_route(request):
|
||||
return await self.save_settings(request)
|
||||
|
||||
@routes.get("/prompt_manager/backup")
|
||||
async def backup_database_route(request):
|
||||
return await self.backup_database(request)
|
||||
@@ -651,7 +644,15 @@ class AdminRoutesMixin:
|
||||
job.finish()
|
||||
|
||||
async def cleanup_duplicates_endpoint(self, request):
|
||||
"""Cleanup duplicate prompts endpoint."""
|
||||
"""Cleanup duplicate prompts endpoint (holds the registry: records)."""
|
||||
return await self._run_held(
|
||||
MAINTENANCE_JOB_KIND,
|
||||
"Prompt cleanup (admin page)",
|
||||
"records",
|
||||
self._cleanup_duplicates,
|
||||
)
|
||||
|
||||
async def _cleanup_duplicates(self):
|
||||
try:
|
||||
removed_count = await self._run_in_executor(self.db.cleanup_duplicates)
|
||||
|
||||
@@ -766,26 +767,39 @@ class AdminRoutesMixin:
|
||||
return hash_sha256.hexdigest()
|
||||
|
||||
async def delete_duplicate_images_endpoint(self, request):
|
||||
"""Delete duplicate image files from disk (single flight with the scan)."""
|
||||
job = self._acquire_long_job(self.LONG_JOB_DUPLICATES)
|
||||
if job is None:
|
||||
"""Delete duplicate image files from disk.
|
||||
|
||||
Single flight with the duplicate scan, and a records hold on the
|
||||
registry, so no scan, restore or tagging job runs meanwhile. A body
|
||||
without paths is refused before the hold (it is no job).
|
||||
"""
|
||||
data, error_response = await _read_json(request)
|
||||
if error_response is not None:
|
||||
return error_response
|
||||
image_paths = data.get("image_paths", [])
|
||||
if not image_paths or not isinstance(image_paths, list):
|
||||
return web.json_response(
|
||||
{"success": False, "error": "No image paths provided"},
|
||||
status=400,
|
||||
)
|
||||
guard = self._acquire_long_job(self.LONG_JOB_DUPLICATES)
|
||||
if guard is None:
|
||||
return self._long_job_busy_response("Duplicate scan or delete")
|
||||
async with job:
|
||||
return await self._delete_duplicate_images(request)
|
||||
|
||||
async def _delete_duplicate_images(self, request):
|
||||
async def delete():
|
||||
# No await since the check above: the lock is still free.
|
||||
async with guard:
|
||||
return await self._delete_duplicate_images(image_paths)
|
||||
|
||||
return await self._run_held(
|
||||
DUPLICATE_CLEANUP_JOB_KIND,
|
||||
"Duplicate image delete (gallery page)",
|
||||
"records",
|
||||
delete,
|
||||
)
|
||||
|
||||
async def _delete_duplicate_images(self, image_paths):
|
||||
try:
|
||||
data, error_response = await _read_json(request)
|
||||
if error_response is not None:
|
||||
return error_response
|
||||
image_paths = data.get("image_paths", [])
|
||||
|
||||
if not image_paths or not isinstance(image_paths, list):
|
||||
return web.json_response(
|
||||
{"success": False, "error": "No image paths provided"},
|
||||
status=400,
|
||||
)
|
||||
|
||||
output_dir = await self._run_in_executor(self._find_comfyui_output_dir)
|
||||
result = await self._run_in_executor(
|
||||
self._delete_duplicate_images_sync, image_paths, output_dir
|
||||
@@ -882,208 +896,6 @@ class AdminRoutesMixin:
|
||||
status=500,
|
||||
)
|
||||
|
||||
async def get_settings(self, request):
|
||||
"""Get current settings."""
|
||||
try:
|
||||
from ..config import PromptManagerConfig, GalleryConfig
|
||||
|
||||
root_paths = [
|
||||
self._public_path(d) for d in GalleryConfig.MONITORING_DIRECTORIES
|
||||
]
|
||||
return web.json_response(
|
||||
{
|
||||
"success": True,
|
||||
"settings": {
|
||||
"result_timeout": PromptManagerConfig.RESULT_TIMEOUT,
|
||||
"webui_display_mode": PromptManagerConfig.WEBUI_DISPLAY_MODE,
|
||||
"infinite_scroll": PromptManagerConfig.INFINITE_SCROLL,
|
||||
"worker_threads": PromptManagerConfig.WORKER_THREADS,
|
||||
"cpu_count": PromptManagerConfig.max_worker_threads(),
|
||||
"gallery_root_paths": root_paths,
|
||||
"gallery_root_path": root_paths[0] if root_paths else "",
|
||||
"monitored_directories": [
|
||||
self._public_path(d) for d in self._monitored_directories()
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
except Exception as e:
|
||||
self.logger.error(f"Get settings error: {e}", exc_info=True)
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Failed to get settings"},
|
||||
status=500,
|
||||
)
|
||||
|
||||
def _monitored_directories(self):
|
||||
"""Directories the running image monitor watches, else the configured roots."""
|
||||
from ..config import GalleryConfig
|
||||
|
||||
monitor = _current_image_monitor()
|
||||
if monitor is not None:
|
||||
return list(getattr(monitor, "monitored_directories", []))
|
||||
return list(GalleryConfig.MONITORING_DIRECTORIES)
|
||||
|
||||
async def save_settings(self, request):
|
||||
"""Save settings: the whole payload is validated before anything changes."""
|
||||
try:
|
||||
from ..config import PromptManagerConfig
|
||||
|
||||
data, error_response = await _read_json(request)
|
||||
if error_response is not None:
|
||||
return error_response
|
||||
|
||||
try:
|
||||
updates = self._validate_settings(data)
|
||||
except ValueError as ve:
|
||||
return web.json_response(
|
||||
{"success": False, "error": str(ve)}, status=400
|
||||
)
|
||||
|
||||
new_roots, error = self._parse_gallery_roots(data)
|
||||
if error:
|
||||
return web.json_response({"success": False, "error": error}, status=400)
|
||||
|
||||
for name, value in updates.items():
|
||||
setattr(PromptManagerConfig, name, value)
|
||||
restart_required = False
|
||||
if new_roots is not None:
|
||||
restart_required = self._apply_gallery_roots(new_roots)
|
||||
|
||||
self._persist_settings()
|
||||
|
||||
return web.json_response(
|
||||
{
|
||||
"success": True,
|
||||
"message": "Settings saved successfully",
|
||||
"restart_required": restart_required,
|
||||
}
|
||||
)
|
||||
except Exception as e:
|
||||
self.logger.error(f"Save settings error: {e}", exc_info=True)
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Failed to save settings"},
|
||||
status=500,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _validate_settings(data):
|
||||
"""Scalar settings of a payload as ``{CONFIG_ATTR: value}``.
|
||||
|
||||
Raises ValueError with a user-facing message on the first bad value;
|
||||
nothing is applied until every value passed.
|
||||
"""
|
||||
from ..config import PromptManagerConfig
|
||||
|
||||
updates = {}
|
||||
if "result_timeout" in data:
|
||||
updates["RESULT_TIMEOUT"] = validate_result_timeout(data["result_timeout"])
|
||||
if "webui_display_mode" in data:
|
||||
mode = data["webui_display_mode"]
|
||||
if mode not in ("popup", "newtab"):
|
||||
raise ValueError("webui_display_mode must be 'popup' or 'newtab'")
|
||||
updates["WEBUI_DISPLAY_MODE"] = mode
|
||||
if "infinite_scroll" in data:
|
||||
if not isinstance(data["infinite_scroll"], bool):
|
||||
raise ValueError("infinite_scroll must be true or false")
|
||||
updates["INFINITE_SCROLL"] = data["infinite_scroll"]
|
||||
if "worker_threads" in data:
|
||||
updates["WORKER_THREADS"] = validate_worker_threads(
|
||||
data["worker_threads"], PromptManagerConfig.max_worker_threads()
|
||||
)
|
||||
return updates
|
||||
|
||||
def _parse_gallery_roots(self, data):
|
||||
"""Extract and validate gallery roots from a settings payload.
|
||||
|
||||
Accepts ``gallery_root_paths`` (list, preferred) or the legacy single
|
||||
``gallery_root_path`` string. Blank entries are dropped.
|
||||
|
||||
Returns:
|
||||
(roots, error): ``roots`` is the validated list, or ``None`` when
|
||||
the payload carries no gallery root key; ``error`` is a message
|
||||
when validation failed (and ``roots`` is then ``None``).
|
||||
"""
|
||||
if "gallery_root_paths" in data:
|
||||
raw = data["gallery_root_paths"]
|
||||
if not isinstance(raw, list):
|
||||
return None, "gallery_root_paths must be a list"
|
||||
elif "gallery_root_path" in data:
|
||||
raw = [data["gallery_root_path"]]
|
||||
else:
|
||||
return None, None
|
||||
|
||||
from ..config import GalleryConfig
|
||||
|
||||
roots = []
|
||||
for entry in raw:
|
||||
if not isinstance(entry, str):
|
||||
return None, "Gallery root paths must be strings"
|
||||
entry = entry.strip()
|
||||
if not entry:
|
||||
continue
|
||||
ok, reason = GalleryConfig.validate_gallery_root(entry)
|
||||
if not ok:
|
||||
return (
|
||||
None,
|
||||
f"Invalid gallery root '{os.path.basename(entry)}': {reason}",
|
||||
)
|
||||
roots.append(GalleryConfig.resolve_gallery_root(entry))
|
||||
return roots, None
|
||||
|
||||
def _apply_gallery_roots(self, roots):
|
||||
"""Install validated gallery roots; returns True when they changed."""
|
||||
from ..config import GalleryConfig
|
||||
|
||||
if roots == list(GalleryConfig.MONITORING_DIRECTORIES):
|
||||
return False
|
||||
GalleryConfig.MONITORING_DIRECTORIES = roots
|
||||
self._cached_output_dir = None
|
||||
self.reset_gallery_roots_cache()
|
||||
return True
|
||||
|
||||
def _persist_settings(self):
|
||||
"""Merge the user-editable settings into the configured config.json.
|
||||
|
||||
Only the keys this endpoint owns are rewritten; every other section
|
||||
(integrations, database, performance, ...) and every other key in
|
||||
``web_ui``/``gallery`` is kept as it was. A config file that cannot
|
||||
be parsed is replaced.
|
||||
"""
|
||||
from ..config import GalleryConfig, PromptManagerConfig, write_private_json
|
||||
|
||||
config_file = PromptManagerConfig.get_config_path()
|
||||
existing = self._read_config_file(config_file)
|
||||
web_ui = dict(existing.get("web_ui") or {})
|
||||
web_ui["result_timeout"] = PromptManagerConfig.RESULT_TIMEOUT
|
||||
web_ui["webui_display_mode"] = PromptManagerConfig.WEBUI_DISPLAY_MODE
|
||||
web_ui["infinite_scroll"] = PromptManagerConfig.INFINITE_SCROLL
|
||||
gallery = dict(existing.get("gallery") or {})
|
||||
monitoring = dict(gallery.get("monitoring") or {})
|
||||
monitoring["directories"] = list(GalleryConfig.MONITORING_DIRECTORIES)
|
||||
gallery["monitoring"] = monitoring
|
||||
performance = dict(existing.get("performance") or {})
|
||||
performance["worker_threads"] = PromptManagerConfig.WORKER_THREADS
|
||||
config_data = {
|
||||
**existing,
|
||||
"web_ui": web_ui,
|
||||
"gallery": gallery,
|
||||
"performance": performance,
|
||||
}
|
||||
try:
|
||||
write_private_json(config_file, config_data)
|
||||
self.logger.info(f"Settings saved to {config_file}")
|
||||
except OSError as save_err:
|
||||
self.logger.warning(f"Could not save config file: {save_err}")
|
||||
|
||||
def _read_config_file(self, config_file):
|
||||
"""Current contents of config.json as a dict, or {} when unusable."""
|
||||
try:
|
||||
with open(config_file, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
except (OSError, ValueError):
|
||||
return {}
|
||||
return data if isinstance(data, dict) else {}
|
||||
|
||||
async def run_diagnostics(self, request):
|
||||
"""Run comprehensive system diagnostics and health checks."""
|
||||
try:
|
||||
@@ -1176,6 +988,11 @@ class AdminRoutesMixin:
|
||||
"vacuum",
|
||||
"cleanup_orphaned_images",
|
||||
)
|
||||
# Operations that only read; a run of only these holds the registry as
|
||||
# "read", every other run as "records".
|
||||
READ_ONLY_MAINTENANCE_OPERATIONS = frozenset(
|
||||
{"check_hash_duplicates", "statistics", "check_consistency"}
|
||||
)
|
||||
|
||||
async def run_maintenance(self, request):
|
||||
"""Perform database maintenance operations and report each outcome."""
|
||||
@@ -1188,7 +1005,27 @@ class AdminRoutesMixin:
|
||||
operations = data.get(
|
||||
"operations", list(self.DEFAULT_MAINTENANCE_OPERATIONS)
|
||||
)
|
||||
except Exception as e:
|
||||
self.logger.error(f"Maintenance error: {e}", exc_info=True)
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Maintenance failed"}, status=500
|
||||
)
|
||||
writes = not (
|
||||
isinstance(operations, list)
|
||||
and all(
|
||||
isinstance(name, str) and name in self.READ_ONLY_MAINTENANCE_OPERATIONS
|
||||
for name in operations
|
||||
)
|
||||
)
|
||||
return await self._run_held(
|
||||
MAINTENANCE_JOB_KIND,
|
||||
"Maintenance (admin page)",
|
||||
"records" if writes else "read",
|
||||
lambda: self._run_maintenance(operations),
|
||||
)
|
||||
|
||||
async def _run_maintenance(self, operations):
|
||||
try:
|
||||
results = await self._run_in_executor(
|
||||
self._run_maintenance_operations, operations
|
||||
)
|
||||
@@ -1342,8 +1179,18 @@ class AdminRoutesMixin:
|
||||
|
||||
The upload is streamed to a temp file (capped at restore_max_bytes),
|
||||
verified with PromptModel.verify_database_file and only then copied
|
||||
over the live database, which is backed up first.
|
||||
over the live database, which is backed up first. The whole request
|
||||
holds the registry (records): it is refused while a scan or another
|
||||
record-changing job runs, and those are refused while it runs.
|
||||
"""
|
||||
return await self._run_held(
|
||||
RESTORE_JOB_KIND,
|
||||
"Database restore (admin page)",
|
||||
"records",
|
||||
lambda: self._restore_database(request),
|
||||
)
|
||||
|
||||
async def _restore_database(self, request):
|
||||
max_bytes = getattr(self, "restore_max_bytes", 100 * 1024 * 1024)
|
||||
temp_path = None
|
||||
try:
|
||||
|
||||
@@ -8,6 +8,7 @@ from pathlib import Path
|
||||
from typing import Optional
|
||||
from urllib.parse import quote
|
||||
|
||||
from .jobs import AUTOTAG_BATCH_JOB_KIND
|
||||
from .prompts import thumbnail_url_for
|
||||
from .. import model_gate
|
||||
|
||||
@@ -63,6 +64,12 @@ def path_is_within(candidate, root) -> bool:
|
||||
return child != parent and child.is_relative_to(parent)
|
||||
|
||||
|
||||
async def _refusal(title):
|
||||
"""The single error event of an auto-tag batch refused by another job."""
|
||||
message = f"{title} already running"
|
||||
yield {"type": "error", "status": "error", "message": message}
|
||||
|
||||
|
||||
class AutotagRoutesMixin:
|
||||
"""Mixin providing auto-tagging API endpoints."""
|
||||
|
||||
@@ -425,6 +432,8 @@ class AutotagRoutesMixin:
|
||||
"general_threshold and character_threshold must be numeric"
|
||||
)
|
||||
|
||||
errors = [] # an error event ends the run; Activity lists it failed
|
||||
|
||||
async def events():
|
||||
try:
|
||||
from ..autotag import get_autotag_service
|
||||
@@ -432,14 +441,40 @@ class AutotagRoutesMixin:
|
||||
async for event in self._autotag_batch_events(
|
||||
get_autotag_service(), opts
|
||||
):
|
||||
if event.get("type") == "error":
|
||||
errors.append(event)
|
||||
yield event
|
||||
except Exception:
|
||||
self.logger.exception("AutoTag error")
|
||||
errors.append(_INTERNAL_ERROR)
|
||||
yield {
|
||||
"type": "error",
|
||||
"message": _INTERNAL_ERROR,
|
||||
}
|
||||
|
||||
# The batch writes prompt tags: a records hold on the registry, so it
|
||||
# never runs beside a scan, restore or the workspace tagging job.
|
||||
busy = self._jobs.blocker(AUTOTAG_BATCH_JOB_KIND, admission="records")
|
||||
if busy is not None:
|
||||
if busy.kind == AUTOTAG_BATCH_JOB_KIND: # unchanged payload
|
||||
return web.json_response(
|
||||
{"success": False, "error": _BUSY_ERROR}, status=409
|
||||
)
|
||||
# Both legacy pages read this route as a stream, where a 409 reads
|
||||
# as a lost connection: refuse with one error frame instead (the
|
||||
# admin page keys on ``type``, the gallery on ``status``).
|
||||
return await self._stream_sse(request, _refusal(busy.title))
|
||||
async with self._jobs.hold(
|
||||
AUTOTAG_BATCH_JOB_KIND,
|
||||
"Auto-tag images (legacy page)",
|
||||
admission="records",
|
||||
) as job:
|
||||
response = await self._start_autotag_batch(request, opts, events)
|
||||
if response.status >= 400 or errors:
|
||||
job.finish("failed")
|
||||
return response
|
||||
|
||||
async def _start_autotag_batch(self, request, opts, events):
|
||||
if not _autotag_batch_lock.acquire(blocking=False):
|
||||
return web.json_response(
|
||||
{"success": False, "error": _BUSY_ERROR}, status=409
|
||||
|
||||
+54
-8
@@ -19,6 +19,7 @@ from typing import Callable, NamedTuple
|
||||
from aiohttp import web
|
||||
from PIL import Image, UnidentifiedImageError
|
||||
|
||||
from .jobs import THUMBNAILS_JOB_KIND
|
||||
from .prompts import bad_request, parse_page_params, publish_image_paths
|
||||
from .prompts import thumbnail_url_for
|
||||
from .prompts import safe_error_message as _safe_error
|
||||
@@ -104,6 +105,10 @@ def _card_header(path: Path) -> tuple:
|
||||
return inspection.width, inspection.height, records
|
||||
|
||||
|
||||
# The legacy gallery page's thumbnail generation, as Activity lists it (D80).
|
||||
THUMBNAIL_GENERATION_TITLE = "Thumbnail generation (gallery page)"
|
||||
|
||||
|
||||
class _CardFacts(NamedTuple):
|
||||
"""One image file's card facts, derived from its header alone.
|
||||
|
||||
@@ -1736,11 +1741,24 @@ class ImageRoutesMixin:
|
||||
return cls.THUMBNAIL_SIZES.get(quality, cls.THUMBNAIL_SIZES["medium"])
|
||||
|
||||
async def generate_thumbnails(self, request):
|
||||
"""Generate every thumbnail in one executor job (POST, blocking)."""
|
||||
"""Generate every thumbnail in one executor job (POST, blocking).
|
||||
|
||||
Holds the registry as guarded work: refused while a record-changing
|
||||
job runs, and such jobs are refused while it runs. A body that cannot
|
||||
be read is refused before the hold, so it is never listed as a job.
|
||||
"""
|
||||
data, error = await _json_body_or_error(request)
|
||||
if error is not None:
|
||||
return error
|
||||
return await self._run_held(
|
||||
THUMBNAILS_JOB_KIND,
|
||||
THUMBNAIL_GENERATION_TITLE,
|
||||
"guarded",
|
||||
lambda: self._generate_thumbnails(data),
|
||||
)
|
||||
|
||||
async def _generate_thumbnails(self, data):
|
||||
try:
|
||||
data, error = await _json_body_or_error(request)
|
||||
if error is not None:
|
||||
return error
|
||||
thumbnail_size = self._thumbnail_size(data.get("quality", "medium"))
|
||||
|
||||
output_dirs = self._get_all_output_dirs()
|
||||
@@ -1935,14 +1953,32 @@ class ImageRoutesMixin:
|
||||
"Connection": "keep-alive",
|
||||
},
|
||||
)
|
||||
await response.prepare(request)
|
||||
busy = self._jobs.blocker(THUMBNAILS_JOB_KIND, admission="guarded")
|
||||
if busy is not None:
|
||||
# The gallery reads this stream: refuse with its error event.
|
||||
await response.prepare(request)
|
||||
message = f"{busy.title} already running"
|
||||
await self._emit_sse(
|
||||
response, "error", {"error": message, "message": message}
|
||||
)
|
||||
return response
|
||||
async with self._jobs.hold(
|
||||
THUMBNAILS_JOB_KIND, THUMBNAIL_GENERATION_TITLE, admission="guarded"
|
||||
) as job:
|
||||
await response.prepare(request)
|
||||
if not await self._thumbnail_progress_stream(response, thumbnail_size):
|
||||
job.finish("failed") # an error event ended the stream
|
||||
return response
|
||||
|
||||
async def _thumbnail_progress_stream(self, response, thumbnail_size) -> bool:
|
||||
"""Stream one generation run; False when it ended with an error event."""
|
||||
try:
|
||||
output_dirs = self._get_all_output_dirs()
|
||||
if not output_dirs:
|
||||
await self._emit_sse(
|
||||
response, "error", {"error": "ComfyUI output directory not found"}
|
||||
)
|
||||
return response
|
||||
return False
|
||||
|
||||
await self._emit_sse(
|
||||
response,
|
||||
@@ -1960,6 +1996,7 @@ class ImageRoutesMixin:
|
||||
remaining=plan["remaining"],
|
||||
)
|
||||
self.invalidate_gallery_cache()
|
||||
return True
|
||||
except Exception:
|
||||
self.logger.exception("Thumbnail generation failed")
|
||||
await self._emit_sse(
|
||||
@@ -1970,7 +2007,7 @@ class ImageRoutesMixin:
|
||||
"message": "Thumbnail generation failed. Check server logs.",
|
||||
},
|
||||
)
|
||||
return response
|
||||
return False
|
||||
|
||||
async def _stream_thumbnails(
|
||||
self, response, targets, thumbnail_size, skipped=0, remaining=0
|
||||
@@ -2179,8 +2216,17 @@ class ImageRoutesMixin:
|
||||
"""Clear generated thumbnails across gallery roots, preserving originals.
|
||||
|
||||
Invalid thumbnail directories return 400; unavailable gallery roots
|
||||
return 404, and unexpected filesystem failures return 500.
|
||||
return 404, and unexpected filesystem failures return 500. Holds the
|
||||
registry as guarded work, like generation.
|
||||
"""
|
||||
return await self._run_held(
|
||||
THUMBNAILS_JOB_KIND,
|
||||
"Thumbnail clear (gallery page)",
|
||||
"guarded",
|
||||
self._clear_thumbnails,
|
||||
)
|
||||
|
||||
async def _clear_thumbnails(self) -> web.Response:
|
||||
try:
|
||||
output_dirs = self._get_all_output_dirs()
|
||||
if not output_dirs:
|
||||
|
||||
+161
-25
@@ -9,9 +9,21 @@ as a snapshot and then every event published after it, until the job ends.
|
||||
|
||||
Jobs live only in server memory and are gone after a restart. The registry
|
||||
keeps the running jobs and this session's last ``MAX_FINISHED_JOBS`` finished
|
||||
ones. Record-changing jobs (the metadata scan, bulk image tagging, and later
|
||||
maintenance, cleanup and restore) never run together, and one job of a kind
|
||||
runs at a time.
|
||||
ones. One job of a kind runs at a time, and each job has an admission class
|
||||
(``ADMISSIONS``) that decides what may run beside it:
|
||||
|
||||
- ``records`` changes library records or files (the metadata scan, bulk image
|
||||
tagging, and the legacy restore, maintenance, cleanup, duplicate delete,
|
||||
LoRA import and batch auto-tag routes). It is refused while any
|
||||
``records`` or ``guarded`` job runs.
|
||||
- ``guarded`` must not overlap record changes but may overlap other guarded
|
||||
work (thumbnail generation and clearing, later backup). It is refused while
|
||||
a ``records`` job runs.
|
||||
- ``read`` only reads; it is refused only by a running job of its own kind.
|
||||
|
||||
A short legacy-page operation takes a ``hold``: a running, non-cancellable
|
||||
job that lasts as long as the request, so the Activity page lists it and a
|
||||
workspace job is refused while it runs (and the other way round).
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
@@ -60,6 +72,41 @@ SCAN_GROUPS = {
|
||||
IMAGE_TAG_JOB_KIND = "image-tags"
|
||||
IMAGE_TAG_GROUP_NAMES = ("tagged", "unchanged", "skipped", "errors")
|
||||
|
||||
ADMISSIONS = ("read", "guarded", "records")
|
||||
# What each admission class is refused beside (besides its own kind).
|
||||
_REFUSED_BESIDE = {
|
||||
"read": frozenset(),
|
||||
"guarded": frozenset({"records"}),
|
||||
"records": frozenset({"records", "guarded"}),
|
||||
}
|
||||
|
||||
# Kinds of the legacy-page operations that hold the registry (D80; titles in
|
||||
# their handlers end in "(admin page)", "(gallery page)" or "(legacy page)").
|
||||
RESTORE_JOB_KIND = "restore"
|
||||
MAINTENANCE_JOB_KIND = "maintenance"
|
||||
DUPLICATE_CLEANUP_JOB_KIND = "duplicate_cleanup"
|
||||
THUMBNAILS_JOB_KIND = "thumbnails"
|
||||
LORA_IMPORT_JOB_KIND = "lora_import"
|
||||
AUTOTAG_BATCH_JOB_KIND = "autotag"
|
||||
|
||||
|
||||
def resolve_admission(admission=None, record_changing=None) -> str:
|
||||
"""The admission class for a start call; ``record_changing`` is an alias.
|
||||
|
||||
``record_changing=True`` means ``records`` and ``False`` means ``read``;
|
||||
with neither given the class is ``read``. Raises ValueError for an
|
||||
unknown class or one that contradicts ``record_changing``.
|
||||
"""
|
||||
if admission is None:
|
||||
return "records" if record_changing else "read"
|
||||
if admission not in ADMISSIONS:
|
||||
raise ValueError(f"Unknown job admission {admission!r}")
|
||||
if record_changing is not None and record_changing != (admission == "records"):
|
||||
raise ValueError(
|
||||
f"record_changing={record_changing!r} contradicts admission {admission!r}"
|
||||
)
|
||||
return admission
|
||||
|
||||
|
||||
def _relative_path(path) -> str:
|
||||
"""``path`` as a root-relative POSIX path; an absolute one keeps its name."""
|
||||
@@ -85,13 +132,23 @@ class Job:
|
||||
"""One tracked operation: state, counters, outcomes, log and subscribers."""
|
||||
|
||||
def __init__(
|
||||
self, kind, title, *, record_changing=False, scope=None, groups=(), clock=None
|
||||
self,
|
||||
kind,
|
||||
title,
|
||||
*,
|
||||
admission=None,
|
||||
record_changing=None,
|
||||
cancellable=True,
|
||||
scope=None,
|
||||
groups=(),
|
||||
clock=None,
|
||||
):
|
||||
self._clock = clock or time.time
|
||||
self.id = uuid.uuid4().hex
|
||||
self.kind = kind
|
||||
self.title = title
|
||||
self.record_changing = record_changing
|
||||
self.admission = resolve_admission(admission, record_changing)
|
||||
self.cancellable = bool(cancellable)
|
||||
self.scope = dict(scope or {})
|
||||
self.state = "running"
|
||||
self.started_at = self._clock()
|
||||
@@ -110,6 +167,10 @@ class Job:
|
||||
self.task = None
|
||||
self._subscribers = set()
|
||||
|
||||
@property
|
||||
def record_changing(self):
|
||||
return self.admission == "records"
|
||||
|
||||
@property
|
||||
def running(self):
|
||||
return self.state == "running"
|
||||
@@ -240,6 +301,9 @@ class Job:
|
||||
# What detail() lists, so a view re-reads it only when it changed.
|
||||
"listed": {group: len(rows) for group, rows in self.outcomes.items()},
|
||||
"log_length": len(self.log),
|
||||
# A view offers Cancel only when the server can stop the job.
|
||||
"cancellable": self.cancellable,
|
||||
"admission": self.admission,
|
||||
}
|
||||
|
||||
def detail(self):
|
||||
@@ -291,33 +355,88 @@ class JobRegistry:
|
||||
self._logger = logger_ or logger
|
||||
self._jobs = {}
|
||||
|
||||
def start(self, kind, title, *, record_changing, scope, runner, groups=()) -> Job:
|
||||
def start(
|
||||
self,
|
||||
kind,
|
||||
title,
|
||||
*,
|
||||
admission=None,
|
||||
record_changing=None,
|
||||
cancellable=True,
|
||||
scope=None,
|
||||
runner,
|
||||
groups=(),
|
||||
) -> Job:
|
||||
"""Create a job and run ``runner(job)`` as a task.
|
||||
|
||||
Raises JobBusy, naming the job in the way, when a job of ``kind`` is
|
||||
running or when ``record_changing`` and another record-changing job
|
||||
runs. Must be called from the event loop.
|
||||
``admission`` is one of ``ADMISSIONS``; ``record_changing=True`` is
|
||||
the older spelling of ``records`` and ``False`` of ``read``. One of
|
||||
the two is required (TypeError otherwise), so no new job runs as
|
||||
``read`` beside a scan by omission. Raises JobBusy, naming the job in
|
||||
the way, when the admission class refuses it (see the module
|
||||
docstring). Must be called from the event loop.
|
||||
"""
|
||||
blocker = self.blocker(kind, record_changing=record_changing)
|
||||
if blocker is not None:
|
||||
raise JobBusy(blocker)
|
||||
self._prune()
|
||||
job = Job(
|
||||
kind, title, record_changing=record_changing, scope=scope, groups=groups
|
||||
if admission is None and record_changing is None:
|
||||
raise TypeError(
|
||||
f"JobRegistry.start({kind!r}) needs an admission class "
|
||||
"(admission= or record_changing=)"
|
||||
)
|
||||
job = self._admit(
|
||||
kind,
|
||||
title,
|
||||
admission=admission,
|
||||
record_changing=record_changing,
|
||||
cancellable=cancellable,
|
||||
scope=scope,
|
||||
groups=groups,
|
||||
)
|
||||
self._jobs[job.id] = job
|
||||
job.task = asyncio.ensure_future(self._run(job, runner))
|
||||
return job
|
||||
|
||||
def blocker(self, kind, *, record_changing):
|
||||
"""The running job a new ``kind`` job would be refused for, or None."""
|
||||
def hold(self, kind, title, *, admission, scope=None):
|
||||
"""A running, non-cancellable job for a legacy-page operation.
|
||||
|
||||
Returns an async context manager. Entering it registers the job and
|
||||
yields it, or raises JobBusy at once, without waiting, when the
|
||||
admission class refuses it; nothing is registered until then, so a
|
||||
hold that is never entered leaves no job behind. The job ends when
|
||||
the block exits: ``finished``, or ``failed`` when it raised (unless
|
||||
the body already ended it). A caller that answers a refusal in its
|
||||
own way checks ``blocker(kind, admission=...)`` first, with no await
|
||||
before the ``async with``.
|
||||
"""
|
||||
resolve_admission(admission) # an unknown class fails at the call
|
||||
return _Hold(
|
||||
lambda: self._admit(
|
||||
kind, title, admission=admission, cancellable=False, scope=scope
|
||||
)
|
||||
)
|
||||
|
||||
def _admit(self, kind, title, *, admission, record_changing=None, **job_args):
|
||||
resolved = resolve_admission(admission, record_changing)
|
||||
blocker = self.blocker(kind, admission=resolved)
|
||||
if blocker is not None:
|
||||
raise JobBusy(blocker)
|
||||
self._prune()
|
||||
job = Job(kind, title, admission=resolved, **job_args)
|
||||
self._jobs[job.id] = job
|
||||
return job
|
||||
|
||||
def blocker(self, kind, *, admission=None, record_changing=None):
|
||||
"""The running job a new ``kind`` job would be refused for, or None.
|
||||
|
||||
The same kind refuses first; then the oldest running job whose class
|
||||
the new job's admission class may not run beside.
|
||||
"""
|
||||
same_kind = self.active(kind)
|
||||
if same_kind is not None:
|
||||
return same_kind
|
||||
if record_changing:
|
||||
for job in self._jobs.values():
|
||||
if job.running and job.record_changing:
|
||||
return job
|
||||
refused = _REFUSED_BESIDE[resolve_admission(admission, record_changing)]
|
||||
running = [job for job in self._jobs.values() if job.running]
|
||||
running.sort(key=lambda job: job._started_seq)
|
||||
for job in running:
|
||||
if job.admission in refused:
|
||||
return job
|
||||
return None
|
||||
|
||||
async def _run(self, job, runner):
|
||||
@@ -362,11 +481,28 @@ class JobRegistry:
|
||||
def cancel(self, job_id) -> bool:
|
||||
"""Ask a running job to stop at its next safe boundary.
|
||||
|
||||
Returns whether a cancellation was requested: False for a finished or
|
||||
unknown job, so a cancel after the end is a no-op.
|
||||
Returns whether a cancellation was requested: False for a finished,
|
||||
unknown or non-cancellable job, so a cancel after the end is a no-op.
|
||||
"""
|
||||
job = self._jobs.get(job_id)
|
||||
if job is None or job.finished:
|
||||
if job is None or job.finished or not job.cancellable:
|
||||
return False
|
||||
job.cancelled = True
|
||||
return True
|
||||
|
||||
|
||||
class _Hold:
|
||||
"""The async context manager ``JobRegistry.hold`` returns."""
|
||||
|
||||
def __init__(self, admit):
|
||||
self._admit = admit
|
||||
self._job = None
|
||||
|
||||
async def __aenter__(self):
|
||||
self._job = self._admit()
|
||||
return self._job
|
||||
|
||||
async def __aexit__(self, exc_type, exc, tb):
|
||||
if self._job.running:
|
||||
self._job.finish("finished" if exc_type is None else "failed")
|
||||
return False
|
||||
|
||||
+104
-20
@@ -5,12 +5,47 @@ from pathlib import Path
|
||||
|
||||
from aiohttp import web
|
||||
|
||||
from ..config_store import ConfigUnreadable, RevisionConflict, get_config_store
|
||||
from .jobs import LORA_IMPORT_JOB_KIND
|
||||
|
||||
# Flags of POST /lora/enable; each must be a JSON boolean when present.
|
||||
_LORA_FLAGS = ("enabled", "trigger_words_enabled", "clear_civitai_api_key")
|
||||
|
||||
_LORA_PATH_ERROR = (
|
||||
"LoraManager path must be a directory under ComfyUI's custom_nodes "
|
||||
"whose name contains 'lora'"
|
||||
)
|
||||
|
||||
|
||||
def _lora_settings_mutation(values):
|
||||
"""The config commit for a LoRA save: only ``integrations.lora_manager``.
|
||||
|
||||
The submitted keys replace their values; other LoRA keys, other
|
||||
integrations and every other section stay as they are on disk.
|
||||
"""
|
||||
|
||||
def mutate(config):
|
||||
integrations = config.get("integrations")
|
||||
integrations = dict(integrations) if isinstance(integrations, dict) else {}
|
||||
lora = integrations.get("lora_manager")
|
||||
lora = dict(lora) if isinstance(lora, dict) else {}
|
||||
lora.update(values)
|
||||
integrations["lora_manager"] = lora
|
||||
config["integrations"] = integrations
|
||||
return config
|
||||
|
||||
return mutate
|
||||
|
||||
|
||||
def _lora_error_frame(message):
|
||||
"""An import stream's error event, as the admin page's modal reads it.
|
||||
|
||||
The modal shows ``status`` and ``progress`` from every frame and closes
|
||||
on ``type: error``, naming ``message``.
|
||||
"""
|
||||
return {"type": "error", "status": message, "progress": 0, "message": message}
|
||||
|
||||
|
||||
class LoraIntegrationMixin:
|
||||
"""Mixin providing LoraManager detection, scanning, and trigger word endpoints."""
|
||||
|
||||
@@ -131,14 +166,25 @@ class LoraIntegrationMixin:
|
||||
return allowed, None
|
||||
|
||||
async def lora_enable(self, request):
|
||||
"""Enable or disable LoraManager integration and save to config.json."""
|
||||
"""Enable or disable LoraManager integration and save it to config.json.
|
||||
|
||||
The values are committed through the config store first (#220) and
|
||||
published at runtime only once config.json holds them; a save that
|
||||
cannot be written reports ``persisted: false`` and changes nothing.
|
||||
The CivitAI key is never echoed back.
|
||||
"""
|
||||
try:
|
||||
data = await request.json()
|
||||
data, error = await self._read_json_body(request)
|
||||
if error is not None:
|
||||
return error
|
||||
for flag in _LORA_FLAGS:
|
||||
if flag in data and not isinstance(data[flag], bool):
|
||||
return self._bad_lora_request(f"{flag} must be true or false")
|
||||
enabled = data.get("enabled", False)
|
||||
path = data.get("path", "")
|
||||
trigger_words = data.get("trigger_words_enabled", False)
|
||||
|
||||
from ..config import IntegrationConfig, PromptManagerConfig
|
||||
from ..config import IntegrationConfig
|
||||
from ..lora_utils import get_trigger_cache
|
||||
|
||||
# The key is never echoed back to the UI, so a blank or absent
|
||||
@@ -161,20 +207,40 @@ class LoraIntegrationMixin:
|
||||
if error is not None:
|
||||
return error
|
||||
|
||||
# Update in-memory config
|
||||
IntegrationConfig.LORA_MANAGER_ENABLED = enabled
|
||||
IntegrationConfig.LORA_MANAGER_PATH = path
|
||||
IntegrationConfig.LORA_TRIGGER_WORDS_ENABLED = trigger_words
|
||||
IntegrationConfig.CIVITAI_API_KEY = civitai_key
|
||||
values = {
|
||||
"enabled": enabled,
|
||||
"path": path,
|
||||
"trigger_words_enabled": trigger_words,
|
||||
"civitai_api_key": civitai_key,
|
||||
}
|
||||
|
||||
# Persist to the configured config.json
|
||||
PromptManagerConfig.save_to_file()
|
||||
def publish(_config):
|
||||
# Under the commit lock, so the last save written is the one
|
||||
# the runtime keeps.
|
||||
IntegrationConfig.LORA_MANAGER_ENABLED = enabled
|
||||
IntegrationConfig.LORA_MANAGER_PATH = path
|
||||
IntegrationConfig.LORA_TRIGGER_WORDS_ENABLED = trigger_words
|
||||
IntegrationConfig.CIVITAI_API_KEY = civitai_key
|
||||
|
||||
try:
|
||||
_saved, revision = await self._run_in_executor(
|
||||
get_config_store().commit,
|
||||
_lora_settings_mutation(values),
|
||||
expected_revision=None,
|
||||
publish=publish,
|
||||
)
|
||||
except (ConfigUnreadable, RevisionConflict, OSError) as exc:
|
||||
return self._config_commit_failure(exc)
|
||||
|
||||
# Load trigger word cache if enabling
|
||||
cache = get_trigger_cache()
|
||||
if enabled and trigger_words and path:
|
||||
count = await self._run_in_executor(cache.load, path)
|
||||
self.logger.info(f"Trigger word cache loaded: {count} LoRAs")
|
||||
try:
|
||||
count = await self._run_in_executor(cache.load, path)
|
||||
self.logger.info(f"Trigger word cache loaded: {count} LoRAs")
|
||||
except Exception as e:
|
||||
# Saved and enabled; only the trigger words are missing.
|
||||
self.logger.warning(f"Trigger word cache could not be loaded: {e}")
|
||||
cache.clear()
|
||||
elif not enabled:
|
||||
cache.clear()
|
||||
|
||||
@@ -184,6 +250,10 @@ class LoraIntegrationMixin:
|
||||
"enabled": enabled,
|
||||
"path": self._public_path(path),
|
||||
"trigger_words_enabled": trigger_words,
|
||||
"persisted": True,
|
||||
"revision": revision,
|
||||
"runtime_applied": True,
|
||||
"cache_loaded": cache.is_loaded,
|
||||
}
|
||||
)
|
||||
except Exception as e:
|
||||
@@ -218,19 +288,33 @@ class LoraIntegrationMixin:
|
||||
reason="OK",
|
||||
headers={"Content-Type": "text/event-stream"},
|
||||
)
|
||||
await response.prepare(request)
|
||||
|
||||
async def send_progress(data):
|
||||
line = f"data: {json.dumps(data)}\n\n"
|
||||
await response.write(line.encode("utf-8"))
|
||||
|
||||
try:
|
||||
await self._lora_scan_stream(allowed, send_progress)
|
||||
except Exception as e:
|
||||
self.logger.exception("LoRA scan failed")
|
||||
await send_progress({"type": "error", "message": self._public_error(e)})
|
||||
# The import replaces prompts and images: a records hold, so it
|
||||
# never runs beside a scan, restore or tagging job.
|
||||
busy = self._jobs.blocker(LORA_IMPORT_JOB_KIND, admission="records")
|
||||
if busy is not None:
|
||||
await response.prepare(request)
|
||||
message = f"{busy.title} already running"
|
||||
await send_progress(_lora_error_frame(message))
|
||||
await response.write_eof()
|
||||
return response
|
||||
|
||||
await response.write_eof()
|
||||
async with self._jobs.hold(
|
||||
LORA_IMPORT_JOB_KIND, "LoRA import (admin page)", admission="records"
|
||||
) as job:
|
||||
await response.prepare(request)
|
||||
try:
|
||||
await self._lora_scan_stream(allowed, send_progress)
|
||||
except Exception as e:
|
||||
self.logger.exception("LoRA scan failed")
|
||||
job.finish("failed")
|
||||
await send_progress(_lora_error_frame(self._public_error(e)))
|
||||
|
||||
await response.write_eof()
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
|
||||
@@ -0,0 +1,339 @@
|
||||
"""Settings API routes: GET and POST /prompt_manager/settings (#220).
|
||||
|
||||
A save validates the payload, commits it to config.json through the config
|
||||
store and publishes it at runtime inside that commit; a save that cannot be
|
||||
written changes nothing at runtime and says so. ``_config_commit_failure``
|
||||
is shared with the LoRA integration's save.
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
from aiohttp import web
|
||||
|
||||
from ..config_store import (
|
||||
ConfigUnreadable,
|
||||
RevisionConflict,
|
||||
get_config_store,
|
||||
)
|
||||
from .admin import _current_image_monitor
|
||||
|
||||
try:
|
||||
from ...utils.validators import validate_result_timeout, validate_worker_threads
|
||||
except ImportError:
|
||||
from utils.validators import validate_result_timeout, validate_worker_threads
|
||||
|
||||
|
||||
# Settings owned by POST /settings: config attribute -> (section, key).
|
||||
_SETTINGS_FILE_KEYS = {
|
||||
"RESULT_TIMEOUT": ("web_ui", "result_timeout"),
|
||||
"WEBUI_DISPLAY_MODE": ("web_ui", "webui_display_mode"),
|
||||
"INFINITE_SCROLL": ("web_ui", "infinite_scroll"),
|
||||
"WORKER_THREADS": ("performance", "worker_threads"),
|
||||
}
|
||||
|
||||
|
||||
def _section(config, name):
|
||||
"""``config[name]`` as a fresh dict (a non-object section starts empty)."""
|
||||
value = config.get(name)
|
||||
return dict(value) if isinstance(value, dict) else {}
|
||||
|
||||
|
||||
def _settings_mutation(updates, new_roots):
|
||||
"""The config commit for a settings save: only the submitted keys change.
|
||||
|
||||
Every other section and key (integrations and the CivitAI key, database,
|
||||
unknown sections) is kept as it is on disk.
|
||||
"""
|
||||
|
||||
def mutate(config):
|
||||
for name, value in updates.items():
|
||||
section, key = _SETTINGS_FILE_KEYS[name]
|
||||
merged = _section(config, section)
|
||||
merged[key] = value
|
||||
config[section] = merged
|
||||
if new_roots is not None:
|
||||
gallery = _section(config, "gallery")
|
||||
monitoring = _section(gallery, "monitoring")
|
||||
monitoring["directories"] = list(new_roots)
|
||||
gallery["monitoring"] = monitoring
|
||||
config["gallery"] = gallery
|
||||
return config
|
||||
|
||||
return mutate
|
||||
|
||||
|
||||
async def _read_json(request):
|
||||
"""Parse a JSON object body through the package's capped reader."""
|
||||
from . import _read_json_body
|
||||
|
||||
return await _read_json_body(request)
|
||||
|
||||
|
||||
class SettingsRoutesMixin:
|
||||
"""Mixin providing the settings read and save endpoints."""
|
||||
|
||||
def _register_settings_routes(self, routes):
|
||||
@routes.get("/prompt_manager/settings")
|
||||
async def get_settings_route(request):
|
||||
return await self.get_settings(request)
|
||||
|
||||
@routes.post("/prompt_manager/settings")
|
||||
async def save_settings_route(request):
|
||||
return await self.save_settings(request)
|
||||
|
||||
async def get_settings(self, request):
|
||||
"""Get current settings."""
|
||||
try:
|
||||
from ..config import PromptManagerConfig, GalleryConfig
|
||||
|
||||
root_paths = [
|
||||
self._public_path(d) for d in GalleryConfig.MONITORING_DIRECTORIES
|
||||
]
|
||||
revision = await self._run_in_executor(self._config_revision)
|
||||
return web.json_response(
|
||||
{
|
||||
"success": True,
|
||||
**revision,
|
||||
"settings": {
|
||||
"result_timeout": PromptManagerConfig.RESULT_TIMEOUT,
|
||||
"webui_display_mode": PromptManagerConfig.WEBUI_DISPLAY_MODE,
|
||||
"infinite_scroll": PromptManagerConfig.INFINITE_SCROLL,
|
||||
"worker_threads": PromptManagerConfig.WORKER_THREADS,
|
||||
"cpu_count": PromptManagerConfig.max_worker_threads(),
|
||||
"gallery_root_paths": root_paths,
|
||||
"gallery_root_path": root_paths[0] if root_paths else "",
|
||||
"monitored_directories": [
|
||||
self._public_path(d) for d in self._monitored_directories()
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
except Exception as e:
|
||||
self.logger.error(f"Get settings error: {e}", exc_info=True)
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Failed to get settings"},
|
||||
status=500,
|
||||
)
|
||||
|
||||
def _monitored_directories(self):
|
||||
"""Directories the running image monitor watches, else the configured roots."""
|
||||
from ..config import GalleryConfig
|
||||
|
||||
monitor = _current_image_monitor()
|
||||
if monitor is not None:
|
||||
return list(getattr(monitor, "monitored_directories", []))
|
||||
return list(GalleryConfig.MONITORING_DIRECTORIES)
|
||||
|
||||
async def save_settings(self, request):
|
||||
"""Save settings: validate, commit to config.json, then publish (#220).
|
||||
|
||||
Nothing changes at runtime until config.json holds the new values, so
|
||||
a save that cannot be written reports ``persisted: false`` and leaves
|
||||
the running settings as they were. ``expected_revision`` (the
|
||||
``revision`` from GET) refuses to overwrite a file changed meanwhile;
|
||||
without it the payload merges into what is on disk (legacy page).
|
||||
"""
|
||||
try:
|
||||
data, error_response = await _read_json(request)
|
||||
if error_response is not None:
|
||||
return error_response
|
||||
|
||||
expected_revision = data.get("expected_revision")
|
||||
if expected_revision is not None and not isinstance(expected_revision, str):
|
||||
return web.json_response(
|
||||
{"success": False, "error": "expected_revision must be a string"},
|
||||
status=400,
|
||||
)
|
||||
try:
|
||||
updates = self._validate_settings(data)
|
||||
except ValueError as ve:
|
||||
return web.json_response(
|
||||
{"success": False, "error": str(ve)}, status=400
|
||||
)
|
||||
|
||||
new_roots, error = self._parse_gallery_roots(data)
|
||||
if error:
|
||||
return web.json_response({"success": False, "error": error}, status=400)
|
||||
|
||||
published = {}
|
||||
|
||||
def publish(_config):
|
||||
# Under the commit lock: saves reach the runtime in the order
|
||||
# they were written, so the runtime ends on the last one.
|
||||
published.update(self._publish_settings(updates, new_roots))
|
||||
|
||||
try:
|
||||
_saved, revision = await self._run_in_executor(
|
||||
get_config_store().commit,
|
||||
_settings_mutation(updates, new_roots),
|
||||
expected_revision=expected_revision,
|
||||
publish=publish,
|
||||
)
|
||||
except (ConfigUnreadable, RevisionConflict, OSError) as exc:
|
||||
return self._config_commit_failure(exc)
|
||||
|
||||
return web.json_response(
|
||||
{
|
||||
"success": True,
|
||||
"message": "Settings saved successfully",
|
||||
"restart_required": bool(published["restart_required_fields"]),
|
||||
"persisted": True,
|
||||
"revision": revision,
|
||||
**published,
|
||||
}
|
||||
)
|
||||
except Exception as e:
|
||||
self.logger.error(f"Save settings error: {e}", exc_info=True)
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Failed to save settings"},
|
||||
status=500,
|
||||
)
|
||||
|
||||
def _publish_settings(self, updates, new_roots):
|
||||
"""Install committed settings at runtime (inside the config commit).
|
||||
|
||||
The values are already durable. A publication that fails is reported
|
||||
as ``runtime_applied: false`` with the fields that need a restart;
|
||||
nothing is retried or rolled back. Never raises.
|
||||
"""
|
||||
from ..config import PromptManagerConfig
|
||||
|
||||
for name, value in updates.items():
|
||||
setattr(PromptManagerConfig, name, value)
|
||||
restart_fields = []
|
||||
runtime_applied = True
|
||||
if new_roots is not None:
|
||||
try:
|
||||
if self._apply_gallery_roots(new_roots):
|
||||
restart_fields.append("gallery_root_paths")
|
||||
except Exception as e:
|
||||
self.logger.error(
|
||||
f"Saved gallery roots could not be applied: {e}", exc_info=True
|
||||
)
|
||||
runtime_applied = False
|
||||
restart_fields.append("gallery_root_paths")
|
||||
return {
|
||||
"runtime_applied": runtime_applied,
|
||||
"restart_required_fields": restart_fields,
|
||||
}
|
||||
|
||||
def _config_commit_failure(self, exc):
|
||||
"""The response for a config commit that wrote nothing."""
|
||||
not_saved = {"success": False, "persisted": False, "runtime_applied": False}
|
||||
if isinstance(exc, ConfigUnreadable):
|
||||
self.logger.warning(f"Settings not saved: {exc}")
|
||||
return web.json_response(
|
||||
{
|
||||
**not_saved,
|
||||
"code": "config-unreadable",
|
||||
"error": (
|
||||
"config.json cannot be read, so nothing was saved. "
|
||||
"Fix or remove the file, then save again."
|
||||
),
|
||||
},
|
||||
status=409,
|
||||
)
|
||||
if isinstance(exc, RevisionConflict):
|
||||
return web.json_response(
|
||||
{
|
||||
**not_saved,
|
||||
"code": "settings-changed",
|
||||
"error": "Settings changed since they were loaded. Reload them.",
|
||||
"revision": exc.current,
|
||||
},
|
||||
status=409,
|
||||
)
|
||||
self.logger.error(f"Could not write config.json: {exc}")
|
||||
return web.json_response(
|
||||
{
|
||||
**not_saved,
|
||||
"code": "config-write-failed",
|
||||
"error": "Settings could not be written to config.json.",
|
||||
},
|
||||
status=500,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _config_revision():
|
||||
"""``{"revision": ...}``, or a null revision with ``config_error``."""
|
||||
try:
|
||||
_data, revision = get_config_store().read()
|
||||
except ConfigUnreadable:
|
||||
return {"revision": None, "config_error": "config-unreadable"}
|
||||
return {"revision": revision}
|
||||
|
||||
@staticmethod
|
||||
def _validate_settings(data):
|
||||
"""Scalar settings of a payload as ``{CONFIG_ATTR: value}``.
|
||||
|
||||
Raises ValueError with a user-facing message on the first bad value;
|
||||
nothing is applied until every value passed.
|
||||
"""
|
||||
from ..config import PromptManagerConfig
|
||||
|
||||
updates = {}
|
||||
if "result_timeout" in data:
|
||||
updates["RESULT_TIMEOUT"] = validate_result_timeout(data["result_timeout"])
|
||||
if "webui_display_mode" in data:
|
||||
mode = data["webui_display_mode"]
|
||||
if mode not in ("popup", "newtab"):
|
||||
raise ValueError("webui_display_mode must be 'popup' or 'newtab'")
|
||||
updates["WEBUI_DISPLAY_MODE"] = mode
|
||||
if "infinite_scroll" in data:
|
||||
if not isinstance(data["infinite_scroll"], bool):
|
||||
raise ValueError("infinite_scroll must be true or false")
|
||||
updates["INFINITE_SCROLL"] = data["infinite_scroll"]
|
||||
if "worker_threads" in data:
|
||||
updates["WORKER_THREADS"] = validate_worker_threads(
|
||||
data["worker_threads"], PromptManagerConfig.max_worker_threads()
|
||||
)
|
||||
return updates
|
||||
|
||||
def _parse_gallery_roots(self, data):
|
||||
"""Extract and validate gallery roots from a settings payload.
|
||||
|
||||
Accepts ``gallery_root_paths`` (list, preferred) or the legacy single
|
||||
``gallery_root_path`` string. Blank entries are dropped.
|
||||
|
||||
Returns:
|
||||
(roots, error): ``roots`` is the validated list, or ``None`` when
|
||||
the payload carries no gallery root key; ``error`` is a message
|
||||
when validation failed (and ``roots`` is then ``None``).
|
||||
"""
|
||||
if "gallery_root_paths" in data:
|
||||
raw = data["gallery_root_paths"]
|
||||
if not isinstance(raw, list):
|
||||
return None, "gallery_root_paths must be a list"
|
||||
elif "gallery_root_path" in data:
|
||||
raw = [data["gallery_root_path"]]
|
||||
else:
|
||||
return None, None
|
||||
|
||||
from ..config import GalleryConfig
|
||||
|
||||
roots = []
|
||||
for entry in raw:
|
||||
if not isinstance(entry, str):
|
||||
return None, "Gallery root paths must be strings"
|
||||
entry = entry.strip()
|
||||
if not entry:
|
||||
continue
|
||||
ok, reason = GalleryConfig.validate_gallery_root(entry)
|
||||
if not ok:
|
||||
return (
|
||||
None,
|
||||
f"Invalid gallery root '{os.path.basename(entry)}': {reason}",
|
||||
)
|
||||
roots.append(GalleryConfig.resolve_gallery_root(entry))
|
||||
return roots, None
|
||||
|
||||
def _apply_gallery_roots(self, roots):
|
||||
"""Install validated gallery roots; returns True when they changed."""
|
||||
from ..config import GalleryConfig
|
||||
|
||||
if roots == list(GalleryConfig.MONITORING_DIRECTORIES):
|
||||
return False
|
||||
GalleryConfig.MONITORING_DIRECTORIES = roots
|
||||
self._cached_output_dir = None
|
||||
self.reset_gallery_roots_cache()
|
||||
return True
|
||||
@@ -0,0 +1,197 @@
|
||||
"""The one serialized writer of config.json (#220).
|
||||
|
||||
Every save that changes config.json goes through ``ConfigStore.commit``: the
|
||||
settings page, the LoRA integration and, later, every System section. A
|
||||
commit runs under one process-wide lock, re-reads the file inside it, applies
|
||||
the caller's change to a copy and replaces the file atomically. So two saves
|
||||
never overwrite each other's keys, sections this process does not know are
|
||||
kept byte-for-byte in meaning, and a write that fails leaves the old file in
|
||||
place and raises instead of being logged and reported as saved.
|
||||
|
||||
A revision names the file content a client last saw: ``"sha256:<hex>"`` of
|
||||
the file bytes, or ``"absent"`` when there is no file. A commit that names a
|
||||
revision refuses to write over different content (``RevisionConflict``).
|
||||
|
||||
Reads take the same lock, so a read handle is never open while the file is
|
||||
replaced (Windows refuses that), and a caller's ``publish`` runs inside the
|
||||
lock right after its write, so runtimes publish saves in the order they were
|
||||
written. A symlinked config.json is written beside its real target.
|
||||
|
||||
config.json can hold the CivitAI key, so the file is written owner-only.
|
||||
"""
|
||||
|
||||
import copy
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
from typing import Callable, Optional, Tuple
|
||||
|
||||
ABSENT_REVISION = "absent"
|
||||
|
||||
# One lock for the process: every ConfigStore writes the same file. It is
|
||||
# re-entrant so a commit's ``publish`` callback may read the config again.
|
||||
_COMMIT_LOCK = threading.RLock()
|
||||
|
||||
# Windows: an antivirus, indexer or backup handle on config.json makes
|
||||
# os.replace fail with PermissionError for a moment, so it is retried briefly.
|
||||
_REPLACE_ATTEMPTS = 5 if os.name == "nt" else 1
|
||||
_REPLACE_BACKOFF_SECONDS = 0.05
|
||||
|
||||
|
||||
class ConfigUnreadable(Exception):
|
||||
"""config.json exists but is not a readable JSON object; it is left alone."""
|
||||
|
||||
|
||||
class RevisionConflict(Exception):
|
||||
"""config.json changed since the revision the caller saw."""
|
||||
|
||||
def __init__(self, current: str):
|
||||
super().__init__("config.json changed since it was read")
|
||||
self.current = current
|
||||
|
||||
|
||||
def revision_of(data: Optional[bytes]) -> str:
|
||||
"""The revision of a file's bytes; ``None`` means the file is absent."""
|
||||
if data is None:
|
||||
return ABSENT_REVISION
|
||||
return "sha256:" + hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def _default_path() -> str:
|
||||
from .config import PromptManagerConfig
|
||||
|
||||
return PromptManagerConfig.get_config_path()
|
||||
|
||||
|
||||
def _replace(source: str, target: str) -> None:
|
||||
"""``os.replace`` with a short retry on a transient Windows PermissionError."""
|
||||
for attempt in range(1, _REPLACE_ATTEMPTS + 1):
|
||||
try:
|
||||
os.replace(source, target)
|
||||
return
|
||||
except PermissionError:
|
||||
if attempt == _REPLACE_ATTEMPTS:
|
||||
raise
|
||||
time.sleep(_REPLACE_BACKOFF_SECONDS)
|
||||
|
||||
|
||||
def _write_atomically(path: str, payload: bytes) -> None:
|
||||
"""Replace ``path`` with ``payload``: temp file beside it, fsync, replace.
|
||||
|
||||
The descriptor is closed before ``os.replace`` (Windows refuses to replace
|
||||
over an open handle), and on any failure the temp file is closed and
|
||||
removed, so a failed write leaves the old file and no stray temp file.
|
||||
``path`` is resolved first: a symlink keeps pointing at the updated
|
||||
target. A single-file bind mount cannot be replaced (EBUSY) and fails as
|
||||
an OSError, which callers report as not saved.
|
||||
"""
|
||||
path = os.path.realpath(path)
|
||||
parent = os.path.dirname(path) or "."
|
||||
os.makedirs(parent, exist_ok=True)
|
||||
fd, temp_path = tempfile.mkstemp(dir=parent, prefix=".config-", suffix=".tmp")
|
||||
try:
|
||||
try:
|
||||
handle = os.fdopen(fd, "wb")
|
||||
except BaseException:
|
||||
os.close(fd)
|
||||
raise
|
||||
with handle: # closed before the replace, also when a write fails
|
||||
handle.write(payload)
|
||||
handle.flush()
|
||||
os.fsync(handle.fileno())
|
||||
if os.name == "posix":
|
||||
os.chmod(temp_path, 0o600)
|
||||
_replace(temp_path, path)
|
||||
except BaseException:
|
||||
try:
|
||||
os.unlink(temp_path)
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
|
||||
|
||||
class ConfigStore:
|
||||
"""Serialized, atomic reads and commits of config.json."""
|
||||
|
||||
def __init__(self, path_getter: Optional[Callable[[], str]] = None):
|
||||
self._path_getter = path_getter or _default_path
|
||||
|
||||
@property
|
||||
def path(self) -> str:
|
||||
"""The config file this store reads and writes (resolved per call)."""
|
||||
return self._path_getter()
|
||||
|
||||
def read(self) -> Tuple[dict, str]:
|
||||
"""``(config, revision)``; a missing file is ``({}, "absent")``.
|
||||
|
||||
Raises ConfigUnreadable when the file cannot be read, is not JSON, or
|
||||
is not a JSON object. Waits for a running commit (see the module
|
||||
docstring).
|
||||
"""
|
||||
with _COMMIT_LOCK:
|
||||
return self._read(self.path)
|
||||
|
||||
@staticmethod
|
||||
def _read(path: str) -> Tuple[dict, str]:
|
||||
try:
|
||||
with open(path, "rb") as handle:
|
||||
raw = handle.read()
|
||||
except (FileNotFoundError, NotADirectoryError):
|
||||
# No file there (a parent that is a file cannot hold one either);
|
||||
# a commit then fails on the write, as config-write-failed.
|
||||
return {}, ABSENT_REVISION
|
||||
except OSError as exc:
|
||||
raise ConfigUnreadable(f"config.json cannot be read: {exc}") from exc
|
||||
try:
|
||||
data = json.loads(raw.decode("utf-8"))
|
||||
except (UnicodeDecodeError, ValueError) as exc:
|
||||
raise ConfigUnreadable("config.json is not valid JSON") from exc
|
||||
if not isinstance(data, dict):
|
||||
raise ConfigUnreadable("config.json is not a JSON object")
|
||||
return data, revision_of(raw)
|
||||
|
||||
def commit(
|
||||
self,
|
||||
mutate: Callable[[dict], dict],
|
||||
*,
|
||||
expected_revision: Optional[str],
|
||||
publish: Optional[Callable[[dict], None]] = None,
|
||||
) -> Tuple[dict, str]:
|
||||
"""Apply ``mutate`` to the current config and write the result.
|
||||
|
||||
``mutate`` receives a deep copy of the file's content and returns the
|
||||
whole new config (an object). ``expected_revision`` None merges into
|
||||
whatever is on disk (legacy partial updates); a revision refuses to
|
||||
write over different content. Returns ``(config, revision)`` as
|
||||
written. Raises ConfigUnreadable, RevisionConflict, TypeError (a
|
||||
non-object result) or OSError (the write failed; the old file stays).
|
||||
|
||||
``publish(config)``, when given, runs under the lock once the file
|
||||
holds the new config, so concurrent saves reach the runtime in the
|
||||
order they were written. It must handle its own errors: the values
|
||||
are already durable when it runs.
|
||||
"""
|
||||
with _COMMIT_LOCK:
|
||||
path = self.path
|
||||
current, revision = self._read(path)
|
||||
if expected_revision is not None and expected_revision != revision:
|
||||
raise RevisionConflict(revision)
|
||||
candidate = mutate(copy.deepcopy(current))
|
||||
if not isinstance(candidate, dict):
|
||||
raise TypeError("A config commit must produce a JSON object")
|
||||
payload = json.dumps(candidate, indent=2).encode("utf-8")
|
||||
_write_atomically(path, payload)
|
||||
if publish is not None:
|
||||
publish(candidate)
|
||||
return candidate, revision_of(payload)
|
||||
|
||||
|
||||
_store = ConfigStore()
|
||||
|
||||
|
||||
def get_config_store() -> ConfigStore:
|
||||
"""The process's config store (on ``PromptManagerConfig.get_config_path``)."""
|
||||
return _store
|
||||
@@ -74,6 +74,24 @@ test('a page whose leave was called off reads its list again on the next input',
|
||||
await expect(page.locator('#resultsTitle')).toHaveText('Recent Prompts');
|
||||
});
|
||||
|
||||
// The gallery's startup thumbnail check runs a second after load. When that second ends while the page is leaving,
|
||||
// the check must start no read: WebKit fails it with "Fetch API cannot load … due to access control checks." and
|
||||
// Playwright reports a page error (seen in CI on PR #289). The page clock makes the second end after the leave began.
|
||||
for (const leave of ['beforeunload', 'pagehide']) {
|
||||
test(`the gallery startup check starts no read after ${leave}`, async ({ page }) => {
|
||||
await page.clock.install();
|
||||
const statusReads = [];
|
||||
page.on('request', req => { if (req.url().includes('/images/thumbnails/status')) statusReads.push(req.url()); });
|
||||
await page.goto('/prompt_manager/gallery');
|
||||
await expect(page.locator('#galleryGrid .image-item')).toHaveCount(2);
|
||||
await page.evaluate(type => window.dispatchEvent(new Event(type)), leave);
|
||||
await page.clock.runFor(1500);
|
||||
// A read made after the clock ran is listed, so the listener saw every status read.
|
||||
await page.evaluate(() => fetch('/prompt_manager/images/thumbnails/status?probe=1'));
|
||||
expect(statusReads.map(url => new URL(url).search)).toEqual(['?probe=1']);
|
||||
});
|
||||
}
|
||||
|
||||
test('actual gallery review generates and persists tags for the selected root', async ({ page, request }) => {
|
||||
await page.goto('/prompt_manager/gallery');
|
||||
await expect(page.locator('#galleryGrid .image-item')).toHaveCount(2);
|
||||
|
||||
@@ -678,6 +678,31 @@ test('cancel scan ends cancelled with partial counts', async ({ page, request })
|
||||
await expect(heading(page).getByRole('button', { name: 'Cancel scan', exact: true })).not.toHaveAttribute('aria-disabled', 'true');
|
||||
});
|
||||
|
||||
// Truth rule: a held legacy-page job (a restore from the admin page) cannot be cancelled, so neither the Activity
|
||||
// list nor the job's view offers Cancel; the cancellable scan beside it keeps its own (review I3).
|
||||
test('a held legacy job shows no Cancel in the list or its view', async ({ page, request }) => {
|
||||
await page.setViewportSize({ width: 1600, height: 1100 });
|
||||
const response = await request.post('/__test__/jobs/hold', { data: { kind: 'restore', title: 'Database restore (admin page)' } });
|
||||
await expect(response).toBeOK();
|
||||
const held = (await response.json()).job;
|
||||
expect(held.cancellable).toBe(false);
|
||||
await page.goto(`${workspace}#/activity`);
|
||||
const item = items(page).filter({ has: page.getByRole('link', { name: /Database restore \(admin page\)$/ }) });
|
||||
await expect(item.locator('.chip')).toHaveText('Running');
|
||||
await expect(item.locator('progress')).toHaveCount(1);
|
||||
await expect(main(page).getByRole('button', { name: 'Cancel job', exact: true })).toHaveCount(0);
|
||||
|
||||
await item.getByRole('link').click();
|
||||
await expect(heading(page).locator('h1')).toHaveText('Database restore (admin page)');
|
||||
await expect(heading(page).locator('.heading-actions > button')).toHaveText(['Download report']);
|
||||
await expect(heading(page).getByRole('button', { name: /^Cancel/ })).toHaveCount(0);
|
||||
|
||||
// When the hold ends the view shows the end, still without a Cancel.
|
||||
await finish(request, held.id);
|
||||
await expect(heading(page).locator('p.muted')).toHaveText(/^Finished/);
|
||||
await expect(heading(page).locator('.heading-actions > button')).toHaveText(['Download report']);
|
||||
});
|
||||
|
||||
test('busy start names the running job', async ({ page, request }) => {
|
||||
await page.setViewportSize({ width: 1600, height: 1100 });
|
||||
const running = await startJob(request, { title: 'Database maintenance', kind: 'maintenance' });
|
||||
|
||||
@@ -14,6 +14,9 @@ reading files, so a test sets its progress and counters exactly:
|
||||
the progress to every subscriber;
|
||||
- ``POST /__test__/jobs/{id}/finish`` ends it as ``finished``, ``cancelled``
|
||||
or ``failed``; a cancel through the real route ends it ``cancelled``;
|
||||
- ``POST /__test__/jobs/hold`` opens a legacy-page hold (``kind``, ``title``):
|
||||
a running, non-cancellable ``records`` job, as a legacy restore or
|
||||
maintenance request takes, until ``finish`` ends it;
|
||||
- ``POST /__test__/jobs/restart`` acts out a ComfyUI restart: the open job
|
||||
streams drop without a final event and the registry starts empty.
|
||||
"""
|
||||
@@ -86,6 +89,7 @@ def register_activity_commands(app, routes, api):
|
||||
|
||||
streams = set()
|
||||
endings = {}
|
||||
holders = set()
|
||||
|
||||
@web.middleware
|
||||
async def track_streams(request, handler):
|
||||
@@ -163,6 +167,36 @@ def register_activity_commands(app, routes, api):
|
||||
endings[job.id] = ending
|
||||
return web.json_response({"success": True, "job": job.summary()})
|
||||
|
||||
@routes.post("/__test__/jobs/hold")
|
||||
async def hold_job(request):
|
||||
data = await body(request)
|
||||
loop = asyncio.get_running_loop()
|
||||
ending = loop.create_future()
|
||||
entered = loop.create_future()
|
||||
|
||||
async def hold():
|
||||
async with api._jobs.hold(
|
||||
str(data.get("kind", "restore")),
|
||||
str(data.get("title", "Database restore (admin page)")),
|
||||
admission="records",
|
||||
) as job:
|
||||
entered.set_result(job)
|
||||
state = await ending
|
||||
if state != "finished":
|
||||
job.finish(state)
|
||||
|
||||
holder = asyncio.ensure_future(hold())
|
||||
holders.add(holder)
|
||||
holder.add_done_callback(holders.discard)
|
||||
await asyncio.wait({entered, holder}, return_when=asyncio.FIRST_COMPLETED)
|
||||
if not entered.done():
|
||||
if isinstance(holder.exception(), JobBusy):
|
||||
raise web.HTTPConflict()
|
||||
raise holder.exception()
|
||||
job = entered.result()
|
||||
endings[job.id] = ending
|
||||
return web.json_response({"success": True, "job": job.summary()})
|
||||
|
||||
@routes.post("/__test__/jobs/{job_id}/step")
|
||||
async def step_job(request):
|
||||
job = job_or_404(request)
|
||||
@@ -210,6 +244,7 @@ def register_activity_commands(app, routes, api):
|
||||
for transport in list(streams):
|
||||
transport.close()
|
||||
tasks = [job.task for job in old.list() if job.task and not job.task.done()]
|
||||
tasks += [holder for holder in holders if not holder.done()]
|
||||
for task in tasks:
|
||||
task.cancel()
|
||||
await asyncio.gather(*tasks, return_exceptions=True)
|
||||
|
||||
@@ -73,6 +73,28 @@ test('Not Now persists the collection state and suppresses the unchanged prompt'
|
||||
assert.equal(f.prompts.length, 1);
|
||||
});
|
||||
|
||||
// WebKit fails a fetch a leaving page starts ("Fetch API cannot load … due to access control checks.") and logs it
|
||||
// as a page error: the delayed startup check starts no read once the page has begun to leave (as admin.js, #290).
|
||||
test('the startup check starts no read once the page is leaving', async () => {
|
||||
const f = fixture({ success: true, total_images: 100, pending: 30, collection_id: 'a' });
|
||||
f.manager.pageLifetime = new AbortController();
|
||||
f.manager.pageLifetime.abort();
|
||||
await f.check();
|
||||
assert.equal(f.requests.length, 0);
|
||||
assert.equal(f.prompts.length, 0);
|
||||
});
|
||||
|
||||
test('the startup check passes the page lifetime to its read', async () => {
|
||||
const f = fixture({ success: true, total_images: 100, pending: 30, collection_id: 'a' });
|
||||
const signals = [];
|
||||
f.manager.pageLifetime = new AbortController();
|
||||
const get = f.manager.api.get;
|
||||
f.manager.api.get = async (url, options) => { signals.push(options?.signal); return get(url); };
|
||||
await f.check();
|
||||
assert.deepEqual(signals, [f.manager.pageLifetime.signal]);
|
||||
assert.equal(f.prompts.length, 1);
|
||||
});
|
||||
|
||||
test('disabled checks do not request collection status', async () => {
|
||||
const f = fixture({}, { checkThumbnailsAtStartup: false });
|
||||
await f.check();
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
// The legacy admin and gallery pages show a server refusal's reason (#220, PR #289 review I1 and M8): a job that
|
||||
// runs in the workspace refuses their record-changing requests, and a corrupt config.json refuses a settings save.
|
||||
// Each page must name the reason, not report a lost connection, a bare HTTP status or a success.
|
||||
const test = require('node:test');
|
||||
const assert = require('node:assert/strict');
|
||||
const fs = require('node:fs');
|
||||
const path = require('node:path');
|
||||
const vm = require('node:vm');
|
||||
|
||||
function elements() {
|
||||
const nodes = {};
|
||||
return {
|
||||
nodes,
|
||||
getElementById: id => nodes[id] ||= { id, textContent: '', innerHTML: '', disabled: false, style: {},
|
||||
value: '', checked: false, addEventListener() {}, querySelectorAll: () => [] },
|
||||
};
|
||||
}
|
||||
|
||||
function jsonResponse(status, body) {
|
||||
return { ok: status >= 200 && status < 300, status, statusText: status === 409 ? 'Conflict' : 'OK',
|
||||
json: async () => body };
|
||||
}
|
||||
|
||||
function sseResponse(frames) {
|
||||
const bytes = new TextEncoder().encode(frames.map(frame => `data: ${JSON.stringify(frame)}\n\n`).join(''));
|
||||
let sent = false;
|
||||
return { ok: true, status: 200, body: { getReader: () => ({
|
||||
read: async () => (sent ? { done: true } : (sent = true, { done: false, value: bytes })),
|
||||
}) } };
|
||||
}
|
||||
|
||||
function load(file, className, fetchUrl, extra = {}, dom = elements()) {
|
||||
const timers = [];
|
||||
const notifications = [];
|
||||
const storage = new Map([['pm_cached', '1'], ['userData', 'keep']]);
|
||||
const localStorage = {
|
||||
getItem: key => storage.get(key) ?? null,
|
||||
setItem: (key, value) => storage.set(key, String(value)),
|
||||
removeItem: key => storage.delete(key),
|
||||
};
|
||||
const context = vm.createContext({
|
||||
document: { readyState: 'loading', addEventListener() {}, getElementById: dom.getElementById,
|
||||
querySelectorAll: () => [] },
|
||||
window: {}, console: { error() {}, warn() {}, debug() {} },
|
||||
setTimeout: callback => timers.push(callback),
|
||||
confirm: () => true,
|
||||
localStorage,
|
||||
Object: { ...Object, keys: target => (target === localStorage ? [...storage.keys()] : Object.keys(target)) },
|
||||
TextDecoder, TextEncoder, JSON,
|
||||
ApiClient: { fetchUrl },
|
||||
...extra,
|
||||
});
|
||||
const source = fs.readFileSync(path.join(__dirname, `../../web/js/${file}`), 'utf8');
|
||||
const Page = vm.runInContext(`${source}\n${className};`, context, { filename: path.join(__dirname, `../../web/js/${file}`) });
|
||||
const page = Object.create(Page.prototype);
|
||||
page.showNotification = (message, type) => notifications.push([type, message]);
|
||||
page.showModal = id => notifications.push(['show', id]);
|
||||
page.hideModal = id => notifications.push(['hide', id]);
|
||||
return { page, dom, timers, notifications, storage };
|
||||
}
|
||||
|
||||
const BUSY = 'Scan for metadata already running';
|
||||
|
||||
test('gallery cache clear reports a refused server clear and never claims success', async () => {
|
||||
const calls = [];
|
||||
const { page, dom, notifications, storage } = load('gallery.js', 'GalleryManager', async url => {
|
||||
calls.push(url);
|
||||
return jsonResponse(409, { success: false, error: BUSY });
|
||||
});
|
||||
page.getSettings = () => ({ thumbnailsGenerated: true });
|
||||
dom.getElementById('thumbnailStatus').textContent = 'Generated';
|
||||
|
||||
await page.clearCache();
|
||||
|
||||
assert.deepEqual(calls, ['/prompt_manager/images/clear-thumbnails']);
|
||||
assert.deepEqual(notifications, [['error', `Thumbnails were not cleared: ${BUSY}`]]);
|
||||
assert.equal(dom.nodes.thumbnailStatus.textContent, 'Generated');
|
||||
assert.equal(storage.get('gallerySettings'), undefined); // thumbnailsGenerated is not reset
|
||||
});
|
||||
|
||||
test('gallery cache clear still says so when the server clears', async () => {
|
||||
const { page, dom, notifications, storage } = load('gallery.js', 'GalleryManager',
|
||||
async () => jsonResponse(200, { success: true, cleared_count: 3 }));
|
||||
page.getSettings = () => ({ thumbnailsGenerated: true });
|
||||
|
||||
await page.clearCache();
|
||||
|
||||
assert.deepEqual(notifications, [['success', 'PromptManager cache cleared successfully! Original images untouched.']]);
|
||||
assert.equal(dom.nodes.thumbnailStatus.textContent, 'Not generated');
|
||||
assert.equal(JSON.parse(storage.get('gallerySettings')).thumbnailsGenerated, false);
|
||||
assert.equal(storage.get('userData'), 'keep');
|
||||
assert.equal(storage.has('pm_cached'), false);
|
||||
});
|
||||
|
||||
test('admin LoRA import shows a refusal frame and closes its modal', async () => {
|
||||
const { page, dom, timers, notifications } = load('admin.js', 'PromptAdmin',
|
||||
async () => sseResponse([{ type: 'error', status: BUSY, progress: 0, message: BUSY }]));
|
||||
page.saveLoraSettings = async () => true;
|
||||
page.loadStatistics = () => assert.fail('a refused import must not reload statistics');
|
||||
|
||||
await page.runLoraImport();
|
||||
|
||||
assert.equal(dom.nodes.loraImportStatus.textContent, `Error: ${BUSY}`);
|
||||
assert.deepEqual(notifications.filter(([type]) => type === 'error'), [['error', `LoRA import failed: ${BUSY}`]]);
|
||||
assert.equal(timers.length, 1);
|
||||
timers[0]();
|
||||
assert.deepEqual(notifications.at(-1), ['hide', 'loraImportModal']);
|
||||
});
|
||||
|
||||
test('admin settings save shows why config.json refused it', async () => {
|
||||
const message = 'config.json cannot be read, so nothing was saved. Fix or remove the file, then save again.';
|
||||
const { page, notifications } = load('admin.js', 'PromptAdmin',
|
||||
async () => jsonResponse(409, { success: false, persisted: false, code: 'config-unreadable', error: message }));
|
||||
page.settings = {};
|
||||
page._collectScanPaths = () => [];
|
||||
page.saveLoraSettings = async () => assert.fail('nothing else is saved after a refusal');
|
||||
|
||||
await page.saveSettings();
|
||||
|
||||
assert.deepEqual(notifications, [['error', `Failed to save settings: ${message}`]]);
|
||||
});
|
||||
|
||||
test('admin settings save keeps its plain message when the reply has no reason', async () => {
|
||||
const { page, notifications } = load('admin.js', 'PromptAdmin', async () => ({ ok: false, status: 500,
|
||||
json: async () => { throw new SyntaxError('not json'); } }));
|
||||
page.settings = {};
|
||||
page._collectScanPaths = () => [];
|
||||
|
||||
await page.saveSettings();
|
||||
|
||||
assert.deepEqual(notifications, [['error', 'Failed to save settings']]);
|
||||
});
|
||||
|
||||
test('admin maintenance names the job that refused it', async () => {
|
||||
const { page, dom, notifications } = load('admin.js', 'PromptAdmin',
|
||||
async () => jsonResponse(409, { success: false, error: BUSY }), {
|
||||
escapeHtml: text => String(text),
|
||||
document: { readyState: 'loading', addEventListener() {}, getElementById: id => dom.getElementById(id),
|
||||
querySelectorAll: selector => (selector === '.maintenance-option:checked' ? [{ id: 'vacuum' }] : []) },
|
||||
});
|
||||
|
||||
await page.runMaintenance();
|
||||
|
||||
const results = dom.nodes.maintenanceResults.innerHTML;
|
||||
assert.match(results, /Maintenance Failed/);
|
||||
assert.match(results, new RegExp(BUSY));
|
||||
assert.doesNotMatch(results, /HTTP 409/);
|
||||
assert.deepEqual(notifications, [['error', '❌ Maintenance failed']]);
|
||||
assert.equal(dom.nodes.runMaintenanceBtn.disabled, false);
|
||||
});
|
||||
@@ -261,8 +261,13 @@ test('summaries are validated and the copy follows the spec', async () => {
|
||||
const loose = parseJob({ id: ID_A, state: 'running', progress: 140.4, processed: -1, total: 1.5, counters: { added: 'x', errors: 2 },
|
||||
title: ' ', started_at: 'x', scope: [] });
|
||||
assert.deepEqual(loose, { id: ID_A, kind: '', title: 'Background job', state: 'running', started_at: null, finished_at: null,
|
||||
progress: 100, status_line: '', processed: 0, total: 0, counters: { added: 0, errors: 2 }, scope: {}, listed: {}, log_length: 0 });
|
||||
progress: 100, status_line: '', processed: 0, total: 0, counters: { added: 0, errors: 2 }, scope: {}, listed: {}, log_length: 0,
|
||||
cancellable: true });
|
||||
assert.equal(parseJob(summary({ progress: -3 })).progress, 0);
|
||||
// A held legacy job cannot be cancelled; an older server sends no key, which keeps Cancel.
|
||||
assert.equal(parseJob(summary({ cancellable: false })).cancellable, false);
|
||||
assert.equal(parseJob(summary({ cancellable: true })).cancellable, true);
|
||||
assert.equal(parseJob(summary()).cancellable, true);
|
||||
|
||||
const scan = parseJob(summary({ progress: 42, processed: 8240, total: 19622, counters: { added: 1, errors: 1 } }));
|
||||
assert.deepEqual(pillParts(scan), { label: 'Scanning · 42%', files: ' · 8,240 / 19,622', text: 'Scanning · 42% · 8,240 / 19,622' });
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
"""A lock stand-in that reports when a second caller has to wait for it.
|
||||
|
||||
Concurrency tests patch it over a module's lock so they can wait until the
|
||||
second caller is really blocked on the lock, instead of hoping a thread
|
||||
start or one event-loop tick got it there.
|
||||
"""
|
||||
|
||||
import threading
|
||||
|
||||
|
||||
class ProbeLock:
|
||||
"""A re-entrant lock whose ``contended`` event is set when a caller blocks."""
|
||||
|
||||
def __init__(self):
|
||||
self._lock = threading.RLock()
|
||||
self.contended = threading.Event()
|
||||
|
||||
def acquire(self, blocking=True, timeout=-1):
|
||||
if self._lock.acquire(blocking=False):
|
||||
return True
|
||||
self.contended.set()
|
||||
if not blocking:
|
||||
return False
|
||||
return self._lock.acquire(timeout=timeout)
|
||||
|
||||
def release(self):
|
||||
self._lock.release()
|
||||
|
||||
def __enter__(self):
|
||||
self.acquire()
|
||||
return self
|
||||
|
||||
def __exit__(self, *exc):
|
||||
self.release()
|
||||
return False
|
||||
+169
-9
@@ -5,6 +5,7 @@ database and a stub ``folder_paths`` module, so the tests run without ComfyUI.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
@@ -734,9 +735,9 @@ class TestSettingsReadMonitorThroughModule(AdminAPITestCase):
|
||||
def test_no_sys_modules_scan(self):
|
||||
import inspect
|
||||
|
||||
from py.api.admin import AdminRoutesMixin
|
||||
from py.api.settings_routes import SettingsRoutesMixin
|
||||
|
||||
source = inspect.getsource(AdminRoutesMixin.get_settings)
|
||||
source = inspect.getsource(SettingsRoutesMixin.get_settings)
|
||||
self.assertNotIn("sys.modules", source)
|
||||
|
||||
async def test_reports_directories_of_running_monitor(self):
|
||||
@@ -1059,14 +1060,29 @@ class TestSettingsMisc(AdminAPITestCase):
|
||||
saved["gallery"]["monitoring"]["directories"], [self._root(self.output_dir)]
|
||||
)
|
||||
|
||||
async def test_save_replaces_a_corrupt_config_file(self):
|
||||
with open(self.config_path, "w") as f:
|
||||
f.write("{not json")
|
||||
async def test_save_refuses_a_corrupt_config_file(self):
|
||||
# #220: a config.json that cannot be parsed is the user's file, never
|
||||
# silently replaced; the save is refused and nothing is published.
|
||||
Path(self.config_path).write_bytes(b"{not json")
|
||||
before_timeout = PromptManagerConfig.RESULT_TIMEOUT
|
||||
resp = await self.client.request(
|
||||
"POST", "/prompt_manager/settings", json={"result_timeout": 4}
|
||||
)
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertEqual(self._read_config()["web_ui"]["result_timeout"], 4)
|
||||
body = await resp.json()
|
||||
self.assertEqual((resp.status, body["code"]), (409, "config-unreadable"))
|
||||
self.assertIs(body["success"], False)
|
||||
self.assertIs(body["persisted"], False)
|
||||
self.assertEqual(PromptManagerConfig.RESULT_TIMEOUT, before_timeout)
|
||||
self.assertEqual(Path(self.config_path).read_bytes(), b"{not json")
|
||||
|
||||
async def test_get_settings_reports_an_unreadable_config(self):
|
||||
Path(self.config_path).write_bytes(b"{not json")
|
||||
resp = await self.client.request("GET", "/prompt_manager/settings")
|
||||
body = await resp.json()
|
||||
self.assertEqual(resp.status, 200) # the legacy page still loads
|
||||
self.assertIsNone(body["revision"])
|
||||
self.assertEqual(body["config_error"], "config-unreadable")
|
||||
self.assertIn("result_timeout", body["settings"])
|
||||
|
||||
async def test_saved_config_file_is_private_to_the_user(self):
|
||||
import stat
|
||||
@@ -1107,12 +1123,156 @@ class TestSettingsMisc(AdminAPITestCase):
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertFalse(data["restart_required"])
|
||||
|
||||
async def test_unwritable_config_path_still_succeeds(self):
|
||||
os.environ[CONFIG_PATH_ENV] = os.path.join(self.tmpdir, "missing", "c.json")
|
||||
async def test_unwritable_config_path_reports_failure_without_runtime_change(self):
|
||||
# #220: the parent is a file, so config.json cannot be written. The
|
||||
# save reports the failure and publishes nothing.
|
||||
blocker = Path(self.tmpdir) / "not-a-dir"
|
||||
blocker.write_text("x")
|
||||
os.environ[CONFIG_PATH_ENV] = str(blocker / "c.json")
|
||||
GalleryConfig.MONITORING_DIRECTORIES = []
|
||||
before_timeout = PromptManagerConfig.RESULT_TIMEOUT
|
||||
with self.assertLogs("prompt_manager.api", level="ERROR"):
|
||||
resp = await self.client.request(
|
||||
"POST",
|
||||
"/prompt_manager/settings",
|
||||
json={
|
||||
"result_timeout": before_timeout + 1,
|
||||
"gallery_root_paths": [self._root(self.output_dir)],
|
||||
},
|
||||
)
|
||||
body = await resp.json()
|
||||
self.assertEqual(
|
||||
(resp.status, body["code"], body["persisted"]),
|
||||
(500, "config-write-failed", False),
|
||||
)
|
||||
self.assertIs(body["success"], False)
|
||||
self.assertIs(body["runtime_applied"], False)
|
||||
self.assertEqual(PromptManagerConfig.RESULT_TIMEOUT, before_timeout)
|
||||
self.assertEqual(GalleryConfig.MONITORING_DIRECTORIES, [])
|
||||
self.assertEqual(blocker.read_text(), "x")
|
||||
|
||||
async def test_save_reports_the_commit(self):
|
||||
resp = await self.client.request(
|
||||
"POST", "/prompt_manager/settings", json={"result_timeout": 3}
|
||||
)
|
||||
body = await resp.json()
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertEqual(body["message"], "Settings saved successfully")
|
||||
self.assertIs(body["persisted"], True)
|
||||
self.assertIs(body["runtime_applied"], True)
|
||||
self.assertIs(body["restart_required"], False)
|
||||
self.assertEqual(body["restart_required_fields"], [])
|
||||
digest = hashlib.sha256(Path(self.config_path).read_bytes()).hexdigest()
|
||||
self.assertEqual(body["revision"], "sha256:" + digest)
|
||||
|
||||
async def test_changed_roots_name_the_restart_field(self):
|
||||
resp = await self.client.request(
|
||||
"POST",
|
||||
"/prompt_manager/settings",
|
||||
json={"gallery_root_paths": [self._root(self.output_dir)]},
|
||||
)
|
||||
body = await resp.json()
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertIs(body["restart_required"], True)
|
||||
self.assertEqual(body["restart_required_fields"], ["gallery_root_paths"])
|
||||
|
||||
async def test_runtime_publication_failure_after_commit_is_reported(self):
|
||||
# The values are durable; the response must not claim they are live.
|
||||
self.api.reset_gallery_roots_cache = _raise
|
||||
with self.assertLogs("prompt_manager.api", level="ERROR"):
|
||||
resp = await self.client.request(
|
||||
"POST",
|
||||
"/prompt_manager/settings",
|
||||
json={"gallery_root_paths": [self._root(self.output_dir)]},
|
||||
)
|
||||
body = await resp.json()
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertIs(body["persisted"], True)
|
||||
self.assertIs(body["runtime_applied"], False)
|
||||
self.assertIs(body["restart_required"], True)
|
||||
self.assertEqual(body["restart_required_fields"], ["gallery_root_paths"])
|
||||
self.assertEqual(
|
||||
self._read_config()["gallery"]["monitoring"]["directories"],
|
||||
[self._root(self.output_dir)],
|
||||
)
|
||||
|
||||
async def test_get_settings_reports_revision(self):
|
||||
resp = await self.client.request("GET", "/prompt_manager/settings")
|
||||
body = await resp.json()
|
||||
self.assertEqual(body["revision"], "absent")
|
||||
self.assertNotIn("config_error", body)
|
||||
|
||||
self._write_config({"web_ui": {"result_timeout": 2}})
|
||||
body = await (
|
||||
await self.client.request("GET", "/prompt_manager/settings")
|
||||
).json()
|
||||
digest = hashlib.sha256(Path(self.config_path).read_bytes()).hexdigest()
|
||||
self.assertEqual(body["revision"], "sha256:" + digest)
|
||||
|
||||
async def test_native_save_with_stale_revision_is_409(self):
|
||||
self._write_config({"web_ui": {"result_timeout": 2}})
|
||||
body = await (
|
||||
await self.client.request("GET", "/prompt_manager/settings")
|
||||
).json()
|
||||
seen = body["revision"]
|
||||
self._write_config({"web_ui": {"result_timeout": 6}}) # another writer
|
||||
before = Path(self.config_path).read_bytes()
|
||||
before_timeout = PromptManagerConfig.RESULT_TIMEOUT
|
||||
|
||||
resp = await self.client.request(
|
||||
"POST",
|
||||
"/prompt_manager/settings",
|
||||
json={"result_timeout": 9, "expected_revision": seen},
|
||||
)
|
||||
body = await resp.json()
|
||||
self.assertEqual((resp.status, body["code"]), (409, "settings-changed"))
|
||||
self.assertIs(body["persisted"], False)
|
||||
digest = hashlib.sha256(before).hexdigest()
|
||||
self.assertEqual(body["revision"], "sha256:" + digest)
|
||||
self.assertEqual(Path(self.config_path).read_bytes(), before)
|
||||
self.assertEqual(PromptManagerConfig.RESULT_TIMEOUT, before_timeout)
|
||||
|
||||
async def test_native_save_with_current_revision_commits(self):
|
||||
body = await (
|
||||
await self.client.request("GET", "/prompt_manager/settings")
|
||||
).json()
|
||||
resp = await self.client.request(
|
||||
"POST",
|
||||
"/prompt_manager/settings",
|
||||
json={"result_timeout": 9, "expected_revision": body["revision"]},
|
||||
)
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertEqual(self._read_config()["web_ui"]["result_timeout"], 9)
|
||||
|
||||
async def test_expected_revision_must_be_a_string(self):
|
||||
resp = await self.client.request(
|
||||
"POST",
|
||||
"/prompt_manager/settings",
|
||||
json={"result_timeout": 9, "expected_revision": 5},
|
||||
)
|
||||
self.assertEqual(resp.status, 400)
|
||||
self.assertFalse(os.path.exists(self.config_path))
|
||||
|
||||
async def test_legacy_partial_update_without_revision_still_merges(self):
|
||||
self._write_config(
|
||||
{
|
||||
"integrations": {"lora_manager": {"civitai_api_key": "secret"}},
|
||||
"web_ui": {"result_timeout": 1, "webui_display_mode": "popup"},
|
||||
"custom_section": {"keep": True},
|
||||
}
|
||||
)
|
||||
resp = await self.client.request(
|
||||
"POST", "/prompt_manager/settings", json={"result_timeout": 8}
|
||||
)
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertNotIn("secret", await resp.text())
|
||||
saved = self._read_config()
|
||||
self.assertEqual(
|
||||
saved["integrations"]["lora_manager"]["civitai_api_key"], "secret"
|
||||
)
|
||||
self.assertEqual(saved["custom_section"], {"keep": True})
|
||||
self.assertEqual(saved["web_ui"]["result_timeout"], 8)
|
||||
self.assertEqual(saved["web_ui"]["webui_display_mode"], "popup") # not sent
|
||||
|
||||
async def test_save_unexpected_error_is_500(self):
|
||||
self.api._parse_gallery_roots = _raise
|
||||
|
||||
@@ -5,12 +5,16 @@ stream the upload to disk, reject anything that is not a healthy prompts
|
||||
database and leave the live database untouched when it does.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import glob
|
||||
import hashlib
|
||||
import os
|
||||
import sqlite3
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
_mock_server = MagicMock()
|
||||
@@ -142,6 +146,85 @@ class TestBackupRoute(BackupRestoreTestCase):
|
||||
self.assertIn("backup", data["error"].lower())
|
||||
|
||||
|
||||
class TestRestoreHoldsTheRegistry(BackupRestoreTestCase):
|
||||
"""A legacy restore and a workspace job refuse each other (#269 gap)."""
|
||||
|
||||
def _digest(self):
|
||||
return {
|
||||
suffix: hashlib.sha256(Path(self.db_path + suffix).read_bytes()).hexdigest()
|
||||
for suffix in ("", "-wal")
|
||||
if os.path.exists(self.db_path + suffix)
|
||||
}
|
||||
|
||||
async def test_legacy_restore_refused_while_scan_job_runs(self):
|
||||
self._save("keep me")
|
||||
upload = self._make_db_file(["restored"])
|
||||
gate = asyncio.Event()
|
||||
|
||||
async def runner(job):
|
||||
await gate.wait()
|
||||
|
||||
scan = self.api._jobs.start(
|
||||
"scan", "Scan for metadata", record_changing=True, scope={}, runner=runner
|
||||
)
|
||||
before = self._digest()
|
||||
try:
|
||||
resp = await self._restore(upload)
|
||||
self.assertEqual(resp.status, 409)
|
||||
self.assertEqual(
|
||||
await resp.json(),
|
||||
{"success": False, "error": "Scan for metadata already running"},
|
||||
)
|
||||
self.assertEqual(self._digest(), before)
|
||||
self.assertEqual(
|
||||
[p["text"] for p in self.api.db.search_prompts()], ["keep me"]
|
||||
)
|
||||
self.assertEqual(glob.glob(self.db_path + ".backup_*"), [])
|
||||
finally:
|
||||
gate.set()
|
||||
await asyncio.wait_for(scan.task, 5)
|
||||
|
||||
async def test_scan_refused_while_legacy_restore_holds(self):
|
||||
upload = self._make_db_file(["restored"])
|
||||
entered = threading.Event()
|
||||
release = threading.Event()
|
||||
model = self.api.db.model
|
||||
real_verify = model.verify_database_file
|
||||
|
||||
def gated(path, **kwargs):
|
||||
entered.set()
|
||||
release.wait(10)
|
||||
return real_verify(path, **kwargs)
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
with patch.object(model, "verify_database_file", gated):
|
||||
restore = asyncio.ensure_future(self._restore(upload))
|
||||
self.assertTrue(await loop.run_in_executor(None, entered.wait, 10))
|
||||
resp = await self.client.post("/prompt_manager/jobs/scan")
|
||||
body = await resp.json()
|
||||
listed = (await (await self.client.get("/prompt_manager/jobs")).json())[
|
||||
"jobs"
|
||||
]
|
||||
release.set()
|
||||
restored = await restore
|
||||
|
||||
self.assertEqual((resp.status, body["code"]), (409, "busy"))
|
||||
self.assertEqual(body["running"]["title"], "Database restore (admin page)")
|
||||
self.assertEqual(
|
||||
[(j["kind"], j["state"], j["cancellable"]) for j in listed],
|
||||
[("restore", "running", False)],
|
||||
)
|
||||
self.assertEqual(restored.status, 200)
|
||||
self.assertEqual(self.api._jobs.list()[0].state, "finished")
|
||||
self.assertIsNone(self.api._jobs.active("restore"))
|
||||
|
||||
async def test_rejected_upload_ends_its_hold_failed(self):
|
||||
resp = await self._restore(self._write("notes.txt", b"not a database"))
|
||||
self.assertEqual(resp.status, 400)
|
||||
job = self.api._jobs.list()[0]
|
||||
self.assertEqual((job.kind, job.state), ("restore", "failed"))
|
||||
|
||||
|
||||
class TestRestoreRoute(BackupRestoreTestCase):
|
||||
async def test_restore_replaces_database_and_reports_backup(self):
|
||||
self._save("old prompt")
|
||||
|
||||
@@ -5,11 +5,13 @@ Uses aiohttp's test client against a real PromptManagerAPI wired to a
|
||||
temporary SQLite database and a temporary output directory.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import pathlib
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from contextlib import ExitStack
|
||||
from pathlib import Path
|
||||
@@ -638,6 +640,93 @@ class TestGenerateThumbnailsPost(ThumbnailTestCase):
|
||||
self.assertEqual(data["total_images"], 0)
|
||||
|
||||
|
||||
class TestThumbnailRoutesHoldTheRegistry(ThumbnailTestCase):
|
||||
"""Thumbnail work is ``guarded``: never beside a record-changing job."""
|
||||
|
||||
def _run_scan(self):
|
||||
gate = asyncio.Event()
|
||||
|
||||
async def runner(job):
|
||||
await gate.wait()
|
||||
|
||||
job = self.api._jobs.start(
|
||||
"scan", "Scan for metadata", admission="records", scope={}, runner=runner
|
||||
)
|
||||
|
||||
async def release():
|
||||
gate.set()
|
||||
await asyncio.wait_for(job.task, 5)
|
||||
|
||||
self.addAsyncCleanup(release)
|
||||
return job
|
||||
|
||||
async def test_thumbnail_generate_refused_while_scan_runs(self):
|
||||
make_png(self.output_dir / "a.png")
|
||||
self._run_scan()
|
||||
|
||||
resp = await self.client.request(
|
||||
"POST", "/prompt_manager/images/generate-thumbnails", json={}
|
||||
)
|
||||
self.assertEqual(resp.status, 409)
|
||||
self.assertEqual(
|
||||
await resp.json(),
|
||||
{"success": False, "error": "Scan for metadata already running"},
|
||||
)
|
||||
|
||||
resp = await self.client.request("POST", self.PROGRESS_URL)
|
||||
text = await resp.text()
|
||||
self.assertEqual(resp.status, 200) # the gallery reads an SSE stream
|
||||
self.assertEqual(
|
||||
text,
|
||||
"event: error\ndata: "
|
||||
+ json.dumps(
|
||||
{
|
||||
"error": "Scan for metadata already running",
|
||||
"message": "Scan for metadata already running",
|
||||
}
|
||||
)
|
||||
+ "\n\n",
|
||||
)
|
||||
|
||||
thumbs = self.output_dir / "thumbnails"
|
||||
make_png(thumbs / "b_thumb.png")
|
||||
resp = await self.client.request(
|
||||
"POST", "/prompt_manager/images/clear-thumbnails"
|
||||
)
|
||||
self.assertEqual(resp.status, 409)
|
||||
self.assertFalse(self._thumb_path("a").exists())
|
||||
self.assertTrue((thumbs / "b_thumb.png").exists())
|
||||
|
||||
async def test_scan_refused_while_thumbnails_generate(self):
|
||||
make_png(self.output_dir / "a.png")
|
||||
entered = threading.Event()
|
||||
release = threading.Event()
|
||||
real = self.api._generate_gallery_thumbnails_sync
|
||||
|
||||
def gated(*args):
|
||||
entered.set()
|
||||
release.wait(10)
|
||||
return real(*args)
|
||||
|
||||
self.api._generate_gallery_thumbnails_sync = gated
|
||||
loop = asyncio.get_running_loop()
|
||||
generate = asyncio.ensure_future(
|
||||
self.client.request(
|
||||
"POST", "/prompt_manager/images/generate-thumbnails", json={}
|
||||
)
|
||||
)
|
||||
self.assertTrue(await loop.run_in_executor(None, entered.wait, 10))
|
||||
resp = await self.client.request("POST", "/prompt_manager/jobs/scan")
|
||||
body = await resp.json()
|
||||
self.assertEqual((resp.status, body["code"]), (409, "busy"))
|
||||
self.assertEqual(
|
||||
body["running"]["title"], "Thumbnail generation (gallery page)"
|
||||
)
|
||||
release.set()
|
||||
self.assertEqual((await generate).status, 200)
|
||||
self.assertTrue(self._thumb_path("a").exists())
|
||||
|
||||
|
||||
class TestRecentImagesCaps(ImageAPITestCase):
|
||||
"""Every path through /images/recent hands the DB a bounded window."""
|
||||
|
||||
|
||||
+340
-12
@@ -4,11 +4,13 @@ Filesystem detection and network downloads are patched out so the tests run
|
||||
anywhere without a LoraManager install or a CivitAI connection.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import types
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
@@ -34,11 +36,20 @@ from aiohttp.test_utils import AioHTTPTestCase
|
||||
|
||||
from database.operations import PromptDatabase
|
||||
from py.api import PromptManagerAPI
|
||||
from py import lora_utils
|
||||
from py import config_store, lora_utils
|
||||
from py.config import IntegrationConfig, PromptManagerConfig
|
||||
from py.api import settings_routes as settings_module
|
||||
from py.api import lora_integration as lora_module
|
||||
from py.lora_utils import get_trigger_cache
|
||||
from utils.hashing import generate_prompt_hash
|
||||
|
||||
try:
|
||||
from tests.lock_probe import ProbeLock
|
||||
from tests.open_handles import assert_closed
|
||||
except ImportError: # discovered with tests/ as the top-level directory
|
||||
from lock_probe import ProbeLock
|
||||
from open_handles import assert_closed
|
||||
|
||||
SECRET_KEY = "sk-civitai-super-secret-0123456789"
|
||||
|
||||
|
||||
@@ -56,7 +67,13 @@ class LoraAPITestCase(AioHTTPTestCase):
|
||||
return_value=[self.custom_nodes],
|
||||
)
|
||||
self._custom_nodes_patch.start()
|
||||
self._real_save = PromptManagerConfig.save_to_file
|
||||
# Never the repository's config.json (live user data): every save in
|
||||
# these tests goes to a file in this test's temp directory.
|
||||
self.config_file = self.tmp_root / "config.json"
|
||||
self._env_patch = patch.dict(
|
||||
os.environ, {"PROMPT_MANAGER_CONFIG_PATH": str(self.config_file)}
|
||||
)
|
||||
self._env_patch.start()
|
||||
|
||||
self._config_patch = patch.multiple(
|
||||
IntegrationConfig,
|
||||
@@ -70,8 +87,6 @@ class LoraAPITestCase(AioHTTPTestCase):
|
||||
"py.lora_utils.detect_lora_manager", return_value=None
|
||||
)
|
||||
self.detect_mock = self._detect_patch.start()
|
||||
self._save_patch = patch.object(PromptManagerConfig, "save_to_file")
|
||||
self.save_mock = self._save_patch.start()
|
||||
get_trigger_cache().clear()
|
||||
|
||||
app = web.Application()
|
||||
@@ -83,7 +98,7 @@ class LoraAPITestCase(AioHTTPTestCase):
|
||||
|
||||
async def tearDownAsync(self):
|
||||
await super().tearDownAsync() # closes the aiohttp test client
|
||||
self._save_patch.stop()
|
||||
self._env_patch.stop()
|
||||
self._detect_patch.stop()
|
||||
self._config_patch.stop()
|
||||
self._custom_nodes_patch.stop()
|
||||
@@ -96,8 +111,19 @@ class LoraAPITestCase(AioHTTPTestCase):
|
||||
):
|
||||
if os.path.exists(path):
|
||||
os.unlink(path)
|
||||
assert_closed(self, self.tmp_root)
|
||||
shutil.rmtree(self.tmp_root, ignore_errors=True)
|
||||
|
||||
def _saved(self):
|
||||
return json.loads(self.config_file.read_text())
|
||||
|
||||
def _make_unwritable(self):
|
||||
"""Point the config path under a regular file, so no write can succeed."""
|
||||
blocker = self.tmp_root / "not-a-dir"
|
||||
blocker.write_text("x")
|
||||
self.config_file = blocker / "config.json"
|
||||
os.environ["PROMPT_MANAGER_CONFIG_PATH"] = str(self.config_file)
|
||||
|
||||
async def _post_json(self, url, body):
|
||||
return await self.client.request(
|
||||
"POST",
|
||||
@@ -250,7 +276,7 @@ class TestLoraEnableRoute(LoraAPITestCase):
|
||||
)
|
||||
self.assertEqual(resp.status, 400)
|
||||
self.assertFalse(IntegrationConfig.LORA_MANAGER_ENABLED)
|
||||
self.save_mock.assert_not_called()
|
||||
self.assertFalse(self.config_file.exists())
|
||||
|
||||
async def test_enable_persists_and_loads_trigger_cache(self):
|
||||
lm = _make_lora_manager(self.custom_nodes)
|
||||
@@ -285,7 +311,15 @@ class TestLoraEnableRoute(LoraAPITestCase):
|
||||
os.path.normcase(os.path.realpath(str(lm))),
|
||||
)
|
||||
self.assertEqual(IntegrationConfig.CIVITAI_API_KEY, SECRET_KEY)
|
||||
self.save_mock.assert_called_once_with()
|
||||
saved = self._saved()["integrations"]["lora_manager"]
|
||||
self.assertEqual(saved["civitai_api_key"], SECRET_KEY)
|
||||
self.assertEqual(saved["path"], IntegrationConfig.LORA_MANAGER_PATH)
|
||||
self.assertIs(saved["enabled"], True)
|
||||
self.assertIs(saved["trigger_words_enabled"], True)
|
||||
self.assertEqual(
|
||||
(data["persisted"], data["runtime_applied"], data["cache_loaded"]),
|
||||
(True, True, True),
|
||||
)
|
||||
self.assertTrue(get_trigger_cache().is_loaded)
|
||||
self.assertEqual(get_trigger_cache().get_trigger_words("neon"), ["glow"])
|
||||
|
||||
@@ -319,7 +353,7 @@ class TestLoraEnableRoute(LoraAPITestCase):
|
||||
self.assertFalse(json.loads(body)["success"])
|
||||
self.assertNotIn(str(self.tmp_root), body)
|
||||
self.assertFalse(IntegrationConfig.LORA_MANAGER_ENABLED)
|
||||
self.save_mock.assert_not_called()
|
||||
self.assertFalse(self.config_file.exists())
|
||||
|
||||
async def test_enable_with_custom_node_lacking_lora_in_its_name_is_400(self):
|
||||
lm = _make_lora_manager(self.custom_nodes).parent / "ComfyUI-Other"
|
||||
@@ -344,7 +378,6 @@ class TestLoraEnableRoute(LoraAPITestCase):
|
||||
|
||||
async def test_enable_writes_the_configured_config_file(self):
|
||||
config_file = self.tmp_root / "cfg" / "config.json"
|
||||
self.save_mock.side_effect = self._real_save
|
||||
|
||||
with patch.dict(os.environ, {"PROMPT_MANAGER_CONFIG_PATH": str(config_file)}):
|
||||
resp = await self._post_json(
|
||||
@@ -353,7 +386,6 @@ class TestLoraEnableRoute(LoraAPITestCase):
|
||||
)
|
||||
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.save_mock.assert_called_once_with()
|
||||
self.assertTrue(config_file.is_file())
|
||||
saved = json.loads(config_file.read_text())["integrations"]["lora_manager"]
|
||||
self.assertFalse(saved["enabled"])
|
||||
@@ -414,9 +446,299 @@ class TestLoraEnableRoute(LoraAPITestCase):
|
||||
self.assertEqual(resp.status, 400)
|
||||
|
||||
async def test_persist_failure_is_500(self):
|
||||
self.save_mock.side_effect = OSError("read-only")
|
||||
resp = await self._post_json("/prompt_manager/lora/enable", {"enabled": False})
|
||||
self._make_unwritable()
|
||||
with self.assertLogs("prompt_manager.api", level="ERROR"):
|
||||
resp = await self._post_json(
|
||||
"/prompt_manager/lora/enable", {"enabled": False}
|
||||
)
|
||||
self.assertEqual(resp.status, 500)
|
||||
self.assertEqual((await resp.json())["code"], "config-write-failed")
|
||||
|
||||
async def test_enable_write_failure_reports_and_keeps_runtime(self):
|
||||
# #220: a save that cannot be written publishes nothing and says so.
|
||||
lm = _make_lora_manager(self.custom_nodes)
|
||||
self.detect_mock.return_value = str(lm)
|
||||
IntegrationConfig.CIVITAI_API_KEY = SECRET_KEY
|
||||
self._make_unwritable()
|
||||
|
||||
with self.assertLogs("prompt_manager.api", level="ERROR"):
|
||||
resp = await self._post_json(
|
||||
"/prompt_manager/lora/enable",
|
||||
{
|
||||
"enabled": True,
|
||||
"path": str(lm),
|
||||
"trigger_words_enabled": True,
|
||||
"clear_civitai_api_key": True,
|
||||
},
|
||||
)
|
||||
|
||||
text = await resp.text()
|
||||
body = json.loads(text)
|
||||
self.assertEqual(resp.status, 500)
|
||||
self.assertEqual(body["code"], "config-write-failed")
|
||||
self.assertEqual((body["persisted"], body["runtime_applied"]), (False, False))
|
||||
self.assertFalse(IntegrationConfig.LORA_MANAGER_ENABLED)
|
||||
self.assertEqual(IntegrationConfig.LORA_MANAGER_PATH, "")
|
||||
self.assertFalse(IntegrationConfig.LORA_TRIGGER_WORDS_ENABLED)
|
||||
self.assertEqual(IntegrationConfig.CIVITAI_API_KEY, SECRET_KEY)
|
||||
self.assertFalse(get_trigger_cache().is_loaded)
|
||||
self.assertNotIn(SECRET_KEY, text)
|
||||
self.assertNotIn(str(self.tmp_root), text)
|
||||
|
||||
async def test_enable_preserves_unknown_sections(self):
|
||||
stored = "stored-value" # a plain marker: what is on disk must stay
|
||||
self.config_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"custom_section": {"keep": True},
|
||||
"web_ui": {"result_timeout": 11},
|
||||
"integrations": {
|
||||
"lora_manager": {"civitai_api_key": stored, "extra": 1},
|
||||
"other_tool": {"on": True},
|
||||
},
|
||||
}
|
||||
)
|
||||
)
|
||||
IntegrationConfig.CIVITAI_API_KEY = stored
|
||||
|
||||
resp = await self._post_json(
|
||||
"/prompt_manager/lora/enable",
|
||||
{"enabled": False, "trigger_words_enabled": True},
|
||||
)
|
||||
|
||||
self.assertEqual(resp.status, 200)
|
||||
saved = self._saved()
|
||||
self.assertEqual(saved["custom_section"], {"keep": True})
|
||||
self.assertEqual(saved["web_ui"], {"result_timeout": 11})
|
||||
self.assertEqual(saved["integrations"]["other_tool"], {"on": True})
|
||||
lora = saved["integrations"]["lora_manager"]
|
||||
self.assertEqual(lora["extra"], 1)
|
||||
self.assertEqual(lora["civitai_api_key"], stored)
|
||||
self.assertIs(lora["trigger_words_enabled"], True)
|
||||
self.assertNotIn("database", saved) # no runtime defaults written
|
||||
|
||||
async def test_enable_refuses_a_corrupt_config_file(self):
|
||||
self.config_file.write_bytes(b"{not json")
|
||||
resp = await self._post_json(
|
||||
"/prompt_manager/lora/enable",
|
||||
{"enabled": False, "trigger_words_enabled": True},
|
||||
)
|
||||
body = await resp.json()
|
||||
self.assertEqual((resp.status, body["code"]), (409, "config-unreadable"))
|
||||
self.assertEqual(self.config_file.read_bytes(), b"{not json")
|
||||
self.assertFalse(IntegrationConfig.LORA_TRIGGER_WORDS_ENABLED)
|
||||
|
||||
async def test_enable_requires_boolean_flags_and_body_cap(self):
|
||||
for body in (
|
||||
{"enabled": "yes"},
|
||||
{"enabled": False, "trigger_words_enabled": 1},
|
||||
{"enabled": False, "clear_civitai_api_key": "true"},
|
||||
):
|
||||
with self.subTest(body=body):
|
||||
resp = await self._post_json("/prompt_manager/lora/enable", body)
|
||||
self.assertEqual(resp.status, 400)
|
||||
resp = await self._post_json("/prompt_manager/lora/enable", [True])
|
||||
self.assertEqual(resp.status, 400)
|
||||
resp = await self._post_json(
|
||||
"/prompt_manager/lora/enable",
|
||||
{"enabled": False, "civitai_api_key": "k" * (2 * 1024 * 1024)},
|
||||
)
|
||||
self.assertEqual(resp.status, 413)
|
||||
self.assertFalse(self.config_file.exists())
|
||||
self.assertFalse(IntegrationConfig.LORA_MANAGER_ENABLED)
|
||||
|
||||
async def test_blank_key_unchanged_and_explicit_clear(self):
|
||||
self.config_file.write_text(
|
||||
json.dumps({"integrations": {"lora_manager": {"civitai_api_key": "old"}}})
|
||||
)
|
||||
IntegrationConfig.CIVITAI_API_KEY = "old"
|
||||
|
||||
resp = await self._post_json(
|
||||
"/prompt_manager/lora/enable", {"enabled": False, "civitai_api_key": " "}
|
||||
)
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertEqual(
|
||||
self._saved()["integrations"]["lora_manager"]["civitai_api_key"], "old"
|
||||
)
|
||||
self.assertEqual(IntegrationConfig.CIVITAI_API_KEY, "old")
|
||||
|
||||
resp = await self._post_json(
|
||||
"/prompt_manager/lora/enable",
|
||||
{"enabled": False, "clear_civitai_api_key": True},
|
||||
)
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertEqual(
|
||||
self._saved()["integrations"]["lora_manager"]["civitai_api_key"], ""
|
||||
)
|
||||
self.assertEqual(IntegrationConfig.CIVITAI_API_KEY, "")
|
||||
|
||||
async def test_cache_load_failure_reports_persisted_but_unloaded(self):
|
||||
lm = _make_lora_manager(self.custom_nodes)
|
||||
self.detect_mock.return_value = str(lm)
|
||||
|
||||
with (
|
||||
patch.object(
|
||||
type(get_trigger_cache()), "load", side_effect=OSError("unreadable")
|
||||
),
|
||||
self.assertLogs("prompt_manager.api", level="WARNING"),
|
||||
):
|
||||
resp = await self._post_json(
|
||||
"/prompt_manager/lora/enable",
|
||||
{"enabled": True, "path": str(lm), "trigger_words_enabled": True},
|
||||
)
|
||||
|
||||
text = await resp.text()
|
||||
body = json.loads(text)
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertIs(body["success"], True)
|
||||
self.assertEqual((body["persisted"], body["cache_loaded"]), (True, False))
|
||||
self.assertTrue(IntegrationConfig.LORA_MANAGER_ENABLED)
|
||||
self.assertIs(self._saved()["integrations"]["lora_manager"]["enabled"], True)
|
||||
self.assertNotIn("unreadable", text)
|
||||
|
||||
|
||||
class TestConfigWritersShareOneStore(LoraAPITestCase):
|
||||
|
||||
async def asyncSetUp(self):
|
||||
await super().asyncSetUp()
|
||||
self._orig_timeout = PromptManagerConfig.RESULT_TIMEOUT
|
||||
|
||||
async def asyncTearDown(self):
|
||||
PromptManagerConfig.RESULT_TIMEOUT = self._orig_timeout
|
||||
await super().asyncTearDown()
|
||||
|
||||
async def test_settings_and_lora_saves_do_not_overwrite_each_other(self):
|
||||
# #220: the LoRA save is sent while the settings save is inside its
|
||||
# commit (lock held, file already read). The LoRA commit re-reads the
|
||||
# file under the same lock, so both edits survive and nothing else
|
||||
# is lost (before, the LoRA save rewrote the whole runtime config).
|
||||
self.config_file.write_text(json.dumps({"custom_section": {"keep": True}}))
|
||||
inside = threading.Event()
|
||||
release = threading.Event()
|
||||
real_mutation = settings_module._settings_mutation
|
||||
|
||||
def gated_mutation(updates, new_roots):
|
||||
mutate = real_mutation(updates, new_roots)
|
||||
|
||||
def gated(config):
|
||||
inside.set()
|
||||
release.wait(10)
|
||||
return mutate(config)
|
||||
|
||||
return gated
|
||||
|
||||
lora_calls = []
|
||||
real_lora_mutation = lora_module._lora_settings_mutation
|
||||
|
||||
def recorded_lora_mutation(values):
|
||||
mutate = real_lora_mutation(values)
|
||||
|
||||
def recorded(config):
|
||||
lora_calls.append(dict(config))
|
||||
return mutate(config)
|
||||
|
||||
return recorded
|
||||
|
||||
probe = ProbeLock()
|
||||
loop = asyncio.get_running_loop()
|
||||
with (
|
||||
patch.object(settings_module, "_settings_mutation", gated_mutation),
|
||||
patch.object(
|
||||
lora_module, "_lora_settings_mutation", recorded_lora_mutation
|
||||
),
|
||||
patch.object(config_store, "_COMMIT_LOCK", probe),
|
||||
):
|
||||
settings = asyncio.ensure_future(
|
||||
self._post_json("/prompt_manager/settings", {"result_timeout": 13})
|
||||
)
|
||||
self.assertTrue(await loop.run_in_executor(None, inside.wait, 10))
|
||||
lora = asyncio.ensure_future(
|
||||
self._post_json(
|
||||
"/prompt_manager/lora/enable",
|
||||
{"enabled": False, "trigger_words_enabled": True},
|
||||
)
|
||||
)
|
||||
# Wait until the LoRA commit is blocked on the lock the settings
|
||||
# commit holds; its mutation must not have run yet.
|
||||
self.assertTrue(await loop.run_in_executor(None, probe.contended.wait, 10))
|
||||
self.assertEqual(lora_calls, [])
|
||||
release.set()
|
||||
settings_resp, lora_resp = await asyncio.gather(settings, lora)
|
||||
self.assertEqual(len(lora_calls), 1)
|
||||
self.assertEqual(lora_calls[0]["web_ui"]["result_timeout"], 13)
|
||||
|
||||
self.assertEqual((settings_resp.status, lora_resp.status), (200, 200))
|
||||
saved = self._saved()
|
||||
self.assertEqual(saved["custom_section"], {"keep": True})
|
||||
self.assertEqual(saved["web_ui"]["result_timeout"], 13)
|
||||
self.assertIs(
|
||||
saved["integrations"]["lora_manager"]["trigger_words_enabled"], True
|
||||
)
|
||||
settings_body = await settings_resp.json()
|
||||
lora_body = await lora_resp.json()
|
||||
self.assertNotEqual(settings_body["revision"], lora_body["revision"])
|
||||
|
||||
def _delay_first_commit_result(self):
|
||||
"""Hold back the first config commit's result from its handler.
|
||||
|
||||
The commit itself has run (the file is written); only the handler's
|
||||
continuation waits until the returned event is set. Before #220's fix
|
||||
round the handler published after this point, so a later save could
|
||||
publish first and the earlier one overwrite it at runtime.
|
||||
"""
|
||||
original = self.api._run_in_executor
|
||||
written = asyncio.Event()
|
||||
resume = asyncio.Event()
|
||||
delayed = []
|
||||
|
||||
async def run(func, *args, **kwargs):
|
||||
result = await original(func, *args, **kwargs)
|
||||
if getattr(func, "__func__", None) is config_store.ConfigStore.commit:
|
||||
if not delayed:
|
||||
delayed.append(func)
|
||||
written.set()
|
||||
await resume.wait()
|
||||
return result
|
||||
|
||||
patcher = patch.object(self.api, "_run_in_executor", run)
|
||||
patcher.start()
|
||||
self.addCleanup(patcher.stop)
|
||||
return written, resume
|
||||
|
||||
async def test_concurrent_settings_saves_end_on_the_last_written(self):
|
||||
written, resume = self._delay_first_commit_result()
|
||||
first = asyncio.ensure_future(
|
||||
self._post_json("/prompt_manager/settings", {"result_timeout": 11})
|
||||
)
|
||||
await asyncio.wait_for(written.wait(), 10)
|
||||
second = await self._post_json(
|
||||
"/prompt_manager/settings", {"result_timeout": 22}
|
||||
)
|
||||
resume.set()
|
||||
first = await first
|
||||
self.assertEqual((first.status, second.status), (200, 200))
|
||||
self.assertEqual(self._saved()["web_ui"]["result_timeout"], 22)
|
||||
self.assertEqual(PromptManagerConfig.RESULT_TIMEOUT, 22)
|
||||
|
||||
async def test_concurrent_lora_saves_end_on_the_last_written(self):
|
||||
written, resume = self._delay_first_commit_result()
|
||||
first = asyncio.ensure_future(
|
||||
self._post_json(
|
||||
"/prompt_manager/lora/enable",
|
||||
{"enabled": False, "trigger_words_enabled": True},
|
||||
)
|
||||
)
|
||||
await asyncio.wait_for(written.wait(), 10)
|
||||
second = await self._post_json(
|
||||
"/prompt_manager/lora/enable",
|
||||
{"enabled": False, "trigger_words_enabled": False},
|
||||
)
|
||||
resume.set()
|
||||
first = await first
|
||||
self.assertEqual((first.status, second.status), (200, 200))
|
||||
lora = self._saved()["integrations"]["lora_manager"]
|
||||
self.assertIs(lora["trigger_words_enabled"], False)
|
||||
self.assertIs(IntegrationConfig.LORA_TRIGGER_WORDS_ENABLED, False)
|
||||
|
||||
|
||||
class TestLoraScanRoute(LoraAPITestCase):
|
||||
@@ -445,6 +767,12 @@ class TestLoraScanRoute(LoraAPITestCase):
|
||||
body = await resp.text()
|
||||
self.assertNotIn(str(self.tmp_root), body)
|
||||
self.assertIn("x.db", body)
|
||||
# The admin page's modal shows the failure as its status and closes.
|
||||
error = self._sse_events(body)[-1]
|
||||
self.assertEqual(
|
||||
(error["type"], error["status"], error["progress"]),
|
||||
("error", error["message"], 0),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _sse_events(text):
|
||||
|
||||
@@ -0,0 +1,419 @@
|
||||
"""Tests for the serialized config store (py/config_store.py, #220).
|
||||
|
||||
Every test works on a config.json inside its own temporary directory; the
|
||||
repository's config.json (live user data) is never read or written.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
_mock_server = MagicMock()
|
||||
_mock_server.PromptServer.instance.routes = MagicMock()
|
||||
sys.modules.setdefault("server", _mock_server)
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from py import config_store # noqa: E402
|
||||
from py.config_store import ( # noqa: E402
|
||||
ABSENT_REVISION,
|
||||
ConfigStore,
|
||||
ConfigUnreadable,
|
||||
RevisionConflict,
|
||||
get_config_store,
|
||||
revision_of,
|
||||
)
|
||||
|
||||
try:
|
||||
from tests.lock_probe import ProbeLock # noqa: E402
|
||||
from tests.open_handles import assert_closed # noqa: E402
|
||||
except ImportError: # discovered with tests/ as the top-level directory
|
||||
from lock_probe import ProbeLock # noqa: E402
|
||||
from open_handles import assert_closed # noqa: E402
|
||||
|
||||
|
||||
class ConfigStoreTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmpdir = os.path.realpath(tempfile.mkdtemp())
|
||||
self.path = os.path.join(self.tmpdir, "config.json")
|
||||
self.store = ConfigStore(lambda: self.path)
|
||||
|
||||
def tearDown(self):
|
||||
try:
|
||||
assert_closed(self, self.tmpdir)
|
||||
finally:
|
||||
shutil.rmtree(self.tmpdir, ignore_errors=True)
|
||||
|
||||
def _write(self, data):
|
||||
Path(self.path).write_text(json.dumps(data), encoding="utf-8")
|
||||
|
||||
def _bytes(self):
|
||||
return Path(self.path).read_bytes()
|
||||
|
||||
def _leftovers(self):
|
||||
return sorted(name for name in os.listdir(self.tmpdir) if name != "config.json")
|
||||
|
||||
# ── read ──────────────────────────────────────────────────────────
|
||||
|
||||
def test_read_of_a_missing_file_is_empty_and_absent(self):
|
||||
self.assertEqual(self.store.read(), ({}, ABSENT_REVISION))
|
||||
|
||||
def test_read_reports_the_sha256_of_the_file_bytes(self):
|
||||
self._write({"web_ui": {"result_timeout": 3}})
|
||||
data, revision = self.store.read()
|
||||
self.assertEqual(data, {"web_ui": {"result_timeout": 3}})
|
||||
expected = "sha256:" + hashlib.sha256(self._bytes()).hexdigest()
|
||||
self.assertEqual(revision, expected)
|
||||
self.assertEqual(revision_of(self._bytes()), expected)
|
||||
|
||||
def test_non_object_config_is_unreadable(self):
|
||||
self._write([1, 2])
|
||||
with self.assertRaises(ConfigUnreadable):
|
||||
self.store.read()
|
||||
|
||||
def test_unreadable_path_is_unreadable(self):
|
||||
os.mkdir(self.path) # a directory where the file should be
|
||||
with self.assertRaises(ConfigUnreadable):
|
||||
self.store.read()
|
||||
|
||||
# ── commit ────────────────────────────────────────────────────────
|
||||
|
||||
def test_commit_preserves_unknown_sections_and_secret(self):
|
||||
self._write(
|
||||
{
|
||||
"custom_section": {"keep": True},
|
||||
"integrations": {"lora_manager": {"civitai_api_key": "secret"}},
|
||||
"web_ui": {"result_timeout": 1, "show_test_button": True},
|
||||
}
|
||||
)
|
||||
|
||||
def mutate(config):
|
||||
config.setdefault("web_ui", {})["result_timeout"] = 7
|
||||
return config
|
||||
|
||||
saved, revision = self.store.commit(mutate, expected_revision=None)
|
||||
|
||||
on_disk = json.loads(self._bytes())
|
||||
self.assertEqual(saved, on_disk)
|
||||
self.assertEqual(on_disk["custom_section"], {"keep": True})
|
||||
self.assertEqual(
|
||||
on_disk["integrations"]["lora_manager"]["civitai_api_key"], "secret"
|
||||
)
|
||||
self.assertEqual(
|
||||
on_disk["web_ui"], {"result_timeout": 7, "show_test_button": True}
|
||||
)
|
||||
self.assertEqual(revision, revision_of(self._bytes()))
|
||||
self.assertEqual(self.store.read(), (on_disk, revision))
|
||||
|
||||
def test_commit_creates_a_missing_file_and_its_parent(self):
|
||||
self.path = os.path.join(self.tmpdir, "nested", "config.json")
|
||||
saved, revision = self.store.commit(
|
||||
lambda config: {**config, "a": 1}, expected_revision=ABSENT_REVISION
|
||||
)
|
||||
self.assertEqual(saved, {"a": 1})
|
||||
self.assertEqual(json.loads(Path(self.path).read_text()), {"a": 1})
|
||||
self.assertNotEqual(revision, ABSENT_REVISION)
|
||||
|
||||
def test_mutate_works_on_a_copy(self):
|
||||
self._write({"web_ui": {"result_timeout": 1}})
|
||||
seen = {}
|
||||
|
||||
def mutate(config):
|
||||
seen["config"] = config
|
||||
config["web_ui"]["result_timeout"] = 2
|
||||
raise RuntimeError("validation failed late")
|
||||
|
||||
before = self._bytes()
|
||||
with self.assertRaises(RuntimeError):
|
||||
self.store.commit(mutate, expected_revision=None)
|
||||
self.assertEqual(self._bytes(), before)
|
||||
self.assertEqual(self.store.read()[0], {"web_ui": {"result_timeout": 1}})
|
||||
|
||||
def test_mutate_must_return_an_object(self):
|
||||
self._write({"a": 1})
|
||||
before = self._bytes()
|
||||
with self.assertRaises(TypeError):
|
||||
self.store.commit(lambda config: None, expected_revision=None)
|
||||
self.assertEqual(self._bytes(), before)
|
||||
|
||||
def test_malformed_existing_config_is_refused_and_untouched(self):
|
||||
Path(self.path).write_bytes(b"{not json")
|
||||
calls = []
|
||||
with self.assertRaises(ConfigUnreadable):
|
||||
self.store.commit(
|
||||
lambda config: calls.append(config) or config, expected_revision=None
|
||||
)
|
||||
self.assertEqual(calls, [])
|
||||
self.assertEqual(self._bytes(), b"{not json")
|
||||
self.assertEqual(self._leftovers(), [])
|
||||
|
||||
def test_write_failure_leaves_bytes_and_raises(self):
|
||||
self._write({"a": 1})
|
||||
before = self._bytes()
|
||||
with patch.object(config_store.os, "fsync", side_effect=OSError("disk full")):
|
||||
with self.assertRaises(OSError):
|
||||
self.store.commit(lambda c: {**c, "b": 2}, expected_revision=None)
|
||||
self.assertEqual(self._bytes(), before)
|
||||
self.assertEqual(self._leftovers(), [])
|
||||
|
||||
def test_unwritable_directory_raises_oserror(self):
|
||||
blocker = os.path.join(self.tmpdir, "afile")
|
||||
Path(blocker).write_text("x")
|
||||
self.path = os.path.join(blocker, "config.json") # parent is a file
|
||||
with self.assertRaises(OSError):
|
||||
self.store.commit(lambda c: {"a": 1}, expected_revision=None)
|
||||
|
||||
def test_stale_revision_conflicts_without_write(self):
|
||||
self._write({"a": 1})
|
||||
_data, revision = self.store.read()
|
||||
self._write({"a": 2}) # another writer got there first
|
||||
before = self._bytes()
|
||||
calls = []
|
||||
with self.assertRaises(RevisionConflict) as caught:
|
||||
self.store.commit(
|
||||
lambda c: calls.append(c) or c, expected_revision=revision
|
||||
)
|
||||
self.assertEqual(caught.exception.current, revision_of(before))
|
||||
self.assertEqual(calls, [])
|
||||
self.assertEqual(self._bytes(), before)
|
||||
|
||||
def test_matching_revision_commits(self):
|
||||
self._write({"a": 1})
|
||||
_data, revision = self.store.read()
|
||||
saved, _ = self.store.commit(
|
||||
lambda c: {**c, "b": 2}, expected_revision=revision
|
||||
)
|
||||
self.assertEqual(saved, {"a": 1, "b": 2})
|
||||
|
||||
def test_temp_file_removed_after_replace_failure(self):
|
||||
self._write({"a": 1})
|
||||
before = self._bytes()
|
||||
with patch.object(
|
||||
config_store.os, "replace", side_effect=PermissionError("locked")
|
||||
):
|
||||
with self.assertRaises(OSError):
|
||||
self.store.commit(lambda c: {**c, "b": 2}, expected_revision=None)
|
||||
self.assertEqual(self._bytes(), before)
|
||||
self.assertEqual(self._leftovers(), [])
|
||||
|
||||
@unittest.skipUnless(os.name == "posix", "POSIX file modes")
|
||||
def test_committed_file_is_private_to_the_user(self):
|
||||
import stat
|
||||
|
||||
self._write({"a": 1})
|
||||
os.chmod(self.path, 0o644)
|
||||
self.store.commit(lambda c: c, expected_revision=None)
|
||||
self.assertEqual(stat.S_IMODE(os.stat(self.path).st_mode), 0o600)
|
||||
|
||||
def _probe_lock(self):
|
||||
probe = ProbeLock()
|
||||
patcher = patch.object(config_store, "_COMMIT_LOCK", probe)
|
||||
patcher.start()
|
||||
self.addCleanup(patcher.stop)
|
||||
return probe
|
||||
|
||||
def _hold_lock_in_a_commit(self, value="first"):
|
||||
"""Start a commit that holds the lock until the returned event is set."""
|
||||
inside = threading.Event()
|
||||
release = threading.Event()
|
||||
errors = []
|
||||
|
||||
def slow(config):
|
||||
inside.set()
|
||||
release.wait(10)
|
||||
config[value] = 1
|
||||
return config
|
||||
|
||||
def run():
|
||||
try:
|
||||
self.store.commit(slow, expected_revision=None)
|
||||
except Exception as exc: # pragma: no cover - reported by the test
|
||||
errors.append(exc)
|
||||
|
||||
thread = threading.Thread(target=run)
|
||||
thread.start()
|
||||
self.assertTrue(inside.wait(10))
|
||||
return thread, release, errors
|
||||
|
||||
def test_concurrent_commits_serialize(self):
|
||||
# The second commit is observed blocked on the lock before the first
|
||||
# one may finish, so this fails whenever commit stops taking the lock.
|
||||
self._write({"keep": True})
|
||||
probe = self._probe_lock()
|
||||
first, release, errors = self._hold_lock_in_a_commit()
|
||||
calls = []
|
||||
|
||||
def fast(config):
|
||||
calls.append(dict(config))
|
||||
config["second"] = 2
|
||||
return config
|
||||
|
||||
second = threading.Thread(
|
||||
target=lambda: self.store.commit(fast, expected_revision=None)
|
||||
)
|
||||
second.start()
|
||||
self.assertTrue(probe.contended.wait(10), "second commit never waited")
|
||||
self.assertEqual(calls, []) # not called while the first holds the lock
|
||||
release.set()
|
||||
first.join(10)
|
||||
second.join(10)
|
||||
|
||||
self.assertEqual(errors, [])
|
||||
self.assertEqual(calls, [{"keep": True, "first": 1}]) # re-read inside
|
||||
self.assertEqual(
|
||||
json.loads(self._bytes()), {"keep": True, "first": 1, "second": 2}
|
||||
)
|
||||
|
||||
def test_read_waits_for_a_running_commit(self):
|
||||
# Windows cannot replace a file another handle has open, so a read
|
||||
# runs under the commit lock and never overlaps a replace.
|
||||
self._write({"keep": True})
|
||||
probe = self._probe_lock()
|
||||
first, release, errors = self._hold_lock_in_a_commit()
|
||||
seen = []
|
||||
reader = threading.Thread(target=lambda: seen.append(self.store.read()))
|
||||
reader.start()
|
||||
self.assertTrue(probe.contended.wait(10), "read did not take the lock")
|
||||
self.assertEqual(seen, [])
|
||||
release.set()
|
||||
first.join(10)
|
||||
reader.join(10)
|
||||
self.assertEqual(errors, [])
|
||||
self.assertEqual(seen[0][0], {"keep": True, "first": 1})
|
||||
self.assertEqual(seen[0][1], revision_of(self._bytes()))
|
||||
|
||||
def test_replace_is_retried_after_a_transient_permission_error(self):
|
||||
# Windows: an antivirus or indexer handle makes os.replace fail with
|
||||
# PermissionError for a moment; the commit retries before failing.
|
||||
self._write({"a": 1})
|
||||
real_replace = os.replace
|
||||
attempts = []
|
||||
|
||||
def flaky(src, dst):
|
||||
attempts.append(dst)
|
||||
if len(attempts) == 1:
|
||||
raise PermissionError("in use by another process")
|
||||
return real_replace(src, dst)
|
||||
|
||||
with (
|
||||
patch.object(config_store, "_REPLACE_ATTEMPTS", 5),
|
||||
patch.object(config_store.os, "replace", side_effect=flaky),
|
||||
patch.object(config_store.time, "sleep") as sleep,
|
||||
):
|
||||
saved, _ = self.store.commit(
|
||||
lambda c: {**c, "b": 2}, expected_revision=None
|
||||
)
|
||||
self.assertEqual(saved, {"a": 1, "b": 2})
|
||||
self.assertEqual(json.loads(self._bytes()), {"a": 1, "b": 2})
|
||||
self.assertEqual(len(attempts), 2)
|
||||
sleep.assert_called_once_with(config_store._REPLACE_BACKOFF_SECONDS)
|
||||
self.assertEqual(self._leftovers(), [])
|
||||
|
||||
def test_replace_gives_up_after_the_last_attempt(self):
|
||||
self._write({"a": 1})
|
||||
before = self._bytes()
|
||||
with (
|
||||
patch.object(config_store, "_REPLACE_ATTEMPTS", 3),
|
||||
patch.object(
|
||||
config_store.os, "replace", side_effect=PermissionError("locked")
|
||||
) as replace,
|
||||
patch.object(config_store.time, "sleep") as sleep,
|
||||
):
|
||||
with self.assertRaises(PermissionError):
|
||||
self.store.commit(lambda c: {**c, "b": 2}, expected_revision=None)
|
||||
self.assertEqual((replace.call_count, sleep.call_count), (3, 2))
|
||||
self.assertEqual(self._bytes(), before)
|
||||
self.assertEqual(self._leftovers(), [])
|
||||
|
||||
def test_replace_is_tried_once_outside_windows(self):
|
||||
expected = 5 if os.name == "nt" else 1
|
||||
self.assertEqual(config_store._REPLACE_ATTEMPTS, expected)
|
||||
|
||||
@unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
|
||||
def test_symlinked_config_is_written_through_the_link(self):
|
||||
# A config.json that is a symlink (or a single-file bind mount) is
|
||||
# written beside its real target, so the link stays a link.
|
||||
real_dir = os.path.join(self.tmpdir, "real")
|
||||
os.mkdir(real_dir)
|
||||
target = os.path.join(real_dir, "settings.json")
|
||||
Path(target).write_text(json.dumps({"a": 1}), encoding="utf-8")
|
||||
try:
|
||||
os.symlink(target, self.path)
|
||||
except (OSError, NotImplementedError): # pragma: no cover - Windows
|
||||
self.skipTest("symlinks are not permitted here")
|
||||
|
||||
saved, revision = self.store.commit(
|
||||
lambda c: {**c, "b": 2}, expected_revision=None
|
||||
)
|
||||
|
||||
self.assertTrue(os.path.islink(self.path))
|
||||
self.assertEqual(json.loads(Path(target).read_text()), {"a": 1, "b": 2})
|
||||
self.assertEqual(self.store.read(), (saved, revision))
|
||||
self.assertEqual(os.listdir(real_dir), ["settings.json"])
|
||||
|
||||
def test_publish_runs_under_the_lock_after_the_write(self):
|
||||
# Runtimes publish inside the commit, so two saves publish in the
|
||||
# order they were written and the runtime ends on the last one.
|
||||
self._write({"a": 1})
|
||||
seen = []
|
||||
|
||||
# The lock is re-entrant, so another thread probes whether it is held.
|
||||
def probe_publish(config):
|
||||
result = []
|
||||
thread = threading.Thread(
|
||||
target=lambda: result.append(
|
||||
config_store._COMMIT_LOCK.acquire(blocking=False)
|
||||
)
|
||||
)
|
||||
thread.start()
|
||||
thread.join(10)
|
||||
seen.append((config, json.loads(self._bytes()), result[0]))
|
||||
|
||||
saved, _ = self.store.commit(
|
||||
lambda c: {**c, "b": 2}, expected_revision=None, publish=probe_publish
|
||||
)
|
||||
self.assertEqual(seen, [(saved, {"a": 1, "b": 2}, False)])
|
||||
|
||||
def test_publish_is_skipped_when_the_write_fails(self):
|
||||
self._write({"a": 1})
|
||||
published = []
|
||||
with patch.object(config_store.os, "fsync", side_effect=OSError("disk full")):
|
||||
with self.assertRaises(OSError):
|
||||
self.store.commit(
|
||||
lambda c: {**c, "b": 2},
|
||||
expected_revision=None,
|
||||
publish=published.append,
|
||||
)
|
||||
self.assertEqual(published, [])
|
||||
|
||||
def test_handles_closed_after_commit(self):
|
||||
self._write({"a": 1})
|
||||
self.store.read()
|
||||
self.store.commit(lambda c: {**c, "b": 2}, expected_revision=None)
|
||||
assert_closed(self, self.tmpdir)
|
||||
os.unlink(self.path) # Windows refuses this while a handle is open
|
||||
self.assertEqual(self._leftovers(), [])
|
||||
|
||||
|
||||
class SingletonTests(unittest.TestCase):
|
||||
def test_get_config_store_is_one_process_store_on_the_configured_path(self):
|
||||
store = get_config_store()
|
||||
self.assertIs(store, get_config_store())
|
||||
tmpdir = tempfile.mkdtemp()
|
||||
try:
|
||||
custom = os.path.join(tmpdir, "c.json")
|
||||
with patch.dict(os.environ, {"PROMPT_MANAGER_CONFIG_PATH": custom}):
|
||||
self.assertEqual(store.path, custom)
|
||||
finally:
|
||||
shutil.rmtree(tmpdir, ignore_errors=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -420,6 +420,23 @@ class TestStartAndExclusion(ImageTagJobTestCase):
|
||||
# A refused start never leaves the gate taken.
|
||||
self.assertIsNone(model_gate.holder())
|
||||
|
||||
async def test_image_tag_job_refused_while_legacy_restore_holds(self):
|
||||
self.png("a.png")
|
||||
async with self.api._jobs.hold(
|
||||
"restore", "Database restore (admin page)", admission="records"
|
||||
):
|
||||
status, body = await self.start(self.add_body(["a.png"], ["x"]))
|
||||
self.assertEqual((status, body["code"]), (409, "busy"))
|
||||
self.assertEqual(body["running"]["title"], "Database restore (admin page)")
|
||||
status, body = await self.start(self.auto_body(["a.png"]))
|
||||
self.assertEqual((status, body["code"]), (409, "busy"))
|
||||
self.assertIsNone(model_gate.holder())
|
||||
async with self.api._jobs.hold(
|
||||
"thumbnails", "Thumbnail generation (gallery page)", admission="guarded"
|
||||
):
|
||||
status, body = await self.start(self.add_body(["a.png"], ["x"]))
|
||||
self.assertEqual((status, body["code"]), (409, "busy"))
|
||||
|
||||
async def test_record_changing_excludes_running_scan(self):
|
||||
self.png("a.png")
|
||||
gate = asyncio.Event()
|
||||
|
||||
@@ -0,0 +1,552 @@
|
||||
"""Registry admission classes and the legacy routes that hold the registry.
|
||||
|
||||
Record-changing work never runs together (Activity design): a ``records``
|
||||
job is refused while any ``records`` or ``guarded`` job runs, a ``guarded``
|
||||
job (thumbnails) while a ``records`` job runs, and ``read`` jobs only share
|
||||
the one-per-kind rule. The legacy admin and gallery routes that change
|
||||
records or files take a registry hold for their duration, so a workspace job
|
||||
and a legacy-page operation refuse each other, naming the one that runs.
|
||||
|
||||
Fake runners and gated executor calls only; ordering is driven by events.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
_mock_server = MagicMock()
|
||||
_mock_server.PromptServer.instance.routes = MagicMock()
|
||||
sys.modules.setdefault("server", _mock_server)
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from py.api.jobs import ( # noqa: E402
|
||||
ADMISSIONS,
|
||||
Job,
|
||||
JobBusy,
|
||||
JobRegistry,
|
||||
)
|
||||
|
||||
try:
|
||||
from tests.test_api_admin import AdminAPITestCase # noqa: E402
|
||||
except ImportError: # discovered with tests/ as the top-level directory
|
||||
from test_api_admin import AdminAPITestCase # noqa: E402
|
||||
|
||||
|
||||
def _gated(gate):
|
||||
async def runner(job):
|
||||
await gate.wait()
|
||||
|
||||
return runner
|
||||
|
||||
|
||||
async def _noop(job):
|
||||
return None
|
||||
|
||||
|
||||
class TestAdmissionClasses(unittest.IsolatedAsyncioTestCase):
|
||||
|
||||
def _start(self, registry, kind, admission, gate=None, **kwargs):
|
||||
return registry.start(
|
||||
kind,
|
||||
kind.title(),
|
||||
admission=admission,
|
||||
scope={},
|
||||
runner=_gated(gate) if gate is not None else _noop,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
async def test_admission_names_and_explicit_read(self):
|
||||
self.assertEqual(ADMISSIONS, ("read", "guarded", "records"))
|
||||
registry = JobRegistry()
|
||||
job = registry.start(
|
||||
"report", "Report", admission="read", scope={}, runner=_noop
|
||||
)
|
||||
await asyncio.wait_for(job.task, 2)
|
||||
self.assertEqual(job.admission, "read")
|
||||
self.assertFalse(job.record_changing)
|
||||
with self.assertRaises(ValueError):
|
||||
registry.start("x", "X", admission="exclusive", scope={}, runner=_noop)
|
||||
with self.assertRaises(ValueError):
|
||||
registry.start(
|
||||
"x",
|
||||
"X",
|
||||
admission="read",
|
||||
record_changing=True,
|
||||
scope={},
|
||||
runner=_noop,
|
||||
)
|
||||
|
||||
async def test_start_requires_an_admission_class(self):
|
||||
# A job that forgets its class must not silently run as "read" beside
|
||||
# a scan or restore: every start call decides.
|
||||
registry = JobRegistry()
|
||||
with self.assertRaises(TypeError) as raised:
|
||||
registry.start("review", "Review", scope={}, runner=_noop)
|
||||
self.assertIn("admission", str(raised.exception))
|
||||
self.assertEqual(registry.list(), [])
|
||||
job = registry.start(
|
||||
"review", "Review", record_changing=False, scope={}, runner=_noop
|
||||
)
|
||||
await asyncio.wait_for(job.task, 2)
|
||||
self.assertEqual(job.admission, "read")
|
||||
|
||||
async def test_record_changing_is_an_alias_for_records(self):
|
||||
registry = JobRegistry()
|
||||
gate = asyncio.Event()
|
||||
job = registry.start(
|
||||
"scan", "Scan", record_changing=True, scope={}, runner=_gated(gate)
|
||||
)
|
||||
self.assertEqual(job.admission, "records")
|
||||
self.assertTrue(job.record_changing)
|
||||
self.assertEqual(job.summary()["admission"], "records")
|
||||
self.assertIs(
|
||||
registry.blocker("image-tags", record_changing=True), job
|
||||
) # the existing call shape keeps working
|
||||
gate.set()
|
||||
await asyncio.wait_for(job.task, 2)
|
||||
|
||||
async def test_records_job_refused_while_guarded_runs(self):
|
||||
registry = JobRegistry()
|
||||
gate = asyncio.Event()
|
||||
thumbs = self._start(registry, "thumbnails", "guarded", gate)
|
||||
with self.assertRaises(JobBusy) as raised:
|
||||
self._start(registry, "scan", "records")
|
||||
self.assertIs(raised.exception.running, thumbs)
|
||||
self.assertIs(registry.blocker("restore", admission="records"), thumbs)
|
||||
gate.set()
|
||||
await asyncio.wait_for(thumbs.task, 2)
|
||||
scan = self._start(registry, "scan", "records")
|
||||
await asyncio.wait_for(scan.task, 2)
|
||||
self.assertEqual(scan.state, "finished")
|
||||
|
||||
async def test_guarded_refused_while_records_runs(self):
|
||||
registry = JobRegistry()
|
||||
gate = asyncio.Event()
|
||||
scan = self._start(registry, "scan", "records", gate)
|
||||
with self.assertRaises(JobBusy) as raised:
|
||||
self._start(registry, "thumbnails", "guarded")
|
||||
self.assertIs(raised.exception.running, scan)
|
||||
# A second guarded kind may run beside a guarded one, not beside records.
|
||||
gate.set()
|
||||
await asyncio.wait_for(scan.task, 2)
|
||||
hold = asyncio.Event()
|
||||
thumbs = self._start(registry, "thumbnails", "guarded", hold)
|
||||
backup = self._start(registry, "backup", "guarded")
|
||||
await asyncio.wait_for(backup.task, 2)
|
||||
self.assertEqual(backup.state, "finished")
|
||||
hold.set()
|
||||
await asyncio.wait_for(thumbs.task, 2)
|
||||
|
||||
async def test_read_jobs_run_in_parallel(self):
|
||||
registry = JobRegistry()
|
||||
gate = asyncio.Event()
|
||||
scan = self._start(registry, "scan", "records", gate)
|
||||
health = self._start(registry, "health", "read", gate)
|
||||
duplicates = self._start(registry, "duplicates", "read", gate)
|
||||
self.assertTrue(all(j.running for j in (scan, health, duplicates)))
|
||||
with self.assertRaises(JobBusy) as raised: # one job of a kind
|
||||
self._start(registry, "health", "read")
|
||||
self.assertIs(raised.exception.running, health)
|
||||
gate.set()
|
||||
await asyncio.wait_for(
|
||||
asyncio.gather(scan.task, health.task, duplicates.task), 2
|
||||
)
|
||||
|
||||
async def test_non_cancellable_job_ignores_cancel(self):
|
||||
registry = JobRegistry()
|
||||
gate = asyncio.Event()
|
||||
job = self._start(registry, "restore", "records", gate, cancellable=False)
|
||||
self.assertIs(job.summary()["cancellable"], False)
|
||||
self.assertFalse(registry.cancel(job.id))
|
||||
self.assertFalse(job.cancelled)
|
||||
gate.set()
|
||||
await asyncio.wait_for(job.task, 2)
|
||||
self.assertEqual(job.state, "finished")
|
||||
|
||||
async def test_hold_registers_a_running_non_cancellable_job(self):
|
||||
registry = JobRegistry()
|
||||
held = registry.hold(
|
||||
"restore", "Database restore (admin page)", admission="records"
|
||||
)
|
||||
async with held as job:
|
||||
self.assertIs(registry.active("restore"), job)
|
||||
self.assertEqual(job.state, "running")
|
||||
self.assertEqual(
|
||||
(job.summary()["cancellable"], job.summary()["admission"]),
|
||||
(False, "records"),
|
||||
)
|
||||
self.assertFalse(registry.cancel(job.id))
|
||||
with self.assertRaises(JobBusy) as raised:
|
||||
self._start(registry, "scan", "records")
|
||||
self.assertIn("Database restore (admin page)", str(raised.exception))
|
||||
self.assertEqual(job.state, "finished")
|
||||
self.assertIsNone(registry.active("restore"))
|
||||
self.assertIs(registry.list()[0], job)
|
||||
|
||||
async def test_hold_refused_without_waiting(self):
|
||||
registry = JobRegistry()
|
||||
gate = asyncio.Event()
|
||||
scan = self._start(registry, "scan", "records", gate)
|
||||
with self.assertRaises(JobBusy) as raised:
|
||||
async with registry.hold("maintenance", "Maintenance", admission="records"):
|
||||
self.fail("a refused hold must not run its body")
|
||||
self.assertIs(raised.exception.running, scan)
|
||||
self.assertEqual([j.kind for j in registry.list()], ["scan"])
|
||||
gate.set()
|
||||
await asyncio.wait_for(scan.task, 2)
|
||||
|
||||
async def test_hold_released_on_exception(self):
|
||||
registry = JobRegistry()
|
||||
with self.assertRaises(RuntimeError):
|
||||
async with registry.hold(
|
||||
"maintenance", "Maintenance", admission="records"
|
||||
) as job:
|
||||
raise RuntimeError("boom")
|
||||
self.assertEqual(job.state, "failed")
|
||||
again = registry.hold("maintenance", "Maintenance", admission="records")
|
||||
async with again as second:
|
||||
second.finish("failed") # the body may end it itself
|
||||
self.assertEqual(second.state, "failed")
|
||||
|
||||
async def test_hold_admits_on_entry_not_at_the_call(self):
|
||||
# A hold created but never entered (an await or an exception between
|
||||
# the call and ``async with``) must not leave a running records job
|
||||
# behind that blocks all record work until a restart.
|
||||
registry = JobRegistry()
|
||||
held = registry.hold("restore", "Restore", admission="records")
|
||||
self.assertEqual(registry.list(), [])
|
||||
self.assertIsNone(registry.blocker("scan", admission="records"))
|
||||
del held
|
||||
async with registry.hold("restore", "Restore", admission="records") as job:
|
||||
self.assertIs(registry.active("restore"), job)
|
||||
self.assertEqual(job.state, "finished")
|
||||
|
||||
def test_job_defaults_stay_read_and_cancellable(self):
|
||||
job = Job("duplicates", "Duplicate scan")
|
||||
self.assertEqual((job.admission, job.cancellable), ("read", True))
|
||||
|
||||
|
||||
class _LegacyHoldCase(AdminAPITestCase):
|
||||
"""The admin test app plus a running workspace job on demand."""
|
||||
|
||||
def _run_job(self, kind="scan", title="Scan for metadata", admission="records"):
|
||||
gate = asyncio.Event()
|
||||
job = self.api._jobs.start(
|
||||
kind, title, admission=admission, scope={}, runner=_gated(gate)
|
||||
)
|
||||
self.addAsyncCleanup(self._release, gate, job)
|
||||
return gate, job
|
||||
|
||||
@staticmethod
|
||||
async def _release(gate, job):
|
||||
gate.set()
|
||||
await asyncio.wait_for(job.task, 5)
|
||||
|
||||
def _db_digest(self):
|
||||
path = Path(self.api.db.model.db_path)
|
||||
return {
|
||||
suffix: hashlib.sha256(Path(f"{path}{suffix}").read_bytes()).hexdigest()
|
||||
for suffix in ("", "-wal")
|
||||
if Path(f"{path}{suffix}").exists()
|
||||
}
|
||||
|
||||
async def _busy(self, resp, title="Scan for metadata"):
|
||||
body = await resp.json()
|
||||
self.assertEqual(resp.status, 409)
|
||||
self.assertEqual(body, {"success": False, "error": f"{title} already running"})
|
||||
|
||||
|
||||
class TestLegacyRoutesHoldTheRegistry(_LegacyHoldCase):
|
||||
|
||||
async def test_hold_is_listed_as_running_job(self):
|
||||
held = self.api._jobs.hold(
|
||||
"restore", "Database restore (admin page)", admission="records"
|
||||
)
|
||||
async with held:
|
||||
resp = await self.client.request("GET", "/prompt_manager/jobs")
|
||||
jobs = (await resp.json())["jobs"]
|
||||
self.assertEqual(len(jobs), 1)
|
||||
job = jobs[0]
|
||||
self.assertEqual(
|
||||
(job["kind"], job["title"], job["state"]),
|
||||
("restore", "Database restore (admin page)", "running"),
|
||||
)
|
||||
self.assertIs(job["cancellable"], False)
|
||||
self.assertEqual(job["admission"], "records")
|
||||
|
||||
async def test_legacy_maintenance_and_cleanup_refused_while_scan_runs(self):
|
||||
self.api.db.save_prompt(text="keep me", prompt_hash="h1")
|
||||
before = self._db_digest()
|
||||
self._run_job()
|
||||
resp = await self.client.request(
|
||||
"POST",
|
||||
"/prompt_manager/maintenance",
|
||||
json={"operations": ["cleanup_duplicates", "vacuum"]},
|
||||
)
|
||||
await self._busy(resp)
|
||||
resp = await self.client.request("POST", "/prompt_manager/cleanup")
|
||||
await self._busy(resp)
|
||||
self.assertEqual(self._db_digest(), before)
|
||||
|
||||
async def test_read_only_maintenance_runs_during_scan(self):
|
||||
# The legacy statistics and consistency checks only read: they hold
|
||||
# the registry as "read", so a long scan does not block them.
|
||||
self.api.db.save_prompt(text="keep me", prompt_hash="h1")
|
||||
_gate, scan = self._run_job()
|
||||
listed = []
|
||||
real = self.api._run_maintenance_operations
|
||||
|
||||
def observed(operations):
|
||||
listed.extend(
|
||||
(job.kind, job.admission, job.state) for job in self.api._jobs.list()
|
||||
)
|
||||
return real(operations)
|
||||
|
||||
self.api._run_maintenance_operations = observed
|
||||
resp = await self.client.request(
|
||||
"POST",
|
||||
"/prompt_manager/maintenance",
|
||||
json={
|
||||
"operations": [
|
||||
"statistics",
|
||||
"check_consistency",
|
||||
"check_hash_duplicates",
|
||||
]
|
||||
},
|
||||
)
|
||||
body = await resp.json()
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertIs(body["success"], True)
|
||||
self.assertEqual(
|
||||
set(body["results"]),
|
||||
{"statistics", "check_consistency", "check_hash_duplicates"},
|
||||
)
|
||||
self.assertIn(("maintenance", "read", "running"), listed)
|
||||
self.assertEqual(scan.state, "running")
|
||||
maintenance = self.api._jobs.list()[-1]
|
||||
self.assertEqual(
|
||||
(maintenance.kind, maintenance.admission, maintenance.state),
|
||||
("maintenance", "read", "finished"),
|
||||
)
|
||||
|
||||
async def test_mixed_maintenance_selection_is_refused_during_scan(self):
|
||||
# One write operation among read-only ones makes the run "records".
|
||||
self.api.db.save_prompt(text="keep me", prompt_hash="h1")
|
||||
before = self._db_digest()
|
||||
self._run_job()
|
||||
resp = await self.client.request(
|
||||
"POST",
|
||||
"/prompt_manager/maintenance",
|
||||
json={"operations": ["statistics", "vacuum"]},
|
||||
)
|
||||
await self._busy(resp)
|
||||
self.assertEqual(self._db_digest(), before)
|
||||
|
||||
async def test_default_maintenance_selection_is_refused_during_scan(self):
|
||||
# No "operations" key means the default list, which vacuums.
|
||||
self.api.db.save_prompt(text="keep me", prompt_hash="h1")
|
||||
before = self._db_digest()
|
||||
self._run_job()
|
||||
resp = await self.client.request("POST", "/prompt_manager/maintenance", json={})
|
||||
await self._busy(resp)
|
||||
self.assertEqual(self._db_digest(), before)
|
||||
self.assertEqual([j.kind for j in self.api._jobs.list()], ["scan"])
|
||||
|
||||
async def test_legacy_maintenance_refused_while_image_tags_runs(self):
|
||||
self._run_job("image-tags", "Add tags to 3 images")
|
||||
resp = await self.client.request("POST", "/prompt_manager/maintenance")
|
||||
await self._busy(resp, "Add tags to 3 images")
|
||||
|
||||
async def test_scan_refused_while_legacy_maintenance_holds(self):
|
||||
entered = threading.Event()
|
||||
release = threading.Event()
|
||||
real = self.api._run_maintenance_operations
|
||||
|
||||
def gated(operations):
|
||||
entered.set()
|
||||
release.wait(10)
|
||||
return real(operations)
|
||||
|
||||
self.api._run_maintenance_operations = gated
|
||||
loop = asyncio.get_running_loop()
|
||||
maintenance = asyncio.ensure_future(
|
||||
self.client.request(
|
||||
"POST", "/prompt_manager/maintenance", json={"operations": ["vacuum"]}
|
||||
)
|
||||
)
|
||||
self.assertTrue(await loop.run_in_executor(None, entered.wait, 10))
|
||||
resp = await self.client.request("POST", "/prompt_manager/jobs/scan")
|
||||
body = await resp.json()
|
||||
self.assertEqual((resp.status, body["code"]), (409, "busy"))
|
||||
self.assertEqual(body["running"]["title"], "Maintenance (admin page)")
|
||||
release.set()
|
||||
self.assertEqual((await maintenance).status, 200)
|
||||
listed = (
|
||||
await (await self.client.request("GET", "/prompt_manager/jobs")).json()
|
||||
)["jobs"]
|
||||
self.assertEqual(
|
||||
[(j["kind"], j["state"]) for j in listed], [("maintenance", "finished")]
|
||||
)
|
||||
|
||||
async def test_legacy_duplicate_delete_refused_while_scan_runs(self):
|
||||
victim = self.output_dir / "dup.png"
|
||||
victim.write_bytes(b"x")
|
||||
self._run_job()
|
||||
resp = await self.client.request(
|
||||
"POST",
|
||||
"/prompt_manager/delete_duplicate_images",
|
||||
json={"image_paths": [str(victim)]},
|
||||
)
|
||||
await self._busy(resp)
|
||||
self.assertTrue(victim.exists())
|
||||
# The duplicate family lock was not taken by the refused request.
|
||||
self.assertIsNotNone(self.api._acquire_long_job(self.api.LONG_JOB_DUPLICATES))
|
||||
|
||||
async def test_lora_scan_refused_while_scan_runs(self):
|
||||
from py.config import IntegrationConfig
|
||||
|
||||
lm = Path(self.tmpdir) / "custom_nodes" / "ComfyUI-Lora-Manager"
|
||||
(lm / "py").mkdir(parents=True)
|
||||
self._run_job()
|
||||
imported = []
|
||||
with (
|
||||
patch.multiple(
|
||||
IntegrationConfig, LORA_MANAGER_ENABLED=True, LORA_MANAGER_PATH=str(lm)
|
||||
),
|
||||
patch("py.lora_utils.resolve_lora_manager_path", return_value=str(lm)),
|
||||
patch.object(
|
||||
self.api, "_lora_scan_stream", side_effect=lambda *a: imported.append(a)
|
||||
),
|
||||
):
|
||||
resp = await self.client.request("POST", "/prompt_manager/lora/scan")
|
||||
text = await resp.text()
|
||||
self.assertEqual(resp.status, 200) # the legacy page reads a stream
|
||||
events = [json.loads(line[6:]) for line in text.split("\n\n") if line]
|
||||
# The admin page's modal shows ``status`` and closes on ``type: error``.
|
||||
message = "Scan for metadata already running"
|
||||
self.assertEqual(
|
||||
events,
|
||||
[{"type": "error", "status": message, "progress": 0, "message": message}],
|
||||
)
|
||||
self.assertEqual(imported, [])
|
||||
|
||||
async def test_legacy_autotag_refused_while_scan_runs(self):
|
||||
# Both pages read this route as a stream (a 409 reads as a lost
|
||||
# connection there), so a refusal by another job is one error frame:
|
||||
# the admin page keys on ``type``, the gallery on ``status``.
|
||||
self._run_job()
|
||||
resp = await self.client.request(
|
||||
"POST", "/prompt_manager/autotag/start", json={}
|
||||
)
|
||||
text = await resp.text()
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertEqual(resp.headers["Content-Type"], "text/event-stream")
|
||||
events = [json.loads(line[6:]) for line in text.split("\n\n") if line]
|
||||
message = "Scan for metadata already running"
|
||||
self.assertEqual(
|
||||
events, [{"type": "error", "status": "error", "message": message}]
|
||||
)
|
||||
self.assertEqual(
|
||||
[j.kind for j in self.api._jobs.list() if j.kind == "autotag"], []
|
||||
)
|
||||
|
||||
async def test_second_legacy_autotag_batch_keeps_its_json_409(self):
|
||||
from py.api.autotag_routes import _BUSY_ERROR
|
||||
|
||||
self._run_job("autotag", "Auto-tag images (legacy page)")
|
||||
resp = await self.client.request(
|
||||
"POST", "/prompt_manager/autotag/start", json={}
|
||||
)
|
||||
self.assertEqual(resp.status, 409)
|
||||
self.assertEqual(await resp.json(), {"success": False, "error": _BUSY_ERROR})
|
||||
|
||||
async def test_legacy_payloads_unchanged_when_idle(self):
|
||||
resp = await self.client.request(
|
||||
"POST", "/prompt_manager/maintenance", json={"operations": ["vacuum"]}
|
||||
)
|
||||
body = await resp.json()
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertEqual(
|
||||
set(body),
|
||||
{"success", "operations_completed", "all_successful", "results", "message"},
|
||||
)
|
||||
resp = await self.client.request("POST", "/prompt_manager/cleanup")
|
||||
body = await resp.json()
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertEqual(set(body), {"success", "message", "duplicates_removed"})
|
||||
resp = await self.client.request(
|
||||
"POST", "/prompt_manager/delete_duplicate_images", json={"image_paths": []}
|
||||
)
|
||||
self.assertEqual(
|
||||
(resp.status, await resp.json()),
|
||||
(400, {"success": False, "error": "No image paths provided"}),
|
||||
)
|
||||
# Each hold ended with the request; refused input never took one.
|
||||
listed = await self._listed()
|
||||
self.assertEqual(
|
||||
[(j["kind"], j["state"]) for j in listed],
|
||||
[("maintenance", "finished"), ("maintenance", "finished")],
|
||||
)
|
||||
self.assertEqual(
|
||||
[j["title"] for j in listed],
|
||||
["Prompt cleanup (admin page)", "Maintenance (admin page)"],
|
||||
)
|
||||
|
||||
async def _listed(self):
|
||||
resp = await self.client.request("GET", "/prompt_manager/jobs")
|
||||
return (await resp.json())["jobs"]
|
||||
|
||||
|
||||
class TestHeldJobStates(_LegacyHoldCase):
|
||||
"""Activity lists what a held legacy operation really did (Minor 7)."""
|
||||
|
||||
async def _listed(self):
|
||||
resp = await self.client.request("GET", "/prompt_manager/jobs")
|
||||
return [(j["kind"], j["state"]) for j in (await resp.json())["jobs"]]
|
||||
|
||||
async def test_refused_input_takes_no_hold(self):
|
||||
resp = await self.client.request(
|
||||
"POST", "/prompt_manager/delete_duplicate_images", json={}
|
||||
)
|
||||
self.assertEqual(resp.status, 400)
|
||||
resp = await self.client.request(
|
||||
"POST",
|
||||
"/prompt_manager/images/generate-thumbnails",
|
||||
data=b"{not json",
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
self.assertEqual(resp.status, 400)
|
||||
self.assertEqual(await self._listed(), [])
|
||||
|
||||
async def test_thumbnail_stream_that_ends_in_error_is_failed(self):
|
||||
with patch.object(self.api, "_get_all_output_dirs", return_value=[]):
|
||||
resp = await self.client.request(
|
||||
"POST", "/prompt_manager/images/generate-thumbnails/progress"
|
||||
)
|
||||
text = await resp.text()
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertIn("event: error", text)
|
||||
self.assertEqual(await self._listed(), [("thumbnails", "failed")])
|
||||
|
||||
async def test_legacy_autotag_that_ends_in_error_is_failed(self):
|
||||
service = MagicMock()
|
||||
service.get_models_status.return_value = {} # model not downloaded
|
||||
with patch("py.autotag.get_autotag_service", return_value=service):
|
||||
resp = await self.client.request(
|
||||
"POST", "/prompt_manager/autotag/start", json={}
|
||||
)
|
||||
text = await resp.text()
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertIn('"type": "error"', text)
|
||||
self.assertEqual(await self._listed(), [("autotag", "failed")])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -182,8 +182,13 @@ class TestJobRegistry(unittest.IsolatedAsyncioTestCase):
|
||||
"scope",
|
||||
"listed",
|
||||
"log_length",
|
||||
"cancellable",
|
||||
"admission",
|
||||
},
|
||||
)
|
||||
self.assertEqual(
|
||||
(summary["cancellable"], summary["admission"]), (True, "records")
|
||||
)
|
||||
self.assertEqual(summary["scope"], {"files": "new", "output_folders": 2})
|
||||
# What the detail would list, so a view reads it only when that changed.
|
||||
self.assertEqual((summary["listed"], summary["log_length"]), ({}, 0))
|
||||
|
||||
+9
-9
@@ -1,9 +1,9 @@
|
||||
{
|
||||
"admin.js": {
|
||||
"lines": 28,
|
||||
"lines": 31,
|
||||
"branches": 58,
|
||||
"functions": 23,
|
||||
"statements": 28
|
||||
"functions": 26,
|
||||
"statements": 31
|
||||
},
|
||||
"api-client.js": {
|
||||
"lines": 98,
|
||||
@@ -42,10 +42,10 @@
|
||||
"statements": 98
|
||||
},
|
||||
"gallery.js": {
|
||||
"lines": 39,
|
||||
"lines": 40,
|
||||
"branches": 63,
|
||||
"functions": 43,
|
||||
"statements": 39
|
||||
"functions": 44,
|
||||
"statements": 40
|
||||
},
|
||||
"html-escape.js": {
|
||||
"lines": 98,
|
||||
@@ -157,7 +157,7 @@
|
||||
},
|
||||
"workspace/activity-job-view.mjs": {
|
||||
"lines": 97,
|
||||
"branches": 83,
|
||||
"branches": 86,
|
||||
"functions": 98,
|
||||
"statements": 97
|
||||
},
|
||||
@@ -174,10 +174,10 @@
|
||||
"statements": 98
|
||||
},
|
||||
"workspace/activity.mjs": {
|
||||
"lines": 95,
|
||||
"lines": 96,
|
||||
"branches": 87,
|
||||
"functions": 98,
|
||||
"statements": 95
|
||||
"statements": 96
|
||||
},
|
||||
"workspace/file-inspector.mjs": {
|
||||
"lines": 96,
|
||||
|
||||
+13
-3
@@ -1180,6 +1180,11 @@
|
||||
`Done — ${data.imported} imported, ${data.skipped} skipped`;
|
||||
this.loadStatistics();
|
||||
setTimeout(() => this.hideModal("loraImportModal"), 2000);
|
||||
} else if (data.type === "error") {
|
||||
// A refusal (another job runs) or a failed import: name it and close the modal.
|
||||
document.getElementById("loraImportStatus").textContent = `Error: ${data.message}`;
|
||||
this.showNotification(`LoRA import failed: ${data.message}`, "error");
|
||||
setTimeout(() => this.hideModal("loraImportModal"), 3000);
|
||||
}
|
||||
} catch (e) { /* skip malformed SSE */ }
|
||||
}
|
||||
@@ -1231,10 +1236,13 @@
|
||||
}
|
||||
this.hideModal("settingsModal");
|
||||
} else {
|
||||
throw new Error("Failed to save settings");
|
||||
// A refusal names its reason (a corrupt config.json, a changed file): show it.
|
||||
const data = await response.json().catch(() => ({}));
|
||||
throw new Error(data.error ? `Failed to save settings: ${data.error}` : "Failed to save settings");
|
||||
}
|
||||
} catch (error) {
|
||||
this.showNotification("Failed to save settings", "error");
|
||||
const reason = error.message.startsWith("Failed to save settings:") ? error.message : "Failed to save settings";
|
||||
this.showNotification(reason, "error");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1848,7 +1856,9 @@
|
||||
throw new Error(data.error || 'Maintenance failed');
|
||||
}
|
||||
} else {
|
||||
throw new Error(`HTTP ${response.status}: ${response.statusText}`);
|
||||
// A refusal names the job that runs ("Scan for metadata already running").
|
||||
const data = await response.json().catch(() => ({}));
|
||||
throw new Error(data.error || `HTTP ${response.status}: ${response.statusText}`);
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Maintenance error:', error);
|
||||
|
||||
+17
-3
@@ -17,6 +17,14 @@
|
||||
this.dupResults = null;
|
||||
|
||||
this.api = ApiClient.createApiClient();
|
||||
// The page's lifetime ends when it starts to leave: WebKit fails a fetch a leaving page starts
|
||||
// ("Fetch API cannot load … due to access control checks.") and logs it as a page error.
|
||||
this.pageLifetime = new AbortController();
|
||||
window.addEventListener("beforeunload", () => this.pageLifetime.abort());
|
||||
window.addEventListener("pagehide", () => this.pageLifetime.abort());
|
||||
window.addEventListener("pageshow", (e) => {
|
||||
if (e.persisted && this.pageLifetime.signal.aborted) this.pageLifetime = new AbortController();
|
||||
});
|
||||
// Only the newest page request may update the grid
|
||||
this.imagesRequest = ApiClient.latestOnly();
|
||||
|
||||
@@ -2129,8 +2137,11 @@ Seed: ${this.currentMetadata.seed || 'Unknown'}`;
|
||||
headers: { 'Content-Type': 'application/json' }
|
||||
});
|
||||
|
||||
if (response.ok) {
|
||||
const data = await response.json();
|
||||
if (!response.ok) {
|
||||
// Refused (a running job holds the thumbnails): say why and claim nothing.
|
||||
const data = await response.json().catch(() => ({}));
|
||||
this.showNotification(`Thumbnails were not cleared: ${data.error || `HTTP ${response.status}`}`, 'error');
|
||||
return;
|
||||
}
|
||||
} catch (e) {
|
||||
console.warn('Could not clear server thumbnails:', e);
|
||||
@@ -2531,8 +2542,11 @@ Seed: ${this.currentMetadata.seed || 'Unknown'}`;
|
||||
return;
|
||||
}
|
||||
|
||||
// A leaving page starts no read, and one it aborts is not reported.
|
||||
const signal = this.pageLifetime?.signal;
|
||||
if (signal?.aborted) return;
|
||||
try {
|
||||
const data = await this.api.get('/prompt_manager/images/thumbnails/status');
|
||||
const data = await this.api.get('/prompt_manager/images/thumbnails/status', { signal });
|
||||
const total = data.total_images;
|
||||
const pending = data.pending;
|
||||
if (!data.success || !Number.isSafeInteger(total) ||
|
||||
|
||||
@@ -274,7 +274,8 @@ export function mountActivityJobView(main, { store, scope, api, jobId }) {
|
||||
const lost = job.state === 'lost';
|
||||
if (lost && document.activeElement === download) parts.h1.focus();
|
||||
download.disabled = lost;
|
||||
if (job.state === 'running') {
|
||||
// Truth rule: a job that cannot be cancelled (a held legacy-page job) shows no Cancel.
|
||||
if (job.state === 'running' && job.cancellable) {
|
||||
if (!cancel.isConnected) parts.actions.append(cancel);
|
||||
} else if (cancel.isConnected) {
|
||||
// A focused Cancel that goes away hands its focus to the job's heading, never to the page body.
|
||||
|
||||
@@ -103,6 +103,8 @@ export function parseJob(raw) {
|
||||
scope: plainObject(raw.scope) ? { ...raw.scope } : {},
|
||||
listed,
|
||||
log_length: count(raw.log_length),
|
||||
// A held legacy-page job cannot be cancelled; an older server sends no key (cancellable then).
|
||||
cancellable: raw.cancellable !== false,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -153,11 +153,15 @@ function itemFor(job, scope) {
|
||||
progress.textContent = `${next.progress}%`;
|
||||
progress.setAttribute('aria-label', `${next.title} progress`);
|
||||
if (!progress.isConnected) item.insertBefore(progress, actions.isConnected ? actions : null);
|
||||
if (!actions.isConnected) item.append(actions);
|
||||
} else {
|
||||
progress.remove();
|
||||
}
|
||||
// Truth rule: a job that cannot be cancelled (a held legacy-page job) shows no Cancel job.
|
||||
if (running && next.cancellable) {
|
||||
if (!actions.isConnected) item.append(actions);
|
||||
} else if (actions.isConnected) {
|
||||
// A focused Cancel job that goes away hands its focus to the item's title link, never to the page body.
|
||||
if (actions.contains(document.activeElement)) link.focus();
|
||||
progress.remove();
|
||||
actions.remove();
|
||||
}
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user