diff --git a/tests/test_api_backup_restore.py b/tests/test_api_backup_restore.py index 06c100f..b075ea5 100644 --- a/tests/test_api_backup_restore.py +++ b/tests/test_api_backup_restore.py @@ -45,7 +45,7 @@ class BackupRestoreTestCase(AioHTTPTestCase): return app async def tearDownAsync(self): - self.api.db.close() + self.api.db.close_all() for name in os.listdir(self.tmpdir): os.unlink(os.path.join(self.tmpdir, name)) os.rmdir(self.tmpdir) @@ -66,7 +66,7 @@ class BackupRestoreTestCase(AioHTTPTestCase): db = PromptDatabase(path) for text in texts: db.save_prompt(text=text, prompt_hash=generate_prompt_hash(text)) - db.close() + db.close_all() return path async def _restore(self, path, field="database_file"): diff --git a/tests/test_api_images.py b/tests/test_api_images.py index 835d4ba..a080ba6 100644 --- a/tests/test_api_images.py +++ b/tests/test_api_images.py @@ -27,6 +27,7 @@ from database.operations import PromptDatabase # noqa: E402 import py.api.images as images_module # noqa: E402 from py.api import PromptManagerAPI # noqa: E402 from py.api.prompts import MAX_PAGE_LIMIT # noqa: E402 +from py.config import GalleryConfig # noqa: E402 from utils.hashing import generate_prompt_hash # noqa: E402 @@ -46,8 +47,17 @@ class ImageAPITestCase(AioHTTPTestCase): self._temp_db = tempfile.NamedTemporaryFile(delete=False, suffix=".db") self._temp_db.close() - self._output_tmp = tempfile.TemporaryDirectory() - self.output_dir = Path(os.path.realpath(self._output_tmp.name)) + self._output_tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + # A fake ComfyUI tree: /output is the gallery root and is the + # anchor that public (response) paths are rendered relative to. + self.comfy_dir = Path(os.path.realpath(self._output_tmp.name)) + self.output_dir = self.comfy_dir / "output" + self.output_dir.mkdir() + anchors = [os.path.normcase(str(self.comfy_dir))] + self._anchor_patch = patch.object( + GalleryConfig, "path_anchors", classmethod(lambda cls: list(anchors)) + ) + self._anchor_patch.start() app = web.Application() routes = web.RouteTableDef() @@ -61,6 +71,9 @@ class ImageAPITestCase(AioHTTPTestCase): return app async def tearDownAsync(self): + self._anchor_patch.stop() + # Windows refuses to unlink a database that still has open handles. + self.api.db.close_all() for path in ( self._temp_db.name, self._temp_db.name + "-wal", diff --git a/tests/test_api_prompts.py b/tests/test_api_prompts.py index 47fe232..168e38b 100644 --- a/tests/test_api_prompts.py +++ b/tests/test_api_prompts.py @@ -43,6 +43,8 @@ class PromptAPITestCase(AioHTTPTestCase): return app async def tearDownAsync(self): + # Windows refuses to unlink a database that still has open handles. + self.api.db.close_all() for path in ( self._temp_db.name, self._temp_db.name + "-wal", diff --git a/tests/test_database.py b/tests/test_database.py index 61eaec7..5440f47 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -31,7 +31,7 @@ class DatabaseTestCase(unittest.TestCase): self.db = PromptDatabase(self.temp_db.name) def tearDown(self): - self.db.close() + self.db.close_all() for suffix in ("", "-wal", "-shm"): path = self.temp_db.name + suffix if os.path.exists(path): @@ -341,14 +341,17 @@ class TestPagination(DatabaseTestCase): # Use a fresh empty DB empty_db_file = tempfile.NamedTemporaryFile(delete=False, suffix=".db") empty_db_file.close() + empty_db = PromptDatabase(empty_db_file.name) try: - empty_db = PromptDatabase(empty_db_file.name) result = empty_db.get_recent_prompts(limit=10, offset=0) self.assertEqual(result["total"], 0) self.assertEqual(len(result["prompts"]), 0) self.assertFalse(result["has_more"]) finally: - os.unlink(empty_db_file.name) + empty_db.close_all() + for suffix in ("", "-wal", "-shm"): + if os.path.exists(empty_db_file.name + suffix): + os.unlink(empty_db_file.name + suffix) def test_total_count_is_integer(self): result = self.db.get_recent_prompts(limit=5) diff --git a/tests/test_database_backup.py b/tests/test_database_backup.py index c2bff76..79f31b4 100644 --- a/tests/test_database_backup.py +++ b/tests/test_database_backup.py @@ -45,7 +45,7 @@ class BackupTestCase(unittest.TestCase): self.addCleanup(self._cleanup) def _cleanup(self): - self.db.close() + self.db.close_all() for name in os.listdir(self.tmpdir): os.unlink(os.path.join(self.tmpdir, name)) os.rmdir(self.tmpdir) @@ -59,7 +59,7 @@ class BackupTestCase(unittest.TestCase): src = PromptDatabase(src_path) for text in texts: self._save(src, text) - src.close() + src.close_all() return src_path @@ -230,7 +230,7 @@ class TestRestoreFromFile(BackupTestCase): def test_restore_without_a_live_database_returns_empty_backup_path(self): src = self._make_source_db(["fresh"]) - self.db.close() + self.db.close_all() os.unlink(self.path) for suffix in ("-wal", "-shm"): if os.path.exists(self.path + suffix): diff --git a/tests/test_database_concurrency.py b/tests/test_database_concurrency.py index 0f39c6d..4b5cedb 100644 --- a/tests/test_database_concurrency.py +++ b/tests/test_database_concurrency.py @@ -29,7 +29,7 @@ class ConcurrencyTestCase(unittest.TestCase): self.addCleanup(self._cleanup) def _cleanup(self): - self.db.close() + self.db.close_all() for suffix in ("", "-wal", "-shm"): if os.path.exists(self.path + suffix): os.unlink(self.path + suffix) diff --git a/tests/test_database_migrations.py b/tests/test_database_migrations.py index 9b3283b..467aa69 100644 --- a/tests/test_database_migrations.py +++ b/tests/test_database_migrations.py @@ -42,7 +42,7 @@ class MigrationTestCase(unittest.TestCase): def _cleanup(self): for db in self.dbs: - db.close() + db.close_all() PromptModel.reset_schema_cache() for name in os.listdir(self.tmpdir): os.unlink(os.path.join(self.tmpdir, name)) @@ -105,7 +105,7 @@ class TestSchemaInitOncePerProcess(MigrationTestCase): def test_deleted_database_file_is_recreated_on_next_construction(self): db = self._open() - db.close() + db.close_all() os.unlink(self.path) for suffix in ("-wal", "-shm"): if os.path.exists(self.path + suffix): @@ -118,7 +118,7 @@ class TestSchemaInitOncePerProcess(MigrationTestCase): def test_close_then_unlink_succeeds(self): db = self._open() db.save_prompt(text="x", prompt_hash="x") - db.close() + db.close_all() os.unlink(self.path) self.assertFalse(os.path.exists(self.path)) diff --git a/tests/test_prompt_usage.py b/tests/test_prompt_usage.py index d96aa87..fd5fa19 100644 --- a/tests/test_prompt_usage.py +++ b/tests/test_prompt_usage.py @@ -22,7 +22,7 @@ class UsageTestCase(unittest.TestCase): def tearDown(self): if getattr(self, "db", None) is not None: - self.db.close() + self.db.close_all() for suffix in ("", "-wal", "-shm"): if os.path.exists(self.path + suffix): os.unlink(self.path + suffix) @@ -164,9 +164,11 @@ class TestUsageMigration(unittest.TestCase): def setUp(self): fd, self.path = tempfile.mkstemp(suffix=".db") os.close(fd) - db = PromptDatabase(self.path) + self.dbs = [] + db = self._open() self.with_images = db.save_prompt(text="has images", prompt_hash="h1") self.no_images = db.save_prompt(text="no images", prompt_hash="h2") + db.close_all() with sqlite3.connect(self.path) as conn: conn.execute( "UPDATE prompts SET created_at = '2026-01-01T10:00:00.000000+00:00'" @@ -188,15 +190,21 @@ class TestUsageMigration(unittest.TestCase): # must look at the file again to see the downgraded schema. PromptModel.reset_schema_cache() + def _open(self): + """Open the database under test and close it again in tearDown.""" + db = PromptDatabase(self.path) + self.dbs.append(db) + return db + def tearDown(self): - if getattr(self, "db", None) is not None: - self.db.close() + for db in self.dbs: + db.close_all() for suffix in ("", "-wal", "-shm"): if os.path.exists(self.path + suffix): os.unlink(self.path + suffix) def test_backfill_estimates_usage_from_linked_images(self): - db = PromptDatabase(self.path) + db = self._open() used = db.get_prompt_by_id(self.with_images) unused = db.get_prompt_by_id(self.no_images) @@ -206,7 +214,7 @@ class TestUsageMigration(unittest.TestCase): self.assertTrue(unused["last_used_at"].startswith("2026-01-01T10:00:00")) def test_backfill_runs_only_when_the_columns_are_added(self): - db = PromptDatabase(self.path) + db = self._open() # A row whose usage is unknown (e.g. written by 3.2.3 after a downgrade) # stays "never run": a restart must not invent a run for it, otherwise # Recently Used could never tell saved-but-unused prompts from run ones. @@ -221,14 +229,14 @@ class TestUsageMigration(unittest.TestCase): self.assertIsNone(db.get_prompt_by_id(legacy_id)["last_used_at"]) PromptModel.reset_schema_cache() - later = PromptDatabase(self.path).get_prompt_by_id(legacy_id) + later = self._open().get_prompt_by_id(legacy_id) self.assertIsNone(later["last_used_at"]) self.assertEqual(later["run_count"], 0) def test_migration_runs_once(self): - db = PromptDatabase(self.path) + db = self._open() db.record_prompt_use(self.no_images) - reopened = PromptDatabase(self.path) + reopened = self._open() self.assertEqual(reopened.get_prompt_by_id(self.no_images)["run_count"], 2)