From e720aeb73a6866f5e1ce2f7db11b9865e47fb2d6 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Wed, 7 Oct 2026 19:50:20 -0700 Subject: [PATCH] fix: describe images relative to the ComfyUI tree in API responses Gallery and prompt responses carried absolute server paths: /images/output entries (path, root_dir), thumbnails_path, the images/prompt lookup, every image list (image_path plus its normalised file_path duplicate) and the preview images attached to prompt listings. They now go through _public_path: /images/output 'path' is relative to its root (what the delete and autotag endpoints resolve against), 'root_dir' and every other image_path are relative to the ComfyUI base, and file_path is dropped. recent/all/search image lists also gain url/thumbnail_url via the same enrichment the prompt-images route already had. Remaining str(e) error bodies in both modules use the basename-only error text. --- py/api/images.py | 98 +++++++++++++++++++-------- py/api/prompts.py | 136 ++++++++++++++++++++++++++++++-------- tests/test_api_images.py | 83 ++++++++++++++++++++++- tests/test_api_prompts.py | 53 +++++++++++++++ 4 files changed, 313 insertions(+), 57 deletions(-) diff --git a/py/api/images.py b/py/api/images.py index ebdfd87..bde28c1 100644 --- a/py/api/images.py +++ b/py/api/images.py @@ -12,7 +12,7 @@ from pathlib import Path from aiohttp import web from PIL import Image, UnidentifiedImageError -from .prompts import bad_request, parse_page_params +from .prompts import bad_request, parse_page_params, publish_image_paths from .prompts import safe_error_message as _safe_error # Only these file types are ever served by the image routes, regardless of @@ -34,6 +34,16 @@ IMAGE_EXTENSIONS = frozenset( ) +def _public_path(path): + """Response-safe form of *path* (relative to the ComfyUI tree). + + Imported lazily: the package defines it after importing this module. + """ + from . import _public_path as public_path + + return public_path(path) + + def _canonical(path): """Resolve symlinks and normalise case so paths compare on every OS.""" return Path(os.path.normcase(os.path.realpath(str(path)))) @@ -225,8 +235,14 @@ def _iter_media_files(output_path, max_files): yield root_path / name -def _output_image_entry(media_path, output_path, root_index): - """Build one gallery entry for *media_path* (blocking stat).""" +def _output_image_entry(media_path, output_path, root_index, public_root=None): + """Build one gallery entry for *media_path* (blocking stat). + + ``path`` is relative to its root (what delete/autotag resolve against) and + ``root_dir`` is the root's public form; neither reveals the server layout. + """ + if public_root is None: + public_root = _public_path(output_path) stat = media_path.stat() rel_path = media_path.relative_to(output_path) extension = media_path.suffix.lower() @@ -245,9 +261,9 @@ def _output_image_entry(media_path, output_path, root_index): return { "id": hashlib.sha1(str(media_path).encode("utf-8")).hexdigest()[:16], "filename": media_path.name, - "path": str(media_path), + "path": rel_path.as_posix(), "relative_path": str(rel_path), - "root_dir": str(output_path), + "root_dir": public_root, "root_index": root_index, "url": f"/prompt_manager/images/serve/{rel_path.as_posix()}", "thumbnail_url": thumbnail_url, @@ -325,6 +341,11 @@ class ImageRoutesMixin: async def get_gallery_subfolders_route(request): return await self.get_gallery_subfolders(request) + def _present_images(self, images): + """Image dicts ready for a response: urls added, server paths hidden.""" + self._enrich_images(images) + return publish_image_paths(images, _public_path) + async def get_prompt_images(self, request): """Get all images for a specific prompt.""" try: @@ -335,7 +356,7 @@ class ImageRoutesMixin: cleaned_images = [self._clean_nan_recursive(image) for image in images] # Add url and thumbnail_url so the frontend can serve them - self._enrich_images(cleaned_images) + cleaned_images = self._present_images(cleaned_images) # Additional fallback: convert to JSON string and clean NaN values manually try: @@ -361,7 +382,9 @@ class ImageRoutesMixin: except Exception as e: self.logger.error(f"Get prompt images error: {e}") - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": _safe_error(e)}, status=500 + ) async def get_recent_images(self, request): """Get recently generated images (bounded window, newest first).""" @@ -371,8 +394,8 @@ class ImageRoutesMixin: except ValueError: return bad_request("limit and offset must be integers") - images = await self._run_in_executor( - self.db.get_recent_images, limit, offset + images = self._present_images( + await self._run_in_executor(self.db.get_recent_images, limit, offset) ) return web.json_response( @@ -404,8 +427,10 @@ class ImageRoutesMixin: except ValueError: return bad_request("limit and offset must be integers") - images = await self._run_in_executor( - self.db.get_all_images, limit=limit, offset=offset + images = self._present_images( + await self._run_in_executor( + self.db.get_all_images, limit=limit, offset=offset + ) ) return web.json_response( @@ -441,8 +466,10 @@ class ImageRoutesMixin: except ValueError: return bad_request("limit and offset must be integers") - images = await self._run_in_executor( - self.db.search_images_by_prompt, query, limit, offset + images = self._present_images( + await self._run_in_executor( + self.db.search_images_by_prompt, query, limit, offset + ) ) return web.json_response( @@ -668,7 +695,9 @@ class ImageRoutesMixin: ) except Exception as e: self.logger.error(f"Subfolders error: {e}") - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": _safe_error(e)}, status=500 + ) # Fixed quality presets: clients choose a name, never a raw pixel size. THUMBNAIL_SIZES = {"low": (150, 150), "medium": (300, 300), "high": (600, 600)} @@ -722,7 +751,7 @@ class ImageRoutesMixin: ) payload = _thumbnail_complete_payload(len(targets), stats, start) payload["success"] = True - payload["thumbnails_path"] = str(thumbnails_dir) + payload["thumbnails_path"] = _public_path(thumbnails_dir) self.logger.info(payload["message"]) return payload @@ -1109,7 +1138,9 @@ class ImageRoutesMixin: except Exception as e: self.logger.error(f"Clear thumbnails error: {e}") - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": _safe_error(e)}, status=500 + ) async def link_image_to_prompt(self, request): """Link a generated image (inside an allowed directory) to a prompt.""" @@ -1175,17 +1206,15 @@ class ImageRoutesMixin: self.db.get_image_prompt_info, image_path ) if prompt_data: - prompt_data["image_path"] = image_path - if prompt_data: + prompt_data["image_path"] = _public_path(image_path) return web.json_response({"success": True, "prompt": prompt_data}) - else: - return web.json_response( - { - "success": False, - "error": "No prompt found for this image", - "image_path": image_path, - } - ) + return web.json_response( + { + "success": False, + "error": "No prompt found for this image", + "image_path": _public_path(image_path), + } + ) except Exception as db_error: self.logger.error(f"Database error in get_image_prompt: {db_error}") @@ -1195,7 +1224,9 @@ class ImageRoutesMixin: except Exception as e: self.logger.error(f"Get image prompt error: {e}") - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": _safe_error(e)}, status=500 + ) async def delete_image(self, request): """Delete an image record.""" @@ -1218,7 +1249,9 @@ class ImageRoutesMixin: ) except Exception as e: self.logger.error(f"Delete image error: {e}") - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": _safe_error(e)}, status=500 + ) async def _collect_output_images(self, output_dirs, subfolder): """Merge cached per-root listings, newest first, optionally filtered.""" @@ -1237,9 +1270,16 @@ class ImageRoutesMixin: def _format_output_page(self, page): entries = [] + public_roots = {} for media_path, _mtime, output_path, root_index in page: + if output_path not in public_roots: + public_roots[output_path] = _public_path(output_path) try: - entries.append(_output_image_entry(media_path, output_path, root_index)) + entries.append( + _output_image_entry( + media_path, output_path, root_index, public_roots[output_path] + ) + ) except Exception as e: self.logger.error(f"Error processing media {media_path.name}: {e}") return entries diff --git a/py/api/prompts.py b/py/api/prompts.py index f9ed615..0545651 100644 --- a/py/api/prompts.py +++ b/py/api/prompts.py @@ -42,6 +42,25 @@ def safe_error_message(exc): return str(exc) or exc.__class__.__name__ +def publish_image_paths(images, public_path): + """Image dicts for a response: ``image_path`` in its public (ComfyUI + relative) form and the absolute ``file_path`` duplicate removed. + + ``url``/``thumbnail_url`` added by enrichment are left untouched, so the + frontend keeps working without ever seeing the server's directory layout. + """ + published = [] + for image in images: + if not isinstance(image, dict): + published.append(image) + continue + image = {k: v for k, v in image.items() if k != "file_path"} + if "image_path" in image: + image["image_path"] = public_path(image["image_path"]) + published.append(image) + return published + + try: from ...utils.validators import ( validate_prompt_text, @@ -159,6 +178,15 @@ class PromptRoutesMixin: async def export_prompts_route(request): return await self.export_prompts(request) + def _present_prompts(self, prompts): + """Prompts ready for a response: image urls added, server paths hidden.""" + self._enrich_prompt_images(prompts) + for prompt in prompts: + prompt["images"] = publish_image_paths( + prompt.get("images", []), self._public_path + ) + return prompts + async def search_prompts(self, request): """Search for prompts using multiple filter criteria.""" try: @@ -195,7 +223,7 @@ class PromptRoutesMixin: folder=folder, sort=sort, ) - self._enrich_prompt_images(results) + self._present_prompts(results) return web.json_response( { @@ -213,7 +241,11 @@ class PromptRoutesMixin: except Exception as e: self.logger.error(f"Search error: {e}", exc_info=True) return web.json_response( - {"success": False, "error": f"Search failed: {str(e)}", "results": []}, + { + "success": False, + "error": f"Search failed: {safe_error_message(e)}", + "results": [], + }, status=500, ) @@ -232,7 +264,9 @@ class PromptRoutesMixin: return web.json_response({"success": True, "subfolders": subfolders}) except Exception as e: self.logger.error(f"Subfolders error: {e}") - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": safe_error_message(e)}, status=500 + ) async def get_recent_prompts(self, request): """Retrieve prompts with pagination and an optional sort (default newest @@ -252,7 +286,7 @@ class PromptRoutesMixin: results = await self._run_in_executor( self.db.get_recent_prompts, limit=limit, offset=offset, sort=sort ) - self._enrich_prompt_images(results["prompts"]) + self._present_prompts(results["prompts"]) return web.json_response( { @@ -275,7 +309,7 @@ class PromptRoutesMixin: return web.json_response( { "success": False, - "error": f"Failed to get recent prompts: {str(e)}", + "error": f"Failed to get recent prompts: {safe_error_message(e)}", "results": [], "pagination": {"total": 0, "page": 1, "total_pages": 0}, }, @@ -292,7 +326,7 @@ class PromptRoutesMixin: return web.json_response( { "success": False, - "error": f"Failed to get categories: {str(e)}", + "error": f"Failed to get categories: {safe_error_message(e)}", "categories": [], }, status=500, @@ -308,7 +342,7 @@ class PromptRoutesMixin: return web.json_response( { "success": False, - "error": f"Failed to get tags: {str(e)}", + "error": f"Failed to get tags: {safe_error_message(e)}", "tags": [], }, status=500, @@ -346,7 +380,9 @@ class PromptRoutesMixin: ) except Exception as e: self.logger.error(f"Tags stats error: {e}", exc_info=True) - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": safe_error_message(e)}, status=500 + ) async def get_tag_prompts(self, request): """Get prompts for a single tag.""" @@ -367,7 +403,7 @@ class PromptRoutesMixin: result = await self._run_in_executor( self.db.get_prompts_by_tags, [tag_name], "and", limit, offset ) - self._enrich_prompt_images(result["prompts"]) + self._present_prompts(result["prompts"]) return web.json_response( { @@ -384,7 +420,9 @@ class PromptRoutesMixin: ) except Exception as e: self.logger.error(f"Tag prompts error: {e}", exc_info=True) - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": safe_error_message(e)}, status=500 + ) async def get_tags_filter(self, request): """Get prompts matching multiple tags with AND/OR mode, or untagged prompts.""" @@ -399,7 +437,7 @@ class PromptRoutesMixin: result = await self._run_in_executor( self.db.get_untagged_prompts, limit, offset ) - self._enrich_prompt_images(result["prompts"]) + self._present_prompts(result["prompts"]) return web.json_response( { "success": True, @@ -434,7 +472,7 @@ class PromptRoutesMixin: result = await self._run_in_executor( self.db.get_prompts_by_tags, tags_list, mode, limit, offset ) - self._enrich_prompt_images(result["prompts"]) + self._present_prompts(result["prompts"]) return web.json_response( { @@ -452,7 +490,9 @@ class PromptRoutesMixin: ) except Exception as e: self.logger.error(f"Tags filter error: {e}", exc_info=True) - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": safe_error_message(e)}, status=500 + ) async def rename_tag_endpoint(self, request): """Rename a tag across all prompts.""" @@ -495,7 +535,9 @@ class PromptRoutesMixin: return web.json_response(resp) except Exception as e: self.logger.error(f"Rename tag error: {e}", exc_info=True) - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": safe_error_message(e)}, status=500 + ) async def delete_tag_endpoint(self, request): """Delete a tag from all prompts.""" @@ -525,7 +567,9 @@ class PromptRoutesMixin: return web.json_response(resp) except Exception as e: self.logger.error(f"Delete tag error: {e}", exc_info=True) - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": safe_error_message(e)}, status=500 + ) async def merge_tags_endpoint(self, request): """Merge source tags into a target tag.""" @@ -566,7 +610,9 @@ class PromptRoutesMixin: return web.json_response(resp) except Exception as e: self.logger.error(f"Merge tags error: {e}", exc_info=True) - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": safe_error_message(e)}, status=500 + ) async def save_prompt(self, request): """Save a new prompt with metadata and duplicate detection.""" @@ -641,7 +687,10 @@ class PromptRoutesMixin: except Exception as e: self.logger.error(f"Save error: {e}", exc_info=True) return web.json_response( - {"success": False, "error": f"Failed to save prompt: {str(e)}"}, + { + "success": False, + "error": f"Failed to save prompt: {safe_error_message(e)}", + }, status=500, ) @@ -671,7 +720,10 @@ class PromptRoutesMixin: except Exception as e: self.logger.error(f"Delete error: {e}") return web.json_response( - {"success": False, "error": f"Failed to delete prompt: {str(e)}"}, + { + "success": False, + "error": f"Failed to delete prompt: {safe_error_message(e)}", + }, status=500, ) @@ -715,7 +767,10 @@ class PromptRoutesMixin: except Exception as e: self.logger.error(f"Update prompt error: {e}") return web.json_response( - {"success": False, "error": f"Failed to update prompt: {str(e)}"}, + { + "success": False, + "error": f"Failed to update prompt: {safe_error_message(e)}", + }, status=500, ) @@ -752,7 +807,10 @@ class PromptRoutesMixin: except Exception as e: self.logger.error(f"Update rating error: {e}") return web.json_response( - {"success": False, "error": f"Failed to update rating: {str(e)}"}, + { + "success": False, + "error": f"Failed to update rating: {safe_error_message(e)}", + }, status=500, ) @@ -795,7 +853,11 @@ class PromptRoutesMixin: except Exception as e: self.logger.error(f"Add tag error: {e}") return web.json_response( - {"success": False, "error": f"Failed to add tag: {str(e)}"}, status=500 + { + "success": False, + "error": f"Failed to add tag: {safe_error_message(e)}", + }, + status=500, ) async def add_tags_to_prompt(self, request): @@ -853,7 +915,11 @@ class PromptRoutesMixin: except Exception as e: self.logger.error(f"Add tags error: {e}") return web.json_response( - {"success": False, "error": f"Failed to add tags: {str(e)}"}, status=500 + { + "success": False, + "error": f"Failed to add tags: {safe_error_message(e)}", + }, + status=500, ) async def remove_prompt_tag(self, request): @@ -890,7 +956,10 @@ class PromptRoutesMixin: except Exception as e: self.logger.error(f"Remove tag error: {e}") return web.json_response( - {"success": False, "error": f"Failed to remove tag: {str(e)}"}, + { + "success": False, + "error": f"Failed to remove tag: {safe_error_message(e)}", + }, status=500, ) @@ -920,7 +989,10 @@ class PromptRoutesMixin: except Exception as e: self.logger.error(f"Bulk delete error: {e}") return web.json_response( - {"success": False, "error": f"Failed to delete prompts: {str(e)}"}, + { + "success": False, + "error": f"Failed to delete prompts: {safe_error_message(e)}", + }, status=500, ) @@ -952,7 +1024,11 @@ class PromptRoutesMixin: except Exception as e: self.logger.error(f"Bulk add tags error: {e}") return web.json_response( - {"success": False, "error": f"Failed to add tags: {str(e)}"}, status=500 + { + "success": False, + "error": f"Failed to add tags: {safe_error_message(e)}", + }, + status=500, ) async def bulk_set_category(self, request): @@ -982,7 +1058,10 @@ class PromptRoutesMixin: except Exception as e: self.logger.error(f"Bulk set category error: {e}") return web.json_response( - {"success": False, "error": f"Failed to set category: {str(e)}"}, + { + "success": False, + "error": f"Failed to set category: {safe_error_message(e)}", + }, status=500, ) @@ -1013,6 +1092,9 @@ class PromptRoutesMixin: except Exception as e: self.logger.error(f"Export error: {e}") return web.json_response( - {"success": False, "error": f"Failed to export prompts: {str(e)}"}, + { + "success": False, + "error": f"Failed to export prompts: {safe_error_message(e)}", + }, status=500, ) diff --git a/tests/test_api_images.py b/tests/test_api_images.py index 48f0796..db0d598 100644 --- a/tests/test_api_images.py +++ b/tests/test_api_images.py @@ -1265,7 +1265,7 @@ class TestImagePromptLookup(ImageRouteCoverageCase): self.assertEqual(status, 200) self.assertTrue(data["success"]) self.assertEqual(data["prompt"]["text"], "the prompt") - self.assertEqual(data["prompt"]["image_path"], str(self.output_dir / "gen.png")) + self.assertEqual(data["prompt"]["image_path"], "output/gen.png") async def test_unknown_image_reports_no_prompt(self): status, data = await self._json("GET", f"{self.URL}unknown.png") @@ -1289,6 +1289,87 @@ class TestImagePromptLookup(ImageRouteCoverageCase): self.assertEqual(status, 500) +class TestPublicPathsInResponses(ImageAPITestCase): + """Image routes describe files relative to the ComfyUI tree, never absolutely.""" + + async def setUpAsync(self): + await super().setUpAsync() + self.png = make_png(self.output_dir / "sub" / "gen.png") + self.prompt_id = self._save_prompt("public paths") + self.assertGreater( + self.api.db.link_image_to_prompt(self.prompt_id, str(self.png)), 0 + ) + + async def _body(self, method, path, **kwargs): + resp = await self.client.request(method, path, **kwargs) + text = await resp.text() + self.assertNotIn(str(self.comfy_dir), text, path) + return resp.status, json.loads(text) + + async def test_output_listing_paths_are_root_relative(self): + status, data = await self._body("GET", "/prompt_manager/images/output") + + self.assertEqual(status, 200) + entry = data["images"][0] + self.assertEqual(entry["path"], "sub/gen.png") + self.assertEqual(entry["root_dir"], "output") + self.assertEqual(entry["root_index"], 0) + + async def test_image_lists_publish_image_path_and_drop_file_path(self): + routes = ( + f"/prompt_manager/prompts/{self.prompt_id}/images", + "/prompt_manager/images/recent", + "/prompt_manager/images/all", + "/prompt_manager/images/search?q=public", + ) + for route in routes: + status, data = await self._body("GET", route) + + self.assertEqual(status, 200, route) + image = data["images"][0] + self.assertEqual(image["image_path"], "output/sub/gen.png", route) + self.assertNotIn("file_path", image, route) + self.assertTrue(image["url"].endswith("sub/gen.png"), route) + + async def test_image_prompt_lookup_publishes_path(self): + _, found = await self._body("GET", "/prompt_manager/images/prompt/sub/gen.png") + _, missing = await self._body("GET", "/prompt_manager/images/prompt/sub/no.png") + + self.assertEqual(found["prompt"]["image_path"], "output/sub/gen.png") + self.assertEqual(missing["image_path"], "output/sub/no.png") + + async def test_thumbnails_path_is_public(self): + status, data = await self._body( + "POST", "/prompt_manager/images/generate-thumbnails", json={} + ) + + self.assertEqual(status, 200) + self.assertEqual(data["thumbnails_path"], "output/thumbnails") + + async def test_oserror_bodies_keep_only_the_basename(self): + secret = str(self.comfy_dir / "hidden.png") + + def boom(*_args, **_kwargs): + raise FileNotFoundError(2, "No such file", secret) + + self.api._get_all_output_dirs = boom + self.api._find_comfyui_output_dir = boom + self.api.db.get_prompt_images = boom + self.api.db.delete_image = boom + routes = ( + ("GET", "/prompt_manager/gallery/subfolders"), + ("GET", "/prompt_manager/prompts/1/images"), + ("DELETE", "/prompt_manager/images/1"), + ("POST", "/prompt_manager/images/clear-thumbnails"), + ("GET", "/prompt_manager/images/prompt/x.png"), + ) + for method, route in routes: + status, data = await self._body(method, route) + + self.assertEqual(status, 500, route) + self.assertIn("hidden.png", data["error"], route) + + class TestDeleteAndLinkErrors(ImageRouteCoverageCase): async def test_delete_image_paths(self): diff --git a/tests/test_api_prompts.py b/tests/test_api_prompts.py index 0b0e43d..c45716a 100644 --- a/tests/test_api_prompts.py +++ b/tests/test_api_prompts.py @@ -2,6 +2,7 @@ Route tests for the prompt API (py/api/prompts.py). """ +import json import os import sys import tempfile @@ -378,6 +379,58 @@ class TestTagRoutePagingBounds(RouteCoverageCase): self.assertFalse(data["success"], route) +class TestPublicPathsInPromptResponses(RouteCoverageCase): + """Preview images on prompt listings never expose the server's directories.""" + + SECRET_DIR = os.path.join(os.sep, "srv", "secret") + + async def setUpAsync(self): + await super().setUpAsync() + self.pid = self._save_prompt("has preview", tags=["sky"]) + image_path = os.path.join(self.SECRET_DIR, "gen.png") + self.assertGreater(self.api.db.link_image_to_prompt(self.pid, image_path), 0) + + async def _body(self, method, path, **kwargs): + resp = await self.client.request(method, path, **kwargs) + text = await resp.text() + self.assertNotIn(self.SECRET_DIR, text, path) + return resp.status, json.loads(text) + + async def test_listing_routes_publish_preview_image_paths(self): + routes = { + "/prompt_manager/recent": "results", + "/prompt_manager/search?q=preview": "results", + "/prompt_manager/tags/sky/prompts": "prompts", + "/prompt_manager/tags/filter?tags=sky": "prompts", + } + for route, key in routes.items(): + status, data = await self._body("GET", route) + + self.assertEqual(status, 200, route) + image = data[key][0]["images"][0] + # No ComfyUI anchor is configured here, so only the name survives. + self.assertEqual(image["image_path"], "gen.png", route) + self.assertNotIn("file_path", image, route) + + async def test_oserror_bodies_keep_only_the_basename(self): + def boom(*_args, **_kwargs): + raise FileNotFoundError( + 2, "No such file", os.path.join(self.SECRET_DIR, "prompts.db") + ) + + for db_method, route in ( + ("get_tags_with_counts", "/prompt_manager/tags/stats"), + ("search_prompts", "/prompt_manager/search"), + ("get_recent_prompts", "/prompt_manager/recent"), + ("get_prompts_by_tags", "/prompt_manager/tags/sky/prompts"), + ): + setattr(self.api.db, db_method, boom) + status, data = await self._body("GET", route) + + self.assertEqual(status, 500, route) + self.assertIn("prompts.db", data["error"], route) + + class TestTagMutationRoutes(RouteCoverageCase): async def test_rename_rejects_bad_json_and_empty_name(self):