diff --git a/database/health.py b/database/health.py new file mode 100644 index 00000000..1149773e --- /dev/null +++ b/database/health.py @@ -0,0 +1,265 @@ +"""The read-only health check behind System › Overview and Maintenance. + +``run_health_checks`` opens its own ``mode=ro`` connection, reads one +snapshot and closes the connection in ``finally``; it never writes. Each +check reports ``passed``, ``issues`` (with an exact ``count``), +``unavailable`` (it could not be checked here, so ``count`` is None) or +``failed`` (it raised). At most ``MAX_DETAIL`` detail strings are returned per +check, and none of them carries a server path. +""" + +import datetime +import logging +import os +import sqlite3 +from pathlib import Path + +try: + from ..utils.hashing import generate_prompt_hash + from .models import normalize_image_path +except ImportError: + from utils.hashing import generate_prompt_hash + from database.models import normalize_image_path + +logger = logging.getLogger("prompt_manager.database.health") + +MAX_DETAIL = 10 # detail strings returned per check +INTEGRITY_LIMIT = 100 # PRAGMA integrity_check(N): problems SQLite reports +BATCH_ROWS = 100 # image rows read per fetch +REVIEW_UNAVAILABLE = "Needs review is not available yet." +CHECK_FAILED = "The check could not finish." +# Check ids and the labels of the concept's health report, in its order; +# pair uniqueness and consistency are the added rows (D63, now D95). +CHECKS = ( + ("integrity", "Integrity check"), + ("foreign_keys", "Foreign keys & pair references"), + ("canonical_hashes", "Canonical hashes"), + ("pair_uniqueness", "Pair uniqueness"), + ("missing_files", "Missing files"), + ("ambiguous_metadata", "Ambiguous metadata"), + ("consistency", "Consistency"), +) +_DATABASE_CHECKS = ( + "integrity", + "foreign_keys", + "canonical_hashes", + "pair_uniqueness", + "missing_files", + "consistency", +) + + +def iter_consistency_issues(conn): + """The legacy consistency issues: tag and image rows of a missing prompt.""" + for ref in conn.execute( + "SELECT pt.prompt_id FROM prompt_tags pt " + "LEFT JOIN prompts p ON pt.prompt_id = p.id WHERE p.id IS NULL" + ): + yield f"prompt_tags entry references non-existent prompt {ref[0]}" + for ref in conn.execute( + "SELECT gi.id, gi.prompt_id FROM generated_images gi " + "LEFT JOIN prompts p ON gi.prompt_id = p.id WHERE p.id IS NULL" + ): + yield f"Image {ref[0]} references non-existent prompt {ref[1]}" + + +def _result(count, detail, *, status=None): + return { + "status": status or ("issues" if count else "passed"), + "count": count, + "detail": detail[:MAX_DETAIL], + } + + +def _counted(lines): + """(count, first MAX_DETAIL lines) of an iterable, consumed once.""" + count, detail = 0, [] + for line in lines: + count += 1 + if len(detail) < MAX_DETAIL: + detail.append(line) + return _result(count, detail) + + +def _integrity(conn, _roots): + rows = [ + row[0] for row in conn.execute(f"PRAGMA integrity_check({INTEGRITY_LIMIT})") + ] + if rows == ["ok"]: + return _result(0, []) + return _result(len(rows), rows) + + +def _foreign_keys(conn, _roots): + return _counted( + f"{table} row {rowid} references a missing {parent} row" + for table, rowid, parent, _fkid in conn.execute("PRAGMA foreign_key_check") + ) + + +def _prompt_hash(text): + return generate_prompt_hash(text) if isinstance(text, str) else None + + +def _canonical_hashes(conn, _roots): + conn.create_function("pm_prompt_hash", 1, _prompt_hash, deterministic=True) + return _counted( + f"Prompt {row[0]}: the stored hash does not match its text" + for row in conn.execute( + "SELECT id FROM prompts WHERE hash IS NOT pm_prompt_hash(text) ORDER BY id" + ) + ) + + +def _pair_uniqueness(conn, _roots): + groups = conn.execute( + "SELECT positive_prompt_id,negative_prompt_id,COUNT(*) FROM prompt_pairs " + "GROUP BY positive_prompt_id,negative_prompt_id HAVING COUNT(*)>1 " + "ORDER BY positive_prompt_id,negative_prompt_id" + ).fetchall() + detail = [ + f"Positive prompt {positive} with " + + ("no negative" if negative is None else f"negative {negative}") + + f": {rows} pair rows" + for positive, negative, rows in groups + ] + # The surplus rows: what keeping one row per pair would leave out. + return _result(sum(rows - 1 for _p, _n, rows in groups), detail) + + +def _consistency(conn, _roots): + return _counted(iter_consistency_issues(conn)) + + +def _readable(root): + return os.path.isdir(root) and os.access(root, os.R_OK | os.X_OK) + + +def _root_prefixes(roots): + return [normalize_image_path(root).rstrip("\\/") + os.sep for root in roots] + + +def _under(path, prefixes): + return any(path.startswith(prefix) for prefix in prefixes) + + +def _missing_files(conn, roots): + """Links whose file is gone, counted only under readable sources. + + A source that is not readable (an unmounted drive) makes the whole check + unavailable rather than turning its links into a count; links outside + every source are named in the detail, not counted. + """ + if not roots: + return _result(None, ["No image source is available."], status="unavailable") + unavailable = [root for root in roots if not _readable(root)] + if unavailable: + noun = "source is" if len(unavailable) == 1 else "sources are" + return _result( + None, + [ + f"{len(unavailable)} image {noun} unavailable, " + "so missing files were not counted." + ], + status="unavailable", + ) + prefixes = _root_prefixes(roots) + count, outside, detail = 0, 0, [] + cursor = conn.execute( + "SELECT id,image_path,filename,file_path FROM generated_images ORDER BY id" + ) + while rows := cursor.fetchmany(BATCH_ROWS): + for image_id, image_path, filename, file_path in rows: + spelled = file_path or normalize_image_path(image_path) + if not _under(spelled, prefixes) and not _under( + normalize_image_path(os.path.realpath(image_path)), prefixes + ): + outside += 1 + elif not os.path.exists(image_path): + count += 1 + if len(detail) < MAX_DETAIL: + name = filename or os.path.basename(image_path) + detail.append(f"Image {image_id}: {name} is missing") + result = _result(count, detail) + if outside and len(result["detail"]) < MAX_DETAIL: + links = "link" if outside == 1 else "links" + verb = "was" if outside == 1 else "were" + result["detail"].append( + f"{outside} {links} outside the image sources {verb} not checked." + ) + return result + + +def _ambiguous(review_count): + if review_count is None: + return _result(None, [REVIEW_UNAVAILABLE], status="unavailable") + return _result(review_count, []) + + +def overall_status(checks) -> str: + """``passed`` only when every check passed; any check not run is incomplete.""" + statuses = {check["status"] for check in checks} + if statuses & {"unavailable", "failed"}: + return "incomplete" + return "issues" if "issues" in statuses else "passed" + + +def _failed(): + return _result(None, [CHECK_FAILED], status="failed") + + +def _run_database_checks(db_path, roots): + """Each database check's result (None when it raised), from one snapshot.""" + results = dict.fromkeys(_DATABASE_CHECKS) + try: + conn = sqlite3.connect(Path(db_path).as_uri() + "?mode=ro", uri=True) + except sqlite3.Error: + logger.exception("Health check could not open the database") + return results + try: + conn.execute("BEGIN") # one read snapshot for every check + for check_id in _DATABASE_CHECKS: + runner = globals()["_" + check_id] # looked up per run (tests patch it) + try: + results[check_id] = runner(conn, roots) + except (sqlite3.Error, OSError): + logger.exception(f"Health check {check_id} failed") + except sqlite3.Error: + logger.exception("Health check could not read the database") + finally: + conn.close() + return results + + +class HealthOperations: + """Mixin for PromptDatabase: the read-only System health check.""" + + def run_health_checks(self, roots, *, review_count=None) -> dict: + """Run every check and return a ``HealthReport``. + + Args: + roots: The image source directories to judge missing files under: + the effective sources plus any configured source that is not + available (which makes ``missing_files`` unavailable). + review_count: The Needs-review pending count, or None while no + review queue exists (``ambiguous_metadata`` is then unavailable). + """ + if review_count is not None and ( + type(review_count) is not int or review_count < 0 + ): + raise ValueError("review_count must be a non-negative integer or None") + roots = [str(root) for root in roots] + database = _run_database_checks(self.model.db_path, roots) + checks = [] + for check_id, label in CHECKS: + if check_id == "ambiguous_metadata": + result = _ambiguous(review_count) + else: + result = database[check_id] or _failed() + checks.append({"id": check_id, "label": label, **result}) + return { + "success": True, + "checked_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), + "status": overall_status(checks), + "checks": checks, + } diff --git a/database/library.py b/database/library.py index 4e532b95..65727f8e 100644 --- a/database/library.py +++ b/database/library.py @@ -628,3 +628,49 @@ class LibraryOperations: "kind": "prompt-manager-library", "positives": [owners[key] for key in order], } + + def library_counts(self) -> dict: + """System › Overview counts from one snapshot, writing nothing. + + ``positive_prompts`` and ``negative_variants`` are the unfiltered + Library's ``total`` and ``variant_total``: eligible parents and active + pairs (every pair's positive is eligible). ``images`` counts indexed + ``generated_images`` rows, not files on disk. + """ + with self.library_snapshot() as (conn, _epoch, _ceiling): + row = conn.execute( + "SELECT (SELECT COUNT(*) FROM prompts p WHERE " + _ELIGIBLE + ")," + "(SELECT COUNT(*) FROM prompt_pairs WHERE archived=0)," + "(SELECT COUNT(*) FROM generated_images)" + ).fetchone() + return { + "positive_prompts": row[0], + "negative_variants": row[1], + "images": row[2], + } + + def usage_by_category(self) -> list[dict]: + """Runs and average rating of the Library's parents, per category. + + Only eligible parents count, so a negative-only prompt's runs (which + repeat its pairs' runs) are never added twice. A blank category groups + with no category as ``None``. Unrated parents are left out of the + average, which is ``None`` when no parent is rated. Most runs first. + """ + with self.library_snapshot() as (conn, _epoch, _ceiling): + rows = conn.execute( + "SELECT * FROM (SELECT NULLIF(TRIM(p.category),'') AS category," + "COALESCE(SUM(p.run_count),0) AS runs,AVG(p.rating) AS rating " + "FROM prompts p WHERE " + _ELIGIBLE + " GROUP BY 1) " + "ORDER BY runs DESC,category IS NULL,category" + ).fetchall() + return [ + { + "category": row["category"], + "runs": row["runs"], + "average_rating": ( + None if row["rating"] is None else round(row["rating"], 2) + ), + } + for row in rows + ] diff --git a/database/operations.py b/database/operations.py index 28bf35db..8424e880 100644 --- a/database/operations.py +++ b/database/operations.py @@ -20,6 +20,7 @@ from .execution_history import ExecutionHistoryOperations from .execution_contexts import ExecutionContextOperations, LIVE_HINT_CTES from .execution_outputs import ExecutionOutputOperations from .history_reads import HistoryReadOperations +from .health import HealthOperations, iter_consistency_issues # Import logging system try: @@ -342,6 +343,7 @@ class PromptDatabase( ImageFavoriteOperations, AnnotationOperations, SavedViewOperations, + HealthOperations, ): """Database operations class for managing prompts.""" @@ -2570,33 +2572,13 @@ class PromptDatabase( return cursor.rowcount def check_consistency(self) -> List[str]: - """Run consistency checks and return a list of issue descriptions.""" - issues: List[str] = [] - with self.model.get_connection() as conn: - # Check for orphaned prompt_tags entries - cursor = conn.execute(""" - SELECT pt.prompt_id, pt.tag_id FROM prompt_tags pt - LEFT JOIN prompts p ON pt.prompt_id = p.id - WHERE p.id IS NULL - """) - for ref in cursor.fetchall(): - issues.append( - "prompt_tags entry references non-existent prompt " - f"{ref['prompt_id']}" - ) + """Run consistency checks and return a list of issue descriptions. - # Check for orphaned image entries - cursor = conn.execute(""" - SELECT gi.id, gi.prompt_id FROM generated_images gi - LEFT JOIN prompts p ON gi.prompt_id = p.id - WHERE p.id IS NULL - """) - for ref in cursor.fetchall(): - issues.append( - f"Image {ref['id']} references non-existent prompt " - f"{ref['prompt_id']}" - ) - return issues + The same queries back the System health check's Consistency row + (database/health.py). + """ + with self.model.get_connection() as conn: + return list(iter_consistency_issues(conn)) def _image_row_to_dict(self, row: sqlite3.Row) -> Dict[str, Any]: """ diff --git a/py/api/__init__.py b/py/api/__init__.py index ad5fb7a6..6181a0d9 100644 --- a/py/api/__init__.py +++ b/py/api/__init__.py @@ -8,6 +8,7 @@ Split into domain-specific mixins for maintainability: - AutotagRoutesMixin: auto-tagging model management and tagging - JobRoutesMixin: the session job registry behind Activity (py/api/jobs.py) - ImageTagJobsMixin: bulk Add tags / Auto-tag as image-tags jobs +- SystemRoutesMixin: System health check, Overview and usage statistics """ import asyncio @@ -35,6 +36,7 @@ from .admin import AdminRoutesMixin from .settings_routes import SettingsRoutesMixin from .job_routes import JobRoutesMixin from .image_tag_jobs import ImageTagJobsMixin +from .system import SystemRoutesMixin from .jobs import JobRegistry from .logging_routes import LoggingRoutesMixin from .autotag_routes import AutotagRoutesMixin @@ -265,6 +267,7 @@ class PromptManagerAPI( SettingsRoutesMixin, JobRoutesMixin, ImageTagJobsMixin, + SystemRoutesMixin, LoggingRoutesMixin, AutotagRoutesMixin, LoraIntegrationMixin, @@ -321,6 +324,8 @@ class PromptManagerAPI( self._scan_job = None # The current duplicate scan (a py/api/jobs.Job outside the registry). self._duplicate_job = None + # The last System health report this session (py/api/system.py). + self._health_report = None # Run cleanup on initialization to remove any existing duplicates try: @@ -478,6 +483,7 @@ class PromptManagerAPI( self._register_admin_routes(routes) self._register_settings_routes(routes) self._register_job_routes(routes) + self._register_system_routes(routes) self._register_logging_routes(routes) self._register_autotag_routes(routes) self._register_lora_routes(routes) @@ -719,6 +725,13 @@ class PromptManagerAPI( Returns: List[Path]: Validated directory paths, possibly empty. """ + return self._output_dirs_with_origin()[0] + + def _output_dirs_with_origin(self): + """``(dirs, auto_detected)``: the ``_get_all_output_dirs`` list, and + whether it comes from detection or ComfyUI's output rather than from + a valid configured root (False when there is no directory at all). + """ from ..config import GalleryConfig configured = list(GalleryConfig.MONITORING_DIRECTORIES) @@ -737,15 +750,16 @@ class PromptManagerAPI( output_dirs.append(Path(real)) if output_dirs: - return output_dirs + return output_dirs, False if configured: # Every configured root was rejected: never let auto-detection # resurrect it, serve ComfyUI's own output directory instead. - return self._comfyui_output_dir_list() - - fallback = self._find_comfyui_output_dir() - return [Path(fallback)] if fallback else [] + fallback_dirs = self._comfyui_output_dir_list() + else: + fallback = self._find_comfyui_output_dir() + fallback_dirs = [Path(fallback)] if fallback else [] + return fallback_dirs, bool(fallback_dirs) def _comfyui_output_dir_list(self): """ComfyUI's output directory as a one-element list, or [] if unknown.""" diff --git a/py/api/jobs.py b/py/api/jobs.py index 568a99e6..09ab15aa 100644 --- a/py/api/jobs.py +++ b/py/api/jobs.py @@ -88,6 +88,9 @@ DUPLICATE_CLEANUP_JOB_KIND = "duplicate_cleanup" THUMBNAILS_JOB_KIND = "thumbnails" LORA_IMPORT_JOB_KIND = "lora_import" AUTOTAG_BATCH_JOB_KIND = "autotag" +# The System health check (py/api/system.py): a read-only ``read`` job with +# no file outcomes, so it may run beside a scan. +HEALTH_JOB_KIND = "health" def resolve_admission(admission=None, record_changing=None) -> str: diff --git a/py/api/system.py b/py/api/system.py new file mode 100644 index 00000000..3447824e --- /dev/null +++ b/py/api/system.py @@ -0,0 +1,177 @@ +"""System routes: the read-only health check, the Overview and usage statistics. + +- ``POST /prompt_manager/system/health`` starts the health check + (``database/health.py``) off-loop as a ``read`` job of kind ``health``: one + at a time, beside a scan, not cancellable and with no file outcomes. The + API keeps the last report for this session; ``GET`` returns it or null. +- ``GET /prompt_manager/system/overview`` is the System › Overview aggregate. + Every value is real or null: there is no backup status until backups report + one, no health result until a check ran, and no Needs-review count until + the review queue exists. +- ``GET /prompt_manager/system/statistics`` lists runs and ratings per + category, with the legacy ``/stats`` totals. + +New routes answer errors with ``{success:false, code, error}``. +""" + +import asyncio +import os + +from aiohttp import web + +from .job_routes import busy_response +from .jobs import HEALTH_JOB_KIND, JobBusy + +HEALTH_JOB_TITLE = "Health check" +_HEALTH_STATUS_LINES = { + "passed": "Health check passed", + "issues": "Health check found issues", + "incomplete": "Health check incomplete", +} + + +def _failure(message, status=500, code=None): + return web.json_response( + {"success": False, "code": code, "error": message}, status=status + ) + + +def health_summary(report): + """What the Overview shows of a report: when, the outcome, checks with issues.""" + if report is None: + return None + return { + "checked_at": report["checked_at"], + "status": report["status"], + "issues": sum(1 for check in report["checks"] if check["status"] == "issues"), + } + + +class SystemRoutesMixin: + """Mixin providing the /prompt_manager/system endpoints.""" + + def _register_system_routes(self, routes): + @routes.post("/prompt_manager/system/health") + async def start_health_check_route(request): + return await self.start_health_check(request) + + @routes.get("/prompt_manager/system/health") + async def health_report_route(request): + return await self.get_health_report(request) + + @routes.get("/prompt_manager/system/overview") + async def system_overview_route(request): + return await self.get_system_overview(request) + + @routes.get("/prompt_manager/system/statistics") + async def system_statistics_route(request): + return await self.get_system_statistics(request) + + # ── health ──────────────────────────────────────────────────────── + + def _health_roots(self): + """The sources missing files are judged under. + + The effective sources, plus every configured source that is not an + existing directory (an unmounted drive): the check then reports + missing files as unavailable instead of counting that source's links. + """ + from ..config import GalleryConfig + + roots = [str(path) for path in self._get_all_output_dirs()] + for configured in GalleryConfig.MONITORING_DIRECTORIES: + try: + resolved = GalleryConfig.resolve_gallery_root(configured) + except (OSError, TypeError, ValueError): + continue + if not os.path.isdir(resolved): + roots.append(resolved) + return roots + + async def start_health_check(self, request): + """Start the health check job (202), or 409 naming the running check.""" + roots = self._health_roots() + + async def run(job): + job.update(status_line="Checking the database and image links") + report = await self._run_in_executor(self.db.run_health_checks, roots) + self._health_report = report + job.update(progress=100, status_line=_HEALTH_STATUS_LINES[report["status"]]) + + try: + job = self._jobs.start( + HEALTH_JOB_KIND, + HEALTH_JOB_TITLE, + admission="read", + cancellable=False, + scope={}, + runner=run, + ) + except JobBusy as busy: + return busy_response(busy) + return web.json_response({"success": True, "job": job.summary()}, status=202) + + async def get_health_report(self, request): + return web.json_response({"success": True, "report": self._health_report}) + + # ── overview ────────────────────────────────────────────────────── + + def _missing_thumbnails(self, output_dirs): + """Pending thumbnails as ``/images/thumbnails/status`` counts them.""" + if not output_dirs: + return None + try: + plan = self._plan_gallery_thumbnails(output_dirs) + except OSError: + self.logger.warning("Overview could not plan thumbnails", exc_info=True) + return None + return len(plan["targets"]) + plan["remaining"] + + async def get_system_overview(self, request): + from ..config import GalleryConfig + + try: + output_dirs, auto_detected = self._output_dirs_with_origin() + counts, missing = await asyncio.gather( + self._run_in_executor(self.db.library_counts), + self._run_in_executor(self._missing_thumbnails, output_dirs), + ) + except Exception: + self.logger.exception("System overview failed") + return _failure("Failed to read the overview") + return web.json_response( + { + "success": True, + "library": { + "positive_prompts": counts["positive_prompts"], + "negative_variants": counts["negative_variants"], + }, + "images": {"indexed": counts["images"]}, + "health": health_summary(self._health_report), + "backup": None, # no verified backup status exists yet (S8) + "needs_review": {"pending": None}, # no review queue yet (Q6) + "sources": { + "configured": len(GalleryConfig.MONITORING_DIRECTORIES), + "effective": len(output_dirs), + "auto_detected": auto_detected, + }, + "thumbnails": {"missing": missing}, + }, + headers={"Cache-Control": "no-store"}, + ) + + # ── statistics ──────────────────────────────────────────────────── + + async def get_system_statistics(self, request): + try: + by_category, totals = await asyncio.gather( + self._run_in_executor(self.db.usage_by_category), + self._run_in_executor(self.db.get_statistics), + ) + except Exception: + self.logger.exception("System statistics failed") + return _failure("Failed to read statistics") + return web.json_response( + {"success": True, "by_category": by_category, "totals": totals}, + headers={"Cache-Control": "no-store"}, + ) diff --git a/tests/test_database_health.py b/tests/test_database_health.py new file mode 100644 index 00000000..e7adfc47 --- /dev/null +++ b/tests/test_database_health.py @@ -0,0 +1,392 @@ +"""The read-only health check (System › Overview and Maintenance). + +``run_health_checks`` reads one snapshot on its own ``mode=ro`` connection: +SQLite's integrity and foreign-key checks, stored prompt hashes against +``utils.hashing``, duplicate pair rows, the legacy consistency queries, image +links whose file is gone (only under readable sources) and the Needs-review +count when one is supplied. It never writes, and it never turns a check it +could not run into a count. +""" + +import hashlib +import os +import sqlite3 +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from database import health # noqa: E402 +from database.operations import PromptDatabase # noqa: E402 +from utils.hashing import generate_prompt_hash # noqa: E402 + +try: + from tests.open_handles import assert_closed # noqa: E402 +except ImportError: # discovered with tests/ as the top-level directory + from open_handles import assert_closed # noqa: E402 + +CHECK_IDS = [ + "integrity", + "foreign_keys", + "canonical_hashes", + "pair_uniqueness", + "missing_files", + "ambiguous_metadata", + "consistency", +] + + +def _digest(path): + if not os.path.exists(path): + return None + with open(path, "rb") as handle: + return hashlib.sha256(handle.read()).hexdigest() + + +class HealthTestCase(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.root = Path(os.path.realpath(self.temp.name)) + self.output = self.root / "output" + self.output.mkdir() + self.db_path = str(self.root / "prompts.db") + self.db = PromptDatabase(self.db_path) + + def tearDown(self): + self.db.close_all() + assert_closed(self, self.root) + self.temp.cleanup() + + def save(self, text, **metadata): + return self.db.save_prompt( + text, prompt_hash=generate_prompt_hash(text), **metadata + ) + + def image(self, prompt_id, name, *, exists=True, folder=None): + folder = Path(folder or self.output) + path = folder / name + if exists: + path.write_bytes(b"png") + return self.db.link_image_to_prompt(prompt_id, str(path)) + + def run_checks(self, roots=None, **kwargs): + roots = [str(self.output)] if roots is None else roots + return self.db.run_health_checks(roots, **kwargs) + + @staticmethod + def check(report, check_id): + return next(c for c in report["checks"] if c["id"] == check_id) + + def raw(self): + """A plain connection for planting damage the app itself never writes.""" + conn = sqlite3.connect(self.db_path) + conn.execute("PRAGMA foreign_keys=OFF") + return conn + + +class TestHealthReport(HealthTestCase): + def test_clean_library_passes_except_unavailable_review(self): + prompt = self.save("A clean prompt") + self.db.save_prompt_pair("Cat", "Blur") + self.image(prompt, "a.png") + report = self.run_checks(review_count=0) + self.assertTrue(report["success"]) + self.assertEqual([c["id"] for c in report["checks"]], CHECK_IDS) + self.assertEqual( + [c["label"] for c in report["checks"]], + [ + "Integrity check", + "Foreign keys & pair references", + "Canonical hashes", + "Pair uniqueness", + "Missing files", + "Ambiguous metadata", + "Consistency", + ], + ) + for check in report["checks"]: + self.assertEqual(check["status"], "passed", check) + self.assertEqual(check["count"], 0, check) + self.assertEqual(report["status"], "passed") + self.assertRegex(report["checked_at"], r"^\d{4}-\d\d-\d\dT.*\+00:00$") + + def test_health_check_never_writes(self): + prompt = self.save("Untouched") + self.image(prompt, "kept.png") + self.image(prompt, "gone.png", exists=False) + with self.raw() as conn: + conn.execute("UPDATE prompts SET hash='stale' WHERE id=?", (prompt,)) + conn.close() + # The app's own connection stays open, as in the running server. + self.db.get_statistics() + files = [self.db_path, self.db_path + "-wal"] + before = [_digest(path) for path in files] + shm_size = os.path.getsize(self.db_path + "-shm") + media = sorted(os.listdir(self.output)) + opened = [] + real_connect = sqlite3.connect + + class Recording(sqlite3.Connection): + changes_at_close = None + + def close(self): + Recording.changes_at_close = self.total_changes + super().close() + + def connect(*args, **kwargs): + conn = real_connect(*args, factory=Recording, **kwargs) + opened.append((args, kwargs, conn)) + return conn + + with patch.object(health.sqlite3, "connect", side_effect=connect): + report = self.run_checks() + self.assertEqual(report["status"], "incomplete") + self.assertEqual(len(opened), 1) + args, kwargs, conn = opened[0] + self.assertTrue(str(args[0]).endswith("?mode=ro"), args) + self.assertTrue(kwargs.get("uri")) + self.assertEqual(Recording.changes_at_close, 0) + with self.assertRaises(sqlite3.ProgrammingError): # closed in finally + conn.execute("SELECT 1") + self.assertEqual([_digest(path) for path in files], before) + # The -shm holds WAL read marks every reader updates; it never grows. + self.assertEqual(os.path.getsize(self.db_path + "-shm"), shm_size) + self.assertEqual(sorted(os.listdir(self.output)), media) + with self.db.model.get_connection() as live: + stored = live.execute( + "SELECT hash FROM prompts WHERE id=?", (prompt,) + ).fetchone()[0] + self.assertEqual(stored, "stale") + + def test_connection_closed_when_a_check_raises(self): + self.save("Any") + with ( + patch.object( + health, + "_canonical_hashes", + side_effect=sqlite3.OperationalError("boom"), + ), + self.assertLogs("prompt_manager.database.health", "ERROR"), + ): + report = self.run_checks(review_count=0) + canonical = self.check(report, "canonical_hashes") + self.assertEqual(canonical["status"], "failed") + self.assertIsNone(canonical["count"]) + self.assertEqual(canonical["detail"], ["The check could not finish."]) + self.assertEqual(self.check(report, "integrity")["status"], "passed") + self.assertEqual(report["status"], "incomplete") + self.db.close_all() + assert_closed(self, self.root) + + def test_unopenable_database_fails_every_database_check(self): + missing = PromptDatabase.__new__(PromptDatabase) + missing.model = type("M", (), {"db_path": str(self.root / "nope.db")})() + with self.assertLogs("prompt_manager.database.health", "ERROR"): + report = missing.run_health_checks([str(self.output)], review_count=0) + statuses = {c["id"]: c["status"] for c in report["checks"]} + self.assertEqual(statuses["integrity"], "failed") + self.assertEqual(statuses["missing_files"], "failed") + self.assertEqual(statuses["ambiguous_metadata"], "passed") + self.assertEqual(report["status"], "incomplete") + self.assertFalse(os.path.exists(self.root / "nope.db")) + + def test_integrity_and_foreign_key_issues_reported(self): + prompt = self.save("Linked") + with self.raw() as conn: + conn.execute("CREATE TABLE probe(a)") + conn.execute("CREATE INDEX probe_a ON probe(a)") + conn.executemany("INSERT INTO probe VALUES(?)", [(i,) for i in range(16)]) + conn.execute( + "INSERT INTO generated_images(prompt_id,image_path,filename,file_path)" + " VALUES(?,?,?,?)", + (9999, str(self.output / "orphan.png"), "orphan.png", "orphan"), + ) + conn.execute( + "INSERT INTO prompt_tags(prompt_id,tag_id) VALUES(?,?)", (prompt, 777) + ) + conn.close() + with self.raw() as conn: + conn.execute("PRAGMA writable_schema=ON") + conn.execute( + "UPDATE sqlite_master SET sql='CREATE INDEX probe_a ON probe(a DESC)'" + " WHERE name='probe_a'" + ) + conn.close() + self.db.close_all() # the next read sees the edited schema + (self.output / "orphan.png").write_bytes(b"png") + with self.raw() as conn: + expected = [r[0] for r in conn.execute("PRAGMA integrity_check(100)")] + conn.close() + self.assertGreater(len(expected), 10) + report = self.run_checks(review_count=0) + integrity = self.check(report, "integrity") + self.assertEqual(integrity["status"], "issues") + self.assertEqual(integrity["count"], len(expected)) + self.assertEqual(integrity["detail"], expected[:10]) + self.assertIn("missing from index probe_a", integrity["detail"][0]) + foreign = self.check(report, "foreign_keys") + self.assertEqual(foreign["status"], "issues") + self.assertEqual(foreign["count"], 2) + self.assertEqual( + sorted(foreign["detail"]), + [ + "generated_images row 1 references a missing prompts row", + "prompt_tags row 1 references a missing tags row", + ], + ) + consistency = self.check(report, "consistency") + self.assertEqual(consistency["status"], "issues") + self.assertEqual(consistency["count"], 1) + self.assertEqual( + consistency["detail"], ["Image 1 references non-existent prompt 9999"] + ) + self.assertEqual(report["status"], "issues") + + def test_canonical_hash_mismatch_counts_rows(self): + ids = [self.save(f"Prompt {n}") for n in range(14)] + with self.raw() as conn: + for n, prompt_id in enumerate(ids[:12]): + conn.execute( + "UPDATE prompts SET hash=? WHERE id=?", (f"bad{n}", prompt_id) + ) + conn.execute("UPDATE prompts SET hash=NULL WHERE id=?", (ids[12],)) + conn.close() + # Normalised spellings hash alike: this one is not a mismatch. + self.save(" MIXED Case ") + report = self.run_checks(review_count=0) + canonical = self.check(report, "canonical_hashes") + self.assertEqual(canonical["status"], "issues") + self.assertEqual(canonical["count"], 13) + self.assertEqual(len(canonical["detail"]), 10) + self.assertEqual( + canonical["detail"][0], + f"Prompt {ids[0]}: the stored hash does not match its text", + ) + + def test_duplicate_pair_rows_reported(self): + pair = self.db.save_prompt_pair("Cat", "") + positive = self.db.get_prompt_pair(pair)["positive_prompt_id"] + self.db.save_prompt_pair("Dog", "Blur") + with self.raw() as conn: + # Only the partial index keeps positive-only pairs unique; a restored + # or hand-edited database may have lost it. + conn.execute("DROP INDEX idx_prompt_pairs_positive_only") + conn.executemany( + "INSERT INTO prompt_pairs(positive_prompt_id,negative_prompt_id)" + " VALUES(?,NULL)", + [(positive,), (positive,)], + ) + conn.close() + report = self.run_checks(review_count=0) + pairs = self.check(report, "pair_uniqueness") + self.assertEqual(pairs["status"], "issues") + self.assertEqual(pairs["count"], 2) + self.assertEqual( + pairs["detail"], + [f"Positive prompt {positive} with no negative: 3 pair rows"], + ) + + +class TestMissingFiles(HealthTestCase): + def test_missing_files_counted_under_readable_sources_only(self): + prompt = self.save("Images") + self.image(prompt, "here.png") + for n in range(12): + self.image(prompt, f"gone-{n}.png", exists=False) + elsewhere = self.root / "elsewhere" + elsewhere.mkdir() + self.image(prompt, "outside.png", exists=False, folder=elsewhere) + report = self.run_checks(review_count=0) + missing = self.check(report, "missing_files") + self.assertEqual(missing["status"], "issues") + self.assertEqual(missing["count"], 12) + self.assertEqual(len(missing["detail"]), 10) + self.assertRegex(missing["detail"][0], r"^Image \d+: gone-0\.png is missing$") + self.assertNotIn(str(self.root), " ".join(missing["detail"])) + with self.db.model.get_connection() as conn: + self.assertEqual( + conn.execute("SELECT COUNT(*) FROM generated_images").fetchone()[0], + 14, + ) + + def test_links_outside_every_source_are_named_not_counted(self): + prompt = self.save("Images") + elsewhere = self.root / "elsewhere" + elsewhere.mkdir() + self.image(prompt, "outside.png", exists=False, folder=elsewhere) + missing = self.check(self.run_checks(review_count=0), "missing_files") + self.assertEqual(missing["status"], "passed") + self.assertEqual(missing["count"], 0) + self.assertEqual( + missing["detail"], + ["1 link outside the image sources was not checked."], + ) + + def test_missing_file_vs_unavailable_root(self): + prompt = self.save("Images") + absent = self.root / "unmounted" + absent.mkdir() + self.image(prompt, "on-nas.png", folder=absent) + self.image(prompt, "gone.png", exists=False) + os.unlink(absent / "on-nas.png") + absent.rmdir() + report = self.run_checks([str(self.output), str(absent)], review_count=0) + missing = self.check(report, "missing_files") + self.assertEqual(missing["status"], "unavailable") + self.assertIsNone(missing["count"]) + self.assertEqual( + missing["detail"], + ["1 image source is unavailable, so missing files were not counted."], + ) + self.assertEqual(report["status"], "incomplete") + + def test_no_sources_is_unavailable(self): + self.save("Images") + missing = self.check(self.run_checks([], review_count=0), "missing_files") + self.assertEqual(missing["status"], "unavailable") + self.assertIsNone(missing["count"]) + self.assertEqual(missing["detail"], ["No image source is available."]) + + +class TestAmbiguousMetadata(HealthTestCase): + def test_ambiguous_metadata_unavailable_without_review_queue(self): + self.save("Any") + report = self.run_checks() + ambiguous = self.check(report, "ambiguous_metadata") + self.assertEqual(ambiguous["status"], "unavailable") + self.assertIsNone(ambiguous["count"]) + self.assertEqual(ambiguous["detail"], ["Needs review is not available yet."]) + self.assertEqual(report["status"], "incomplete") + + def test_review_count_reports_images_needing_review(self): + report = self.run_checks(review_count=3) + ambiguous = self.check(report, "ambiguous_metadata") + self.assertEqual(ambiguous["status"], "issues") + self.assertEqual(ambiguous["count"], 3) + self.assertEqual(ambiguous["detail"], []) + self.assertEqual(report["status"], "issues") + + def test_invalid_review_count_is_rejected(self): + for value in (-1, True, "3", 1.5): + with self.assertRaises(ValueError): + self.run_checks(review_count=value) + + +class TestOverallStatus(unittest.TestCase): + def test_overall_status_precedence(self): + def checks(*statuses): + return [{"status": status} for status in statuses] + + self.assertEqual(health.overall_status(checks("passed", "passed")), "passed") + self.assertEqual(health.overall_status(checks("passed", "issues")), "issues") + self.assertEqual( + health.overall_status(checks("issues", "unavailable")), "incomplete" + ) + self.assertEqual(health.overall_status(checks("failed")), "incomplete") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_library.py b/tests/test_library.py index fd83cb7c..5422be9d 100644 --- a/tests/test_library.py +++ b/tests/test_library.py @@ -1706,6 +1706,82 @@ class TestLibrary(unittest.TestCase): self.db.list_library_images(positive, pair_id=foreign, archive="all") +class LibraryCounts(unittest.TestCase): + """System › Overview counts and usage statistics follow Library eligibility.""" + + setUp = TestLibrary.setUp + tearDown = TestLibrary.tearDown + save = TestLibrary.save + state = TestLibrary.state + + def test_counts_use_library_eligibility(self): + self.db.save_prompt_pair("Cat", "Blur") + archived = self.db.save_prompt_pair("Cat", "Noise") + self.db.archive_prompt_pair(archived) + titled_pair = self.db.save_prompt_pair("Dog", "Titled negative") + titled = self.db.get_prompt_pair(titled_pair)["negative_prompt_id"] + self.state(titled, "A titled parent") # zero pairs of its own: counted + legacy = self.save("Legacy") + self.db.link_image_to_prompt(legacy, str(Path(self.temp.name) / "a.png")) + self.db.link_image_to_prompt(legacy, str(Path(self.temp.name) / "b.png")) + counts = self.db.library_counts() + # Cat, Dog, the titled negative and Legacy; "Blur" and "Noise" are + # negative-only. + self.assertEqual(counts["positive_prompts"], 4) + # Cat+Blur and Dog+Titled negative; the archived pair is excluded. + self.assertEqual(counts["negative_variants"], 2) + self.assertEqual(counts["images"], 2) + listed = self.db.list_library() + self.assertEqual(counts["positive_prompts"], listed["total"]) + self.assertEqual(counts["negative_variants"], listed["variant_total"]) + + def test_counts_of_an_empty_library_are_zero(self): + self.assertEqual( + self.db.library_counts(), + {"positive_prompts": 0, "negative_variants": 0, "images": 0}, + ) + + def test_statistics_sum_real_run_counts(self): + a = self.save("Alpha", category="Landscapes", rating=5) + b = self.save("Beta", category="Landscapes", rating=4) + self.save("Gamma", category="Landscapes") # unrated: not in the average + c = self.save("Delta", category="Portraits", rating=3) + blank = self.save("Epsilon", category=" ") + uncategorised = self.save("Zeta") + self.db.record_prompt_use(a, times=3) + self.db.record_prompt_use(b, times=2) + self.db.record_prompt_use(c) + self.db.record_prompt_use(blank) + self.db.record_prompt_use(uncategorised, times=4) + # A negative-only prompt's runs are the pair's runs: never counted twice. + pair = self.db.save_prompt_pair("Alpha", "Only negative") + negative = self.db.get_prompt_pair(pair)["negative_prompt_id"] + with self.db.model.get_connection() as connection: + connection.execute( + "UPDATE prompts SET category='Landscapes',run_count=50 WHERE id=?", + (negative,), + ) + rows = self.db.usage_by_category() + self.assertEqual( + rows, + [ + {"category": "Landscapes", "runs": 5, "average_rating": 4.5}, + {"category": None, "runs": 5, "average_rating": None}, + {"category": "Portraits", "runs": 1, "average_rating": 3.0}, + ], + ) + + def test_counts_and_usage_write_nothing(self): + self.save("Alpha", category="A") + with self.db.model.get_connection() as connection: + before = connection.total_changes + self.db.library_counts() + self.db.usage_by_category() + with self.db.model.get_connection() as connection: + self.assertEqual(connection.total_changes, before) + self.assertFalse(connection.in_transaction) + + class DisplayTitleUnits(unittest.TestCase): """The first-line title is cut like the Library's JavaScript title().""" diff --git a/tests/test_system_api.py b/tests/test_system_api.py new file mode 100644 index 00000000..7e2b7881 --- /dev/null +++ b/tests/test_system_api.py @@ -0,0 +1,304 @@ +"""System routes (py/api/system.py): health check job, Overview and statistics. + +The health check runs off-loop as a ``read`` registry job of kind ``health`` +(one at a time, beside a scan), and the API keeps its last report. The +Overview reads only real values: Library-eligible counts, indexed images, the +last health result or null, no backup until one exists, the effective sources +and whether they were auto-detected, and the thumbnail plan's missing count. +""" + +import asyncio +import os +import sys +import threading +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 HEALTH_JOB_KIND # noqa: E402 +from py.config import GalleryConfig # noqa: E402 +from utils.hashing import generate_prompt_hash # noqa: E402 + +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 + +WAIT = 5 + + +class SystemAPITestCase(AdminAPITestCase): + async def get_application(self): + app = await super().get_application() + # Auto-detection searches the disk; here it finds the fake ComfyUI only. + self.api._find_comfyui_output_dir = lambda: ( + str(self.output_dir) if self.output_dir.is_dir() else None + ) + return app + + def save(self, text, **metadata): + return self.api.db.save_prompt( + text, prompt_hash=generate_prompt_hash(text), **metadata + ) + + def image(self, prompt_id, name, *, exists=True, folder=None): + path = Path(folder or self.output_dir) / name + if exists: + path.write_bytes(b"png") + return self.api.db.link_image_to_prompt(prompt_id, str(path)) + + async def json(self, method, url, status=200): + response = await self.client.request(method, url) + self.assertEqual(response.status, status, await response.text()) + return await response.json() + + async def run_health(self): + body = await self.json("POST", "/prompt_manager/system/health", 202) + job = self.api._jobs.get(body["job"]["id"]) + await asyncio.wait_for(asyncio.shield(job.task), WAIT) + return body["job"], job + + +class TestHealthRoute(SystemAPITestCase): + async def test_no_report_before_a_check(self): + body = await self.json("GET", "/prompt_manager/system/health") + self.assertEqual(body, {"success": True, "report": None}) + + async def test_health_check_runs_as_a_read_job_and_keeps_its_report(self): + prompt = self.save("Images") + self.image(prompt, "kept.png") + self.image(prompt, "gone.png", exists=False) + summary, job = await self.run_health() + self.assertEqual(summary["kind"], HEALTH_JOB_KIND) + self.assertEqual(summary["title"], "Health check") + self.assertEqual(summary["admission"], "read") + self.assertFalse(summary["cancellable"]) + self.assertEqual(summary["counters"], {}) + self.assertEqual(job.state, "finished") + self.assertEqual(job.progress, 100) + self.assertEqual(job.status_line, "Health check incomplete") + report = (await self.json("GET", "/prompt_manager/system/health"))["report"] + self.assertEqual(report["status"], "incomplete") + missing = next(c for c in report["checks"] if c["id"] == "missing_files") + self.assertEqual((missing["status"], missing["count"]), ("issues", 1)) + ambiguous = next(c for c in report["checks"] if c["id"] == "ambiguous_metadata") + self.assertEqual(ambiguous["status"], "unavailable") + self.assertNotIn(self.tmpdir, str(report)) + + async def test_health_check_is_single_flight_and_runs_beside_a_scan(self): + release = threading.Event() + entered = threading.Event() + real = self.api.db.run_health_checks + + def gated(*args, **kwargs): + entered.set() + release.wait(WAIT) + return real(*args, **kwargs) + + scan_gate = asyncio.Event() + + async def scanning(job): + await scan_gate.wait() + + scan = self.api._jobs.start( + "scan", "Metadata scan", admission="records", runner=scanning + ) + with patch.object(self.api.db, "run_health_checks", side_effect=gated): + first = await self.json("POST", "/prompt_manager/system/health", 202) + loop = asyncio.get_running_loop() + await loop.run_in_executor(None, entered.wait, WAIT) + busy = await self.json("POST", "/prompt_manager/system/health", 409) + self.assertEqual(busy["code"], "busy") + self.assertEqual( + busy["running"], {"id": first["job"]["id"], "title": "Health check"} + ) + release.set() + await asyncio.wait_for(self.api._jobs.get(first["job"]["id"]).task, WAIT) + scan_gate.set() + await asyncio.wait_for(scan.task, WAIT) + + async def test_failed_check_keeps_the_previous_report(self): + await self.run_health() + before = (await self.json("GET", "/prompt_manager/system/health"))["report"] + with ( + patch.object( + self.api.db, "run_health_checks", side_effect=RuntimeError("boom") + ), + self.assertLogs("prompt_manager.api", "ERROR"), + ): + body = await self.json("POST", "/prompt_manager/system/health", 202) + job = self.api._jobs.get(body["job"]["id"]) + await asyncio.wait_for(asyncio.shield(job.task), WAIT) + self.assertEqual(job.state, "failed") + after = (await self.json("GET", "/prompt_manager/system/health"))["report"] + self.assertEqual(after, before) + + async def test_absent_configured_source_makes_missing_files_unavailable(self): + prompt = self.save("Images") + self.image(prompt, "gone.png", exists=False) + absent = self.comfy_dir / "output" / "unmounted" + GalleryConfig.MONITORING_DIRECTORIES = [ + self._root(self.output_dir), + str(absent), + ] + await self.run_health() + report = (await self.json("GET", "/prompt_manager/system/health"))["report"] + missing = next(c for c in report["checks"] if c["id"] == "missing_files") + self.assertEqual(missing["status"], "unavailable") + self.assertIsNone(missing["count"]) + + +class TestOverviewRoute(SystemAPITestCase): + async def test_overview_reports_real_counts(self): + self.api.db.save_prompt_pair("Cat", "Blur") + archived = self.api.db.save_prompt_pair("Cat", "Noise") + self.api.db.archive_prompt_pair(archived) + legacy = self.save("Legacy") + self.image(legacy, "a.png") + self.image(legacy, "b.png") + body = await self.json("GET", "/prompt_manager/system/overview") + self.assertTrue(body["success"]) + self.assertEqual( + body["library"], {"positive_prompts": 2, "negative_variants": 1} + ) + self.assertEqual(body["images"], {"indexed": 2}) + self.assertEqual( + set(body), + { + "success", + "library", + "images", + "health", + "backup", + "needs_review", + "sources", + "thumbnails", + }, + ) + + async def test_overview_never_fabricates_backup_or_health(self): + body = await self.json("GET", "/prompt_manager/system/overview") + self.assertIsNone(body["health"]) + self.assertIsNone(body["backup"]) + self.assertEqual(body["needs_review"], {"pending": None}) + self.save("Any") + await self.run_health() + report = (await self.json("GET", "/prompt_manager/system/health"))["report"] + body = await self.json("GET", "/prompt_manager/system/overview") + self.assertEqual( + body["health"], + {"checked_at": report["checked_at"], "status": "incomplete", "issues": 0}, + ) + self.assertIsNone(body["backup"]) + + async def test_overview_health_counts_checks_with_issues(self): + prompt = self.save("Images") + self.image(prompt, "gone.png", exists=False) + self.image(prompt, "gone-too.png", exists=False) + await self.run_health() + body = await self.json("GET", "/prompt_manager/system/overview") + self.assertEqual(body["health"]["issues"], 1) + + async def test_overview_sources_distinguish_auto_detect(self): + body = await self.json("GET", "/prompt_manager/system/overview") + self.assertEqual( + body["sources"], {"configured": 0, "effective": 1, "auto_detected": True} + ) + second = self.output_dir / "archive" + second.mkdir() + GalleryConfig.MONITORING_DIRECTORIES = [ + self._root(self.output_dir), + self._root(second), + ] + body = await self.json("GET", "/prompt_manager/system/overview") + self.assertEqual( + body["sources"], {"configured": 2, "effective": 2, "auto_detected": False} + ) + # Every configured source unavailable: ComfyUI's output is used instead. + GalleryConfig.MONITORING_DIRECTORIES = [str(self.comfy_dir / "absent")] + body = await self.json("GET", "/prompt_manager/system/overview") + self.assertEqual( + body["sources"], {"configured": 1, "effective": 1, "auto_detected": True} + ) + + async def test_overview_thumbnails_follow_the_status_plan(self): + (self.output_dir / "one.png").write_bytes(b"png") + (self.output_dir / "two.png").write_bytes(b"png") + status = await self.json("GET", "/prompt_manager/images/thumbnails/status") + body = await self.json("GET", "/prompt_manager/system/overview") + self.assertEqual(body["thumbnails"], {"missing": status["pending"]}) + self.assertEqual(body["thumbnails"]["missing"], 2) + + async def test_overview_thumbnails_unknown_without_sources(self): + self.output_dir.rmdir() + body = await self.json("GET", "/prompt_manager/system/overview") + self.assertEqual(body["thumbnails"], {"missing": None}) + self.assertEqual( + body["sources"], {"configured": 0, "effective": 0, "auto_detected": False} + ) + + async def test_overview_thumbnail_failure_is_unknown_not_zero(self): + with ( + patch.object( + self.api, "_plan_gallery_thumbnails", side_effect=OSError("denied") + ), + self.assertLogs("prompt_manager.api", "WARNING"), + ): + body = await self.json("GET", "/prompt_manager/system/overview") + self.assertTrue(body["success"]) + self.assertEqual(body["thumbnails"], {"missing": None}) + + async def test_overview_failure_is_an_error_envelope(self): + with ( + patch.object( + self.api.db, "library_counts", side_effect=RuntimeError("secret") + ), + self.assertLogs("prompt_manager.api", "ERROR"), + ): + body = await self.json("GET", "/prompt_manager/system/overview", 500) + self.assertEqual( + body, + {"success": False, "code": None, "error": "Failed to read the overview"}, + ) + + +class TestStatisticsRoute(SystemAPITestCase): + async def test_statistics_sum_real_run_counts(self): + a = self.save("Alpha", category="Landscapes", rating=5) + b = self.save("Beta", category="Portraits", rating=2) + self.api.db.record_prompt_use(a, times=3) + self.api.db.record_prompt_use(b) + body = await self.json("GET", "/prompt_manager/system/statistics") + self.assertEqual( + body["by_category"], + [ + {"category": "Landscapes", "runs": 3, "average_rating": 5.0}, + {"category": "Portraits", "runs": 1, "average_rating": 2.0}, + ], + ) + self.assertEqual(body["totals"], self.api.db.get_statistics()) + self.assertTrue(body["success"]) + + async def test_statistics_failure_is_an_error_envelope(self): + with ( + patch.object( + self.api.db, "usage_by_category", side_effect=RuntimeError("secret") + ), + self.assertLogs("prompt_manager.api", "ERROR"), + ): + body = await self.json("GET", "/prompt_manager/system/statistics", 500) + self.assertEqual( + body, + {"success": False, "code": None, "error": "Failed to read statistics"}, + ) + + +if __name__ == "__main__": + import unittest + + unittest.main()