fix: no absolute server paths in API responses
Settings, diagnostics, duplicate/output scans and the logging endpoints returned absolute filesystem paths, and several error envelopes echoed str(e) which for OSError carries the full path. - _public_path() renders a path relative to the ComfyUI base directory (folder_paths.base_path, else the output dir's parent) or to the parent of an opted-in extra gallery root, falling back to the basename; _public_error() reduces OSError to strerror + basename. - GET /settings returns gallery roots and monitored directories in that form; POST /settings resolves relative roots against the same anchors (GalleryConfig.resolve_gallery_root) so the GET output round-trips. - Diagnostics output_dirs, monitored_directories and the missing-DB message no longer carry absolute paths; the image monitor is reached through the package-or-top-level import fallback instead of a three-dot relative import that failed outside the package. - Duplicate scan 'path' is now relative to the output root and the delete endpoint accepts relative paths. - /logs/files 'path' and /logs/stats 'log_directory' use the public form; unexpected errors answer with a generic message.
This commit is contained in:
Vito Sansevero
@@ -44,6 +44,37 @@ def _get_project_root():
|
||||
return os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
|
||||
def _public_path(path) -> str:
|
||||
"""Render a server path for an API response without revealing the layout.
|
||||
|
||||
Returns the path relative to the ComfyUI base directory (or to the parent
|
||||
of an opted-in extra gallery root), as a POSIX-style string, or just the
|
||||
basename when the path is not under any of those anchors.
|
||||
"""
|
||||
if not path:
|
||||
return ""
|
||||
from ..config import GalleryConfig
|
||||
|
||||
try:
|
||||
canonical = Path(os.path.normcase(os.path.realpath(str(path))))
|
||||
except (OSError, ValueError):
|
||||
return os.path.basename(str(path))
|
||||
for anchor in GalleryConfig.path_anchors():
|
||||
if canonical.is_relative_to(Path(anchor)):
|
||||
return canonical.relative_to(Path(anchor)).as_posix() or "."
|
||||
return canonical.name
|
||||
|
||||
|
||||
def _public_error(exc: BaseException) -> str:
|
||||
"""Error text safe for API responses: OSError paths reduced to basenames."""
|
||||
if isinstance(exc, OSError):
|
||||
parts = [exc.strerror or type(exc).__name__]
|
||||
if exc.filename:
|
||||
parts.append(os.path.basename(str(exc.filename)))
|
||||
return ": ".join(parts)
|
||||
return str(exc) or type(exc).__name__
|
||||
|
||||
|
||||
# ── Gzip compression middleware ────────────────────────────────────────
|
||||
_GZIP_MIN_SIZE = 1024 # Only compress bodies larger than 1 KB
|
||||
_GZIP_TYPES = frozenset(
|
||||
@@ -140,6 +171,9 @@ class PromptManagerAPI(
|
||||
|
||||
self.logger.info("PromptManager API initialization completed")
|
||||
|
||||
_public_path = staticmethod(_public_path)
|
||||
_public_error = staticmethod(_public_error)
|
||||
|
||||
async def _run_in_executor(self, func, *args, **kwargs):
|
||||
"""Run a blocking function in the default thread pool executor.
|
||||
|
||||
|
||||
+67
-47
@@ -24,6 +24,20 @@ except ImportError:
|
||||
from database.operations import PromptDatabase
|
||||
|
||||
|
||||
def _image_monitor_module():
|
||||
"""The utils.image_monitor module under either package identity."""
|
||||
try:
|
||||
from ...utils import image_monitor
|
||||
except ImportError:
|
||||
from utils import image_monitor
|
||||
return image_monitor
|
||||
|
||||
|
||||
def _current_image_monitor():
|
||||
"""The running ImageMonitor singleton, or None when not started."""
|
||||
return getattr(_image_monitor_module(), "_monitor_instance", None)
|
||||
|
||||
|
||||
class AdminRoutesMixin:
|
||||
"""Mixin providing admin, diagnostics, and maintenance API endpoints."""
|
||||
|
||||
@@ -90,12 +104,9 @@ class AdminRoutesMixin:
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Scan duplicates error: {e}")
|
||||
self.logger.error(f"Scan duplicates error: {e}", exc_info=True)
|
||||
return web.json_response(
|
||||
{
|
||||
"success": False,
|
||||
"error": f"Failed to scan duplicate images: {str(e)}",
|
||||
},
|
||||
{"success": False, "error": "Failed to scan duplicate images"},
|
||||
status=500,
|
||||
)
|
||||
|
||||
@@ -113,9 +124,9 @@ class AdminRoutesMixin:
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Cleanup error: {e}")
|
||||
self.logger.error(f"Cleanup error: {e}", exc_info=True)
|
||||
return web.json_response(
|
||||
{"success": False, "error": f"Failed to cleanup duplicates: {str(e)}"},
|
||||
{"success": False, "error": "Failed to cleanup duplicates"},
|
||||
status=500,
|
||||
)
|
||||
|
||||
@@ -189,7 +200,7 @@ class AdminRoutesMixin:
|
||||
image_info = {
|
||||
"id": str(hash(str(media_path))),
|
||||
"filename": media_path.name,
|
||||
"path": str(media_path),
|
||||
"path": str(rel_path),
|
||||
"relative_path": str(rel_path),
|
||||
"url": f"/prompt_manager/images/serve/{rel_path.as_posix()}",
|
||||
"thumbnail_url": thumbnail_url,
|
||||
@@ -266,6 +277,8 @@ class AdminRoutesMixin:
|
||||
|
||||
output_path = Path(output_dir)
|
||||
file_path = Path(image_path)
|
||||
if not file_path.is_absolute():
|
||||
file_path = output_path / file_path
|
||||
|
||||
# Security check - ensure file is within output directory
|
||||
try:
|
||||
@@ -307,7 +320,7 @@ class AdminRoutesMixin:
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error deleting file {image_path}: {e}")
|
||||
failed_files.append(f"{image_path} ({str(e)})")
|
||||
failed_files.append(f"{image_path} ({self._public_error(e)})")
|
||||
failed_count += 1
|
||||
|
||||
response_data = {
|
||||
@@ -324,12 +337,9 @@ class AdminRoutesMixin:
|
||||
return web.json_response(response_data)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Delete duplicate images error: {e}")
|
||||
self.logger.error(f"Delete duplicate images error: {e}", exc_info=True)
|
||||
return web.json_response(
|
||||
{
|
||||
"success": False,
|
||||
"error": f"Failed to delete duplicate images: {str(e)}",
|
||||
},
|
||||
{"success": False, "error": "Failed to delete duplicate images"},
|
||||
status=500,
|
||||
)
|
||||
|
||||
@@ -341,9 +351,9 @@ class AdminRoutesMixin:
|
||||
return web.json_response({"success": True, "stats": stats})
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Stats error: {e}")
|
||||
self.logger.error(f"Stats error: {e}", exc_info=True)
|
||||
return web.json_response(
|
||||
{"success": False, "error": f"Failed to get statistics: {str(e)}"},
|
||||
{"success": False, "error": "Failed to get statistics"},
|
||||
status=500,
|
||||
)
|
||||
|
||||
@@ -375,27 +385,27 @@ class AdminRoutesMixin:
|
||||
if GalleryConfig.MONITORING_DIRECTORIES:
|
||||
monitored_dirs = GalleryConfig.MONITORING_DIRECTORIES
|
||||
|
||||
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,
|
||||
"gallery_root_paths": list(
|
||||
GalleryConfig.MONITORING_DIRECTORIES
|
||||
),
|
||||
"gallery_root_path": (
|
||||
GalleryConfig.MONITORING_DIRECTORIES[0]
|
||||
if GalleryConfig.MONITORING_DIRECTORIES
|
||||
else ""
|
||||
),
|
||||
"monitored_directories": monitored_dirs,
|
||||
"gallery_root_paths": root_paths,
|
||||
"gallery_root_path": root_paths[0] if root_paths else "",
|
||||
"monitored_directories": [
|
||||
self._public_path(d) for d in monitored_dirs
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
except Exception as e:
|
||||
self.logger.error(f"Get settings error: {e}", exc_info=True)
|
||||
return web.json_response(
|
||||
{"success": False, "error": f"Failed to get settings: {str(e)}"},
|
||||
{"success": False, "error": "Failed to get settings"},
|
||||
status=500,
|
||||
)
|
||||
|
||||
@@ -477,7 +487,7 @@ class AdminRoutesMixin:
|
||||
None,
|
||||
f"Invalid gallery root '{os.path.basename(entry)}': {reason}",
|
||||
)
|
||||
roots.append(entry)
|
||||
roots.append(GalleryConfig.resolve_gallery_root(entry))
|
||||
return roots, None
|
||||
|
||||
def _apply_gallery_roots(self, roots):
|
||||
@@ -548,12 +558,14 @@ class AdminRoutesMixin:
|
||||
else:
|
||||
results["database"] = {
|
||||
"status": "error",
|
||||
"message": f"Database file not found: {db_path}",
|
||||
"message": (
|
||||
f"Database file not found: {os.path.basename(db_path)}"
|
||||
),
|
||||
}
|
||||
except Exception as e:
|
||||
results["database"] = {
|
||||
"status": "error",
|
||||
"message": f"Database error: {str(e)}",
|
||||
"message": f"Database error: {self._public_error(e)}",
|
||||
}
|
||||
|
||||
# Check dependencies
|
||||
@@ -616,16 +628,18 @@ class AdminRoutesMixin:
|
||||
|
||||
results["comfyui_output"] = {
|
||||
"status": "ok" if output_dirs else "warning",
|
||||
"output_dirs": output_dirs,
|
||||
"output_dirs": [self._public_path(d) for d in output_dirs],
|
||||
}
|
||||
|
||||
# Check image monitor status
|
||||
try:
|
||||
from ...utils import image_monitor as im_mod
|
||||
|
||||
monitor = im_mod._monitor_instance
|
||||
monitor = _current_image_monitor()
|
||||
if monitor is not None:
|
||||
monitor_status = monitor.get_status()
|
||||
monitor_status = dict(monitor.get_status())
|
||||
monitor_status["monitored_directories"] = [
|
||||
self._public_path(d)
|
||||
for d in monitor_status.get("monitored_directories", [])
|
||||
]
|
||||
results["image_monitor"] = {
|
||||
"status": (
|
||||
"ok" if monitor_status.get("observer_alive") else "error"
|
||||
@@ -640,14 +654,16 @@ class AdminRoutesMixin:
|
||||
except Exception as e:
|
||||
results["image_monitor"] = {
|
||||
"status": "error",
|
||||
"message": f"Failed to get monitor status: {str(e)}",
|
||||
"message": f"Failed to get monitor status: {self._public_error(e)}",
|
||||
}
|
||||
|
||||
return web.json_response({"success": True, "diagnostics": results})
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Diagnostics error: {e}", exc_info=True)
|
||||
return web.json_response({"success": False, "error": str(e)}, status=500)
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Diagnostics failed"}, status=500
|
||||
)
|
||||
|
||||
async def test_image_link(self, request):
|
||||
"""Test creating an image link."""
|
||||
@@ -695,14 +711,18 @@ class AdminRoutesMixin:
|
||||
"success": False,
|
||||
"result": {
|
||||
"status": "error",
|
||||
"message": f"Failed to create test link: {str(e)}",
|
||||
"message": (
|
||||
f"Failed to create test link: {self._public_error(e)}"
|
||||
),
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Test link error: {e}", exc_info=True)
|
||||
return web.json_response({"success": False, "error": str(e)}, status=500)
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Test link failed"}, status=500
|
||||
)
|
||||
|
||||
async def run_maintenance(self, request):
|
||||
"""Perform comprehensive database maintenance and optimization."""
|
||||
@@ -731,7 +751,7 @@ class AdminRoutesMixin:
|
||||
except Exception as e:
|
||||
results["cleanup_duplicates"] = {
|
||||
"success": False,
|
||||
"error": str(e),
|
||||
"error": self._public_error(e),
|
||||
"message": "Failed to cleanup duplicates",
|
||||
}
|
||||
|
||||
@@ -745,7 +765,7 @@ class AdminRoutesMixin:
|
||||
except Exception as e:
|
||||
results["vacuum"] = {
|
||||
"success": False,
|
||||
"error": str(e),
|
||||
"error": self._public_error(e),
|
||||
"message": "Failed to vacuum database",
|
||||
}
|
||||
|
||||
@@ -760,7 +780,7 @@ class AdminRoutesMixin:
|
||||
except Exception as e:
|
||||
results["cleanup_orphaned_images"] = {
|
||||
"success": False,
|
||||
"error": str(e),
|
||||
"error": self._public_error(e),
|
||||
"message": "Failed to cleanup orphaned images",
|
||||
}
|
||||
|
||||
@@ -775,7 +795,7 @@ class AdminRoutesMixin:
|
||||
except Exception as e:
|
||||
results["check_hash_duplicates"] = {
|
||||
"success": False,
|
||||
"error": str(e),
|
||||
"error": self._public_error(e),
|
||||
"message": "Failed to check hash duplicates",
|
||||
}
|
||||
|
||||
@@ -790,7 +810,7 @@ class AdminRoutesMixin:
|
||||
except Exception as e:
|
||||
results["statistics"] = {
|
||||
"success": False,
|
||||
"error": str(e),
|
||||
"error": self._public_error(e),
|
||||
"message": "Failed to get database statistics",
|
||||
}
|
||||
|
||||
@@ -805,7 +825,7 @@ class AdminRoutesMixin:
|
||||
except Exception as e:
|
||||
results["prune_orphaned_prompts"] = {
|
||||
"success": False,
|
||||
"error": str(e),
|
||||
"error": self._public_error(e),
|
||||
"message": "Failed to prune orphaned prompts",
|
||||
}
|
||||
|
||||
@@ -821,7 +841,7 @@ class AdminRoutesMixin:
|
||||
except Exception as e:
|
||||
results["check_consistency"] = {
|
||||
"success": False,
|
||||
"error": str(e),
|
||||
"error": self._public_error(e),
|
||||
"message": "Failed to check database consistency",
|
||||
}
|
||||
|
||||
@@ -844,7 +864,7 @@ class AdminRoutesMixin:
|
||||
except Exception as e:
|
||||
self.logger.error(f"Maintenance error: {e}", exc_info=True)
|
||||
return web.json_response(
|
||||
{"success": False, "error": f"Maintenance failed: {str(e)}"}, status=500
|
||||
{"success": False, "error": "Maintenance failed"}, status=500
|
||||
)
|
||||
|
||||
async def backup_database(self, request):
|
||||
@@ -1020,7 +1040,7 @@ class AdminRoutesMixin:
|
||||
yield f"data: {json.dumps({'type': 'error', 'message': 'No output directories found. Configure scan directories in Settings.'})}\n\n"
|
||||
return
|
||||
|
||||
dir_names = [str(d) for d in output_dirs]
|
||||
dir_names = [self._public_path(d) for d in output_dirs]
|
||||
yield f"data: {json.dumps({'type': 'progress', 'progress': 0, 'status': f'Scanning {len(output_dirs)} directory(ies) for media files...', 'processed': 0, 'found': 0})}\n\n"
|
||||
|
||||
media_files = await self._run_in_executor(
|
||||
|
||||
@@ -131,14 +131,18 @@ class LoggingRoutesMixin:
|
||||
except Exception as e:
|
||||
self.logger.error(f"Get logs error: {e}")
|
||||
return web.json_response(
|
||||
{"success": False, "error": str(e), "logs": []}, status=500
|
||||
{"success": False, "error": self._public_error(e), "logs": []},
|
||||
status=500,
|
||||
)
|
||||
|
||||
async def get_log_files(self, request):
|
||||
"""Get information about available log files."""
|
||||
try:
|
||||
logger_manager = self._get_logger_manager()
|
||||
log_files = logger_manager.get_log_files()
|
||||
log_files = [
|
||||
{**entry, "path": self._public_path(entry.get("path"))}
|
||||
for entry in logger_manager.get_log_files()
|
||||
]
|
||||
|
||||
return web.json_response(
|
||||
{"success": True, "files": log_files, "count": len(log_files)}
|
||||
@@ -147,7 +151,8 @@ class LoggingRoutesMixin:
|
||||
except Exception as e:
|
||||
self.logger.error(f"Get log files error: {e}")
|
||||
return web.json_response(
|
||||
{"success": False, "error": str(e), "files": []}, status=500
|
||||
{"success": False, "error": self._public_error(e), "files": []},
|
||||
status=500,
|
||||
)
|
||||
|
||||
async def download_log_file(self, request):
|
||||
@@ -182,7 +187,9 @@ class LoggingRoutesMixin:
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Download log file error: {e}")
|
||||
return web.json_response({"success": False, "error": str(e)}, status=500)
|
||||
return web.json_response(
|
||||
{"success": False, "error": self._public_error(e)}, status=500
|
||||
)
|
||||
|
||||
async def truncate_logs(self, request):
|
||||
"""Truncate all log files."""
|
||||
@@ -200,7 +207,9 @@ class LoggingRoutesMixin:
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Truncate logs error: {e}")
|
||||
return web.json_response({"success": False, "error": str(e)}, status=500)
|
||||
return web.json_response(
|
||||
{"success": False, "error": self._public_error(e)}, status=500
|
||||
)
|
||||
|
||||
async def get_log_config(self, request):
|
||||
"""Get current logging configuration."""
|
||||
@@ -212,7 +221,9 @@ class LoggingRoutesMixin:
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Get log config error: {e}")
|
||||
return web.json_response({"success": False, "error": str(e)}, status=500)
|
||||
return web.json_response(
|
||||
{"success": False, "error": self._public_error(e)}, status=500
|
||||
)
|
||||
|
||||
async def update_log_config(self, request):
|
||||
"""Update logging configuration (whitelisted keys only)."""
|
||||
@@ -251,10 +262,13 @@ class LoggingRoutesMixin:
|
||||
"""Get logging statistics."""
|
||||
try:
|
||||
logger_manager = self._get_logger_manager()
|
||||
stats = logger_manager.get_log_stats()
|
||||
stats = dict(logger_manager.get_log_stats())
|
||||
stats["log_directory"] = self._public_path(stats.get("log_directory"))
|
||||
|
||||
return web.json_response({"success": True, "stats": stats})
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Get log stats error: {e}")
|
||||
return web.json_response({"success": False, "error": str(e)}, status=500)
|
||||
return web.json_response(
|
||||
{"success": False, "error": self._public_error(e)}, status=500
|
||||
)
|
||||
|
||||
+73
-3
@@ -73,11 +73,39 @@ def _is_filesystem_root(path: str) -> bool:
|
||||
return os.path.dirname(path) == path
|
||||
|
||||
|
||||
def _comfyui_directories() -> List[str]:
|
||||
"""Directories reported by ComfyUI's folder_paths module, if importable."""
|
||||
def _import_folder_paths():
|
||||
"""ComfyUI's folder_paths module, or None outside ComfyUI."""
|
||||
try:
|
||||
import folder_paths
|
||||
except ImportError:
|
||||
return None
|
||||
return folder_paths
|
||||
|
||||
|
||||
def _comfyui_base_directory() -> Optional[str]:
|
||||
"""ComfyUI's base directory: folder_paths.base_path, else the output dir's parent."""
|
||||
folder_paths = _import_folder_paths()
|
||||
if folder_paths is None:
|
||||
return None
|
||||
base = getattr(folder_paths, "base_path", None)
|
||||
if isinstance(base, str) and base:
|
||||
return base
|
||||
getter = getattr(folder_paths, "get_output_directory", None)
|
||||
if getter is None:
|
||||
return None
|
||||
try:
|
||||
output_dir = getter()
|
||||
except Exception:
|
||||
return None
|
||||
if isinstance(output_dir, str) and output_dir:
|
||||
return os.path.dirname(os.path.realpath(output_dir))
|
||||
return None
|
||||
|
||||
|
||||
def _comfyui_directories() -> List[str]:
|
||||
"""Directories reported by ComfyUI's folder_paths module, if importable."""
|
||||
folder_paths = _import_folder_paths()
|
||||
if folder_paths is None:
|
||||
return []
|
||||
|
||||
found = []
|
||||
@@ -272,6 +300,48 @@ class GalleryConfig:
|
||||
parents.append(canonical)
|
||||
return parents
|
||||
|
||||
@classmethod
|
||||
def path_anchors(cls) -> List[str]:
|
||||
"""Canonical directories that relative gallery paths are resolved against.
|
||||
|
||||
The ComfyUI base directory comes first, followed by the parent of each
|
||||
directory listed in PROMPT_MANAGER_EXTRA_GALLERY_ROOTS, so an extra
|
||||
root is addressed by its own name (``gallery/sub``) rather than by an
|
||||
absolute path. The same anchors drive the public (relative) form of
|
||||
paths in API responses.
|
||||
"""
|
||||
anchors = []
|
||||
base = _comfyui_base_directory()
|
||||
candidates = [base] if base else []
|
||||
candidates.extend(os.path.dirname(root) for root in _extra_gallery_roots())
|
||||
for candidate in candidates:
|
||||
try:
|
||||
canonical = _canonical_path(candidate)
|
||||
except (OSError, ValueError):
|
||||
continue
|
||||
if canonical not in anchors:
|
||||
anchors.append(canonical)
|
||||
return anchors
|
||||
|
||||
@classmethod
|
||||
def resolve_gallery_root(cls, path: str) -> str:
|
||||
"""Canonical absolute path for a gallery root given in any accepted form.
|
||||
|
||||
Absolute paths are canonicalised as-is. Relative paths are tried
|
||||
against each :meth:`path_anchors` entry and the first existing match
|
||||
wins; otherwise the first anchor (or the current directory when there
|
||||
is none) is used, so the caller's existence check reports it.
|
||||
"""
|
||||
path = path.strip()
|
||||
if os.path.isabs(path):
|
||||
return _canonical_path(path)
|
||||
anchors = cls.path_anchors()
|
||||
for anchor in anchors:
|
||||
candidate = os.path.join(anchor, path)
|
||||
if os.path.exists(candidate):
|
||||
return _canonical_path(candidate)
|
||||
return _canonical_path(os.path.join(anchors[0], path) if anchors else path)
|
||||
|
||||
@classmethod
|
||||
def validate_gallery_root(cls, path: Any) -> Tuple[bool, str]:
|
||||
"""Check whether ``path`` may be used as a gallery root.
|
||||
@@ -284,7 +354,7 @@ class GalleryConfig:
|
||||
return False, "Gallery root must be a non-empty path"
|
||||
|
||||
try:
|
||||
canonical = _canonical_path(path.strip())
|
||||
canonical = cls.resolve_gallery_root(path)
|
||||
except (OSError, ValueError):
|
||||
return False, "Gallery root could not be resolved"
|
||||
|
||||
|
||||
@@ -221,5 +221,132 @@ class TestSaveSettingsGalleryRoots(AdminAPITestCase):
|
||||
self.assertEqual(resp.status, 400)
|
||||
|
||||
|
||||
class FakeMonitor:
|
||||
def __init__(self, dirs):
|
||||
self.monitored_directories = list(dirs)
|
||||
|
||||
def get_status(self):
|
||||
return {
|
||||
"running": True,
|
||||
"monitored_directories": self.monitored_directories,
|
||||
"handler_active": True,
|
||||
"observer_alive": True,
|
||||
}
|
||||
|
||||
|
||||
class TestNoAbsolutePathsInResponses(AdminAPITestCase):
|
||||
"""No admin response may reveal the server's absolute filesystem layout."""
|
||||
|
||||
def _install_monitor(self, dirs):
|
||||
import utils.image_monitor as im_mod
|
||||
|
||||
orig = im_mod._monitor_instance
|
||||
im_mod._monitor_instance = FakeMonitor(dirs)
|
||||
self.addCleanup(setattr, im_mod, "_monitor_instance", orig)
|
||||
|
||||
async def _body(self, method, path, **kwargs):
|
||||
resp = await self.client.request(method, path, **kwargs)
|
||||
return resp, await resp.text()
|
||||
|
||||
async def test_settings_show_roots_relative_to_comfyui(self):
|
||||
renders = self.output_dir / "renders"
|
||||
renders.mkdir()
|
||||
GalleryConfig.MONITORING_DIRECTORIES = [str(renders)]
|
||||
self._install_monitor([str(renders)])
|
||||
|
||||
resp, body = await self._body("GET", "/prompt_manager/settings")
|
||||
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertNotIn(self.tmpdir, body)
|
||||
data = json.loads(body)
|
||||
self.assertEqual(data["settings"]["gallery_root_paths"], ["output/renders"])
|
||||
self.assertEqual(data["settings"]["gallery_root_path"], "output/renders")
|
||||
self.assertEqual(data["settings"]["monitored_directories"], ["output/renders"])
|
||||
|
||||
async def test_settings_relative_root_round_trips(self):
|
||||
"""What GET returns can be POSTed back unchanged from any CWD."""
|
||||
renders = self.output_dir / "renders"
|
||||
renders.mkdir()
|
||||
|
||||
resp = await self.client.request(
|
||||
"POST",
|
||||
"/prompt_manager/settings",
|
||||
json={"gallery_root_paths": ["output/renders"]},
|
||||
)
|
||||
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertEqual(
|
||||
[os.path.normcase(p) for p in GalleryConfig.MONITORING_DIRECTORIES],
|
||||
[os.path.normcase(os.path.realpath(renders))],
|
||||
)
|
||||
|
||||
async def test_diagnostics_paths_are_relative(self):
|
||||
GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir)]
|
||||
self._install_monitor([str(self.output_dir)])
|
||||
|
||||
resp, body = await self._body("GET", "/prompt_manager/diagnostics")
|
||||
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertNotIn(self.tmpdir, body)
|
||||
data = json.loads(body)
|
||||
self.assertEqual(
|
||||
data["diagnostics"]["comfyui_output"]["output_dirs"], ["output"]
|
||||
)
|
||||
self.assertEqual(
|
||||
data["diagnostics"]["image_monitor"]["monitored_directories"], ["output"]
|
||||
)
|
||||
|
||||
async def test_diagnostics_missing_db_message_has_no_path(self):
|
||||
self.api.db.model.db_path = os.path.join(self.tmpdir, "missing", "prompts.db")
|
||||
|
||||
resp, body = await self._body("GET", "/prompt_manager/diagnostics")
|
||||
|
||||
self.assertNotIn(self.tmpdir, body)
|
||||
data = json.loads(body)
|
||||
self.assertEqual(data["diagnostics"]["database"]["status"], "error")
|
||||
self.assertIn("prompts.db", data["diagnostics"]["database"]["message"])
|
||||
|
||||
async def test_scan_duplicates_paths_are_relative_to_output(self):
|
||||
(self.output_dir / "a.png").write_bytes(b"same-bytes")
|
||||
(self.output_dir / "sub").mkdir()
|
||||
(self.output_dir / "sub" / "b.png").write_bytes(b"same-bytes")
|
||||
GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir)]
|
||||
|
||||
resp, body = await self._body("GET", "/prompt_manager/scan_duplicates")
|
||||
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertNotIn(self.tmpdir, body)
|
||||
data = json.loads(body)
|
||||
self.assertEqual(len(data["duplicates"]), 1)
|
||||
paths = sorted(img["path"] for img in data["duplicates"][0]["images"])
|
||||
self.assertEqual(paths, ["a.png", os.path.join("sub", "b.png")])
|
||||
|
||||
async def test_delete_duplicates_accepts_relative_paths(self):
|
||||
target = self.output_dir / "sub" / "b.png"
|
||||
target.parent.mkdir()
|
||||
target.write_bytes(b"bytes")
|
||||
GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir)]
|
||||
|
||||
resp, body = await self._body(
|
||||
"POST",
|
||||
"/prompt_manager/delete_duplicate_images",
|
||||
json={"image_paths": [os.path.join("sub", "b.png")]},
|
||||
)
|
||||
|
||||
self.assertEqual(resp.status, 200)
|
||||
data = json.loads(body)
|
||||
self.assertEqual(data["deleted_count"], 1)
|
||||
self.assertFalse(target.exists())
|
||||
|
||||
async def test_scan_images_stream_has_no_absolute_paths(self):
|
||||
GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir)]
|
||||
|
||||
resp, body = await self._body("POST", "/prompt_manager/scan")
|
||||
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertNotIn(self.tmpdir, body)
|
||||
self.assertIn('"directories": ["output"]', body)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
"""Tests for the core API module (py/api/__init__.py): helpers and base routes."""
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import types
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
_mock_server = MagicMock()
|
||||
_mock_server.PromptServer.instance.routes = MagicMock()
|
||||
sys.modules["server"] = _mock_server
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from py.api import _public_error, _public_path # noqa: E402
|
||||
|
||||
EXTRA_ROOTS_ENV = "PROMPT_MANAGER_EXTRA_GALLERY_ROOTS"
|
||||
|
||||
|
||||
class FolderPathsFixture(unittest.TestCase):
|
||||
"""Temp ComfyUI tree exposed through a stub folder_paths module."""
|
||||
|
||||
def setUp(self):
|
||||
self.tmpdir = tempfile.mkdtemp()
|
||||
self.addCleanup(shutil.rmtree, self.tmpdir, True)
|
||||
self.comfy_dir = Path(self.tmpdir) / "ComfyUI"
|
||||
self.output_dir = self.comfy_dir / "output"
|
||||
self.output_dir.mkdir(parents=True)
|
||||
|
||||
self._orig_folder_paths = sys.modules.get("folder_paths")
|
||||
self.addCleanup(self._restore_folder_paths)
|
||||
self._orig_extra = os.environ.pop(EXTRA_ROOTS_ENV, None)
|
||||
self.addCleanup(self._restore_extra_env)
|
||||
|
||||
self.install_folder_paths(base_path=str(self.comfy_dir))
|
||||
|
||||
def install_folder_paths(self, **attrs):
|
||||
attrs.setdefault("get_output_directory", lambda: str(self.output_dir))
|
||||
sys.modules["folder_paths"] = types.SimpleNamespace(**attrs)
|
||||
|
||||
def _restore_folder_paths(self):
|
||||
if self._orig_folder_paths is None:
|
||||
sys.modules.pop("folder_paths", None)
|
||||
else:
|
||||
sys.modules["folder_paths"] = self._orig_folder_paths
|
||||
|
||||
def _restore_extra_env(self):
|
||||
if self._orig_extra is None:
|
||||
os.environ.pop(EXTRA_ROOTS_ENV, None)
|
||||
else:
|
||||
os.environ[EXTRA_ROOTS_ENV] = self._orig_extra
|
||||
|
||||
|
||||
class TestPublicPath(FolderPathsFixture):
|
||||
|
||||
def test_path_under_base_is_relative_posix(self):
|
||||
target = self.output_dir / "renders" / "a.png"
|
||||
self.assertEqual(_public_path(str(target)), "output/renders/a.png")
|
||||
|
||||
def test_base_itself_is_dot(self):
|
||||
self.assertEqual(_public_path(str(self.comfy_dir)), ".")
|
||||
|
||||
def test_path_outside_base_is_basename_only(self):
|
||||
secret = Path(self.tmpdir) / "secret" / "passwords.txt"
|
||||
secret.parent.mkdir()
|
||||
secret.write_text("x")
|
||||
self.assertEqual(_public_path(str(secret)), "passwords.txt")
|
||||
self.assertNotIn(self.tmpdir, _public_path(str(secret)))
|
||||
|
||||
def test_path_object_accepted(self):
|
||||
self.assertEqual(_public_path(self.output_dir), "output")
|
||||
|
||||
def test_falls_back_to_output_parent_without_base_path(self):
|
||||
self.install_folder_paths() # no base_path attribute
|
||||
self.assertEqual(_public_path(str(self.output_dir / "x.png")), "output/x.png")
|
||||
|
||||
def test_env_extra_root_keeps_its_own_name(self):
|
||||
extra = Path(self.tmpdir) / "nas" / "gallery"
|
||||
(extra / "sub").mkdir(parents=True)
|
||||
os.environ[EXTRA_ROOTS_ENV] = str(extra)
|
||||
self.assertEqual(_public_path(str(extra / "sub")), "gallery/sub")
|
||||
|
||||
def test_without_folder_paths_returns_basename(self):
|
||||
sys.modules.pop("folder_paths", None)
|
||||
self.assertEqual(_public_path(str(self.output_dir / "img.png")), "img.png")
|
||||
|
||||
def test_empty_or_none_gives_empty_string(self):
|
||||
self.assertEqual(_public_path(""), "")
|
||||
self.assertEqual(_public_path(None), "")
|
||||
|
||||
|
||||
class TestPublicError(FolderPathsFixture):
|
||||
|
||||
def test_oserror_uses_strerror_and_basename(self):
|
||||
err = FileNotFoundError(
|
||||
2, "No such file or directory", str(self.tmpdir) + "/a/b.png"
|
||||
)
|
||||
text = _public_error(err)
|
||||
self.assertIn("No such file or directory", text)
|
||||
self.assertIn("b.png", text)
|
||||
self.assertNotIn(self.tmpdir, text)
|
||||
|
||||
def test_oserror_without_filename(self):
|
||||
err = PermissionError(13, "Permission denied")
|
||||
self.assertEqual(_public_error(err), "Permission denied")
|
||||
|
||||
def test_oserror_without_strerror_uses_type_name(self):
|
||||
err = OSError()
|
||||
self.assertEqual(_public_error(err), "OSError")
|
||||
|
||||
def test_non_oserror_is_str(self):
|
||||
self.assertEqual(_public_error(ValueError("bad value")), "bad value")
|
||||
|
||||
def test_non_oserror_with_empty_message_uses_type_name(self):
|
||||
self.assertEqual(_public_error(RuntimeError()), "RuntimeError")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,5 +1,6 @@
|
||||
"""Route tests for the logging API (py/api/logging_routes.py)."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
@@ -192,5 +193,42 @@ class TestUpdateLogConfigWhitelist(LoggingAPITestCase):
|
||||
self.assertEqual(self.manager.update_calls, [])
|
||||
|
||||
|
||||
class TestNoAbsolutePathsInLogResponses(LoggingAPITestCase):
|
||||
|
||||
async def test_log_files_have_no_absolute_path(self):
|
||||
(self.log_dir / "prompt_manager.log").write_text("hello")
|
||||
|
||||
resp = await self.client.request("GET", "/prompt_manager/logs/files")
|
||||
|
||||
body = await resp.text()
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertNotIn(self.tmpdir, body)
|
||||
data = json.loads(body)
|
||||
self.assertEqual(data["files"][0]["filename"], "prompt_manager.log")
|
||||
|
||||
async def test_log_stats_directory_is_not_absolute(self):
|
||||
resp = await self.client.request("GET", "/prompt_manager/logs/stats")
|
||||
|
||||
body = await resp.text()
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertNotIn(self.tmpdir, body)
|
||||
self.assertEqual(json.loads(body)["stats"]["log_directory"], "logs")
|
||||
|
||||
async def test_error_from_manager_does_not_leak_path(self):
|
||||
def boom():
|
||||
raise FileNotFoundError(
|
||||
2, "No such file", os.path.join(self.tmpdir, "x.log")
|
||||
)
|
||||
|
||||
self.manager.get_log_stats = boom
|
||||
|
||||
resp = await self.client.request("GET", "/prompt_manager/logs/stats")
|
||||
|
||||
body = await resp.text()
|
||||
self.assertEqual(resp.status, 500)
|
||||
self.assertNotIn(self.tmpdir, body)
|
||||
self.assertIn("x.log", json.loads(body)["error"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -181,14 +181,22 @@ class TestValidateGalleryRootAccepts(GalleryRootTestCase):
|
||||
self.assertTrue(ok_a, reason_a)
|
||||
self.assertTrue(ok_b, reason_b)
|
||||
|
||||
def test_relative_path_is_resolved_before_checking(self):
|
||||
def test_relative_path_resolves_against_comfyui_base_not_cwd(self):
|
||||
sub = self.output_dir / "rel"
|
||||
sub.mkdir()
|
||||
orig_cwd = os.getcwd()
|
||||
os.chdir(str(self.output_dir))
|
||||
os.chdir(self.tmpdir)
|
||||
self.addCleanup(os.chdir, orig_cwd)
|
||||
ok, reason = GalleryConfig.validate_gallery_root("rel")
|
||||
|
||||
ok, reason = GalleryConfig.validate_gallery_root("output/rel")
|
||||
self.assertTrue(ok, reason)
|
||||
self.assertEqual(
|
||||
GalleryConfig.resolve_gallery_root("output/rel"),
|
||||
os.path.normcase(os.path.realpath(sub)),
|
||||
)
|
||||
# A path that only exists relative to the CWD is not accepted.
|
||||
(Path(self.tmpdir) / "rel").mkdir()
|
||||
self.assertFalse(GalleryConfig.validate_gallery_root("rel")[0])
|
||||
|
||||
def test_accepts_without_folder_paths_when_env_lists_dir(self):
|
||||
sys.modules.pop("folder_paths", None)
|
||||
|
||||
Reference in New Issue
Block a user