From 78a12b67fd1800521d97498da2b256b2189f47db Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Wed, 7 Oct 2026 18:59:09 -0700 Subject: [PATCH] test: cover admin and core API routes Route tests for stats, cleanup, duplicate scan/delete, settings, diagnostics, test-link, maintenance (every operation, success and failure), the output scan internals, and contract-level tests for backup/restore (status codes and envelope only). --- tests/test_api_admin.py | 489 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 489 insertions(+) diff --git a/tests/test_api_admin.py b/tests/test_api_admin.py index d694d2a..d660587 100644 --- a/tests/test_api_admin.py +++ b/tests/test_api_admin.py @@ -604,5 +604,494 @@ class TestSettingsReadMonitorThroughModule(AdminAPITestCase): self.assertEqual(data["settings"]["gallery_root_paths"], ["output"]) +def _raise(*args, **kwargs): + raise RuntimeError("boom") + + +class TestStatsCleanupAndErrors(AdminAPITestCase): + + async def test_stats_success(self): + resp = await self.client.request("GET", "/prompt_manager/stats") + data = await resp.json() + self.assertEqual(resp.status, 200) + self.assertIn("total_prompts", data["stats"]) + + async def test_stats_db_failure_is_500(self): + self.api.db.get_statistics = _raise + resp = await self.client.request("GET", "/prompt_manager/stats") + self.assertEqual(resp.status, 500) + self.assertFalse((await resp.json())["success"]) + + async def test_cleanup_success(self): + resp = await self.client.request("POST", "/prompt_manager/cleanup") + data = await resp.json() + self.assertEqual(resp.status, 200) + self.assertEqual(data["duplicates_removed"], 0) + + async def test_cleanup_failure_is_500(self): + self.api.db.cleanup_duplicates = _raise + resp = await self.client.request("POST", "/prompt_manager/cleanup") + self.assertEqual(resp.status, 500) + + async def test_scan_duplicates_failure_is_500(self): + self.api._find_comfyui_output_dir = _raise + resp = await self.client.request("GET", "/prompt_manager/scan_duplicates") + body = await resp.text() + self.assertEqual(resp.status, 500) + self.assertNotIn("boom", body) + + async def test_scan_duplicates_without_output_dir_is_empty(self): + self.api._find_comfyui_output_dir = lambda: None + resp = await self.client.request("GET", "/prompt_manager/scan_duplicates") + data = await resp.json() + self.assertEqual(resp.status, 200) + self.assertEqual(data["duplicates"], []) + + +class TestDeleteDuplicatesEndpoint(AdminAPITestCase): + + async def _post(self, payload): + return await self.client.request( + "POST", "/prompt_manager/delete_duplicate_images", json=payload + ) + + async def test_empty_list_is_400(self): + resp = await self._post({"image_paths": []}) + self.assertEqual(resp.status, 400) + + async def test_non_list_is_400(self): + resp = await self._post({"image_paths": "a.png"}) + self.assertEqual(resp.status, 400) + + async def test_malformed_json_is_400(self): + resp = await self.client.request( + "POST", + "/prompt_manager/delete_duplicate_images", + data=b"{", + headers={"Content-Type": "application/json"}, + ) + self.assertEqual(resp.status, 400) + + async def test_failures_are_reported(self): + GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir)] + resp = await self._post({"image_paths": ["ghost.png"]}) + data = await resp.json() + self.assertEqual(resp.status, 200) + self.assertEqual(data["failed_count"], 1) + self.assertIn("1 failed", data["message"]) + self.assertEqual(len(data["failed_files"]), 1) + + async def test_unexpected_error_is_500(self): + self.api._find_comfyui_output_dir = _raise + resp = await self._post({"image_paths": ["a.png"]}) + self.assertEqual(resp.status, 500) + + def test_thumbnail_removal_failure_does_not_fail_delete(self): + target = self.output_dir / "x.png" + target.write_bytes(b"x") + thumb_dir = self.output_dir / "thumbnails" / "x_thumb.png" + thumb_dir.mkdir(parents=True) + (thumb_dir / "child").write_bytes(b"c") # os.remove on a dir raises OSError + + result = self.api._delete_duplicate_images_sync( + [str(target)], str(self.output_dir) + ) + + self.assertEqual(result["deleted_count"], 1) + self.assertFalse(target.exists()) + self.assertTrue(thumb_dir.exists()) + + def test_unexpected_error_per_file_is_reported_without_path(self): + self.api._delete_one_duplicate = lambda *a: (_ for _ in ()).throw( + PermissionError(13, "Permission denied", os.path.join(self.tmpdir, "x.png")) + ) + result = self.api._delete_duplicate_images_sync(["x.png"], str(self.output_dir)) + self.assertEqual(result["failed_count"], 1) + self.assertIn("Permission denied", result["failed_files"][0]) + self.assertNotIn(self.tmpdir, result["failed_files"][0]) + + +class TestSettingsMisc(AdminAPITestCase): + + async def test_result_timeout_and_display_mode_saved(self): + resp = await self.client.request( + "POST", + "/prompt_manager/settings", + json={"result_timeout": 9, "webui_display_mode": "popup"}, + ) + data = await resp.json() + self.assertEqual(resp.status, 200) + self.assertFalse(data["restart_required"]) + self.assertEqual(PromptManagerConfig.RESULT_TIMEOUT, 9) + self.assertEqual(PromptManagerConfig.WEBUI_DISPLAY_MODE, "popup") + saved = self._read_config() + self.assertEqual(saved["web_ui"]["result_timeout"], 9) + + async def test_invalid_result_timeout_is_400(self): + resp = await self.client.request( + "POST", "/prompt_manager/settings", json={"result_timeout": "soon"} + ) + self.assertEqual(resp.status, 400) + + async def test_malformed_json_is_400(self): + resp = await self.client.request( + "POST", + "/prompt_manager/settings", + data=b"{oops", + headers={"Content-Type": "application/json"}, + ) + self.assertEqual(resp.status, 400) + + async def test_unchanged_roots_do_not_require_restart(self): + GalleryConfig.MONITORING_DIRECTORIES = [ + os.path.normcase(os.path.realpath(self.output_dir)) + ] + resp = await self.client.request( + "POST", + "/prompt_manager/settings", + json={"gallery_root_paths": [str(self.output_dir)]}, + ) + data = await resp.json() + self.assertEqual(resp.status, 200) + self.assertFalse(data["restart_required"]) + + async def test_unwritable_config_path_still_succeeds(self): + os.environ[CONFIG_PATH_ENV] = os.path.join(self.tmpdir, "missing", "c.json") + resp = await self.client.request( + "POST", "/prompt_manager/settings", json={"result_timeout": 3} + ) + self.assertEqual(resp.status, 200) + + async def test_save_unexpected_error_is_500(self): + self.api._parse_gallery_roots = _raise + resp = await self.client.request( + "POST", "/prompt_manager/settings", json={"gallery_root_paths": []} + ) + self.assertEqual(resp.status, 500) + self.assertNotIn("boom", await resp.text()) + + async def test_get_unexpected_error_is_500(self): + self.api._monitored_directories = _raise + resp = await self.client.request("GET", "/prompt_manager/settings") + self.assertEqual(resp.status, 500) + + +class TestDiagnostics(AdminAPITestCase): + + def _install_monitor(self, monitor): + import utils.image_monitor as im_mod + + orig = im_mod._monitor_instance + im_mod._monitor_instance = monitor + self.addCleanup(setattr, im_mod, "_monitor_instance", orig) + + async def _diag(self): + resp = await self.client.request("GET", "/prompt_manager/diagnostics") + return resp, (await resp.json())["diagnostics"] + + async def test_healthy_report(self): + self.api.db.save_prompt(text="hello", prompt_hash="h1") + self._install_monitor(FakeMonitor([str(self.output_dir)])) + + resp, diag = await self._diag() + + self.assertEqual(resp.status, 200) + self.assertEqual(diag["database"]["status"], "ok") + self.assertEqual(diag["database"]["prompt_count"], 1) + self.assertTrue(diag["database"]["has_images_table"]) + self.assertEqual(diag["dependencies"]["status"], "ok") + self.assertEqual(diag["comfyui_output"]["status"], "ok") + self.assertEqual(diag["image_monitor"]["status"], "ok") + + async def test_corrupt_database_reported(self): + bad = os.path.join(self.tmpdir, "bad.db") + with open(bad, "wb") as f: + f.write(b"not a database at all, definitely not sqlite") + self.api.db.model.db_path = bad + + _, diag = await self._diag() + + self.assertEqual(diag["database"]["status"], "error") + self.assertIn("Database error", diag["database"]["message"]) + + async def test_monitor_not_running_and_no_output_dirs(self): + self._install_monitor(None) + sys.modules.pop("folder_paths", None) + orig_cwd = os.getcwd() + os.chdir(self.outside_dir) + self.addCleanup(os.chdir, orig_cwd) + + _, diag = await self._diag() + + self.assertEqual(diag["image_monitor"]["status"], "error") + self.assertIn("not initialized", diag["image_monitor"]["message"]) + self.assertEqual(diag["comfyui_output"]["status"], "warning") + self.assertEqual(diag["comfyui_output"]["output_dirs"], []) + + async def test_relative_output_fallback_without_folder_paths(self): + sys.modules.pop("folder_paths", None) + orig_cwd = os.getcwd() + os.chdir(self.comfy_dir) # has an "output" child + self.addCleanup(os.chdir, orig_cwd) + + _, diag = await self._diag() + + self.assertEqual(diag["comfyui_output"]["status"], "ok") + self.assertEqual(diag["comfyui_output"]["output_dirs"], ["output"]) + + async def test_monitor_status_failure_reported(self): + broken = MagicMock() + broken.get_status = _raise + self._install_monitor(broken) + + _, diag = await self._diag() + + self.assertEqual(diag["image_monitor"]["status"], "error") + self.assertIn("Failed to get monitor status", diag["image_monitor"]["message"]) + + async def test_unexpected_error_is_500(self): + import py.api.admin as admin_module + + orig = admin_module._diagnose_database + admin_module._diagnose_database = _raise + self.addCleanup(setattr, admin_module, "_diagnose_database", orig) + resp = await self.client.request("GET", "/prompt_manager/diagnostics") + self.assertEqual(resp.status, 500) + + +class TestTestImageLink(AdminAPITestCase): + + async def _post(self, payload): + return await self.client.request( + "POST", "/prompt_manager/diagnostics/test-link", json=payload + ) + + async def test_missing_prompt_id_is_400(self): + resp = await self._post({}) + self.assertEqual(resp.status, 400) + + async def test_links_test_image(self): + pid = self.api.db.save_prompt(text="hello", prompt_hash="h1") + resp = await self._post({"prompt_id": pid}) + data = await resp.json() + self.assertEqual(resp.status, 200) + self.assertEqual(data["result"]["status"], "ok") + self.assertIsInstance(data["result"]["image_id"], int) + + async def test_db_failure_is_reported_in_result(self): + self.api.db.link_image_to_prompt = _raise + resp = await self._post({"prompt_id": 1}) + data = await resp.json() + self.assertEqual(resp.status, 200) + self.assertFalse(data["success"]) + self.assertEqual(data["result"]["status"], "error") + + async def test_malformed_json_is_400(self): + resp = await self.client.request( + "POST", + "/prompt_manager/diagnostics/test-link", + data=b"nope", + headers={"Content-Type": "application/json"}, + ) + self.assertEqual(resp.status, 400) + + +class TestMaintenance(AdminAPITestCase): + + ALL_OPS = [ + "cleanup_duplicates", + "vacuum", + "cleanup_orphaned_images", + "check_hash_duplicates", + "statistics", + "prune_orphaned_prompts", + "check_consistency", + ] + + async def _post(self, payload=None, **kwargs): + if payload is None: + return await self.client.request( + "POST", "/prompt_manager/maintenance", **kwargs + ) + return await self.client.request( + "POST", "/prompt_manager/maintenance", json=payload, **kwargs + ) + + async def test_default_operations_without_body(self): + resp = await self._post(data=b"", headers={"Content-Type": "text/plain"}) + data = await resp.json() + self.assertEqual(resp.status, 200) + self.assertEqual(data["operations_completed"], 3) + self.assertTrue(data["all_successful"]) + self.assertEqual( + set(data["results"]), + {"cleanup_duplicates", "vacuum", "cleanup_orphaned_images"}, + ) + + async def test_all_operations(self): + self.api.db.save_prompt(text="orphan", prompt_hash="h1") + resp = await self._post({"operations": self.ALL_OPS}) + data = await resp.json() + self.assertEqual(resp.status, 200) + self.assertEqual(data["operations_completed"], len(self.ALL_OPS)) + self.assertTrue(data["all_successful"], data["results"]) + self.assertIn("info", data["results"]["statistics"]) + self.assertEqual(data["results"]["check_consistency"]["issues_found"], 0) + + async def test_unknown_operation_does_nothing(self): + resp = await self._post({"operations": ["reboot"]}) + data = await resp.json() + self.assertEqual(data["operations_completed"], 0) + self.assertTrue(data["all_successful"]) + + async def test_each_operation_reports_its_own_failure(self): + self.api.db.cleanup_duplicates = _raise + self.api.db.model.vacuum_database = _raise + self.api.db.cleanup_missing_images = _raise + self.api.db.check_hash_duplicates = _raise + self.api.db.model.get_database_info = _raise + self.api.db.prune_orphaned_prompts = _raise + self.api.db.check_consistency = _raise + + resp = await self._post({"operations": self.ALL_OPS}) + data = await resp.json() + + self.assertEqual(resp.status, 200) + self.assertFalse(data["all_successful"]) + self.assertEqual(len(data["results"]), len(self.ALL_OPS)) + for name in self.ALL_OPS: + self.assertFalse(data["results"][name]["success"], name) + self.assertEqual(data["results"][name]["error"], "boom") + + async def test_unexpected_error_is_500(self): + self.api._run_in_executor = _raise + resp = await self._post({"operations": ["vacuum"]}) + self.assertEqual(resp.status, 500) + + +class TestBackupRestoreContract(AdminAPITestCase): + """Contract tests only: status codes and envelope of backup/restore.""" + + async def test_backup_downloads_sqlite_file(self): + self.api.db.save_prompt(text="keep me", prompt_hash="h1") + resp = await self.client.request("GET", "/prompt_manager/backup") + body = await resp.read() + self.assertEqual(resp.status, 200) + self.assertIn("attachment", resp.headers["Content-Disposition"]) + self.assertTrue(body.startswith(b"SQLite format 3")) + + async def test_backup_without_db_file_is_404(self): + self.api.db.model.db_path = os.path.join(self.tmpdir, "nope.db") + resp = await self.client.request("GET", "/prompt_manager/backup") + self.assertEqual(resp.status, 404) + + async def _restore(self, field_name, content): + from aiohttp import FormData + + form = FormData() + form.add_field(field_name, content, filename="upload.db") + return await self.client.request("POST", "/prompt_manager/restore", data=form) + + async def test_restore_wrong_field_is_400(self): + resp = await self._restore("other", b"x") + self.assertEqual(resp.status, 400) + + async def test_restore_empty_file_is_400(self): + resp = await self._restore("database_file", b"") + self.assertEqual(resp.status, 400) + + async def test_restore_non_sqlite_is_400_and_db_untouched(self): + self.api.db.save_prompt(text="keep me", prompt_hash="h1") + resp = await self._restore("database_file", b"definitely not sqlite data") + self.assertEqual(resp.status, 400) + self.assertIsNotNone(self.api.db.get_prompt_by_hash("h1")) + + async def test_restore_valid_db_succeeds(self): + import sqlite3 + + PromptManagerConfig.DEFAULT_DB_PATH = self.api.db.model.db_path + self.api.db.save_prompt(text="restored", prompt_hash="h1") + # Rows may still sit in the WAL file: build a self-contained copy. + snapshot = os.path.join(self.tmpdir, "snapshot.db") + with sqlite3.connect(self.api.db.model.db_path) as src: + with sqlite3.connect(snapshot) as dst: + src.backup(dst) + with open(snapshot, "rb") as f: + upload = f.read() + resp = await self._restore("database_file", upload) + data = await resp.json() + self.assertEqual(resp.status, 200, data) + self.assertTrue(data["success"]) + self.assertEqual(data["prompt_count"], 1) + with sqlite3.connect(self.api.db.model.db_path) as conn: + self.assertEqual( + conn.execute("SELECT COUNT(*) FROM prompts").fetchone()[0], 1 + ) + + +class TestOutputScanInternals(AdminAPITestCase): + + def test_metadata_extraction_failure_yields_empty(self): + self.api._extract_comfyui_metadata = _raise + results = self.api._extract_batch_metadata_sync([Path("a.png")]) + self.assertEqual(results, [(Path("a.png"), {})]) + + def test_ingest_outcomes(self): + f = Path("gen.png") + meta = { + "prompt": json.dumps( + {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a cat"}}} + ) + } + self.assertIsNone(self.api._ingest_scanned_file(f, {})) + self.assertIsNone(self.api._ingest_scanned_file(f, {"prompt": "{}"})) + self.assertEqual( + self.api._ingest_scanned_file(f, {"parameters": " \nNegative prompt: x"}), + "found", + ) + self.assertEqual(self.api._ingest_scanned_file(f, meta), "added") + self.assertEqual(self.api._ingest_scanned_file(f, meta), "linked") + + def test_ingest_link_failures_are_tolerated(self): + f = Path("gen.png") + meta = { + "prompt": json.dumps( + {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a dog"}}} + ) + } + self.api.db.link_image_to_prompt = _raise + self.assertEqual(self.api._ingest_scanned_file(f, meta), "added") + self.assertEqual(self.api._ingest_scanned_file(f, meta), "found") + + def test_ingest_save_failure_counts_as_found(self): + f = Path("gen.png") + meta = { + "prompt": json.dumps( + {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a bird"}}} + ) + } + self.api.db.save_prompt = lambda *a, **k: None + self.assertEqual(self.api._ingest_scanned_file(f, meta), "found") + + def test_batch_counts_and_per_file_errors(self): + meta = { + "prompt": json.dumps( + {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a fish"}}} + ) + } + self.api.db.get_prompt_by_hash = _raise + counts = self.api._ingest_scan_batch_sync( + [(Path("a.png"), meta), (Path("b.png"), {})] + ) + self.assertEqual(counts, {"processed": 2, "found": 0, "added": 0, "linked": 0}) + + async def test_scan_internal_error_streams_error_event(self): + GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir)] + self.api._collect_output_media_sync = _raise + body = await (await self.client.request("POST", "/prompt_manager/scan")).text() + self.assertIn('"type": "error"', body) + self.assertNotIn("boom", body) + + if __name__ == "__main__": unittest.main()