feat(collections): add, remove and rename collections and keep members through duplicate cleanup

database/duplicate_copies.py

@@ -9,6 +9,8 @@ IMMEDIATE`` transaction, so no row ever names a deleted file:
   already links the kept copy: then its saved-pair links join that record
   and it is deleted;
 - its favorite;
+- its collection memberships (#331), joined onto the kept copy's as their
+  union under the same keys a favorite moves by;
 - its image tags (``media_image_tags`` of its byte record), joined onto the
   kept copy's byte record as their union, in order;
 - its pending Needs review items, settled: only an item still ``pending`` at
@@ -50,6 +52,15 @@ INSERT_FAVORITE_SQL = "INSERT OR IGNORE INTO image_favorites VALUES (?, ?, ?)"
 DELETE_FAVORITE_SQL = (
     "DELETE FROM image_favorites WHERE root_key = ? AND relative_path = ?"
 )
+COPY_MEMBERSHIPS_SQL = (
+    "INSERT OR IGNORE INTO image_collection_items "
+    "(collection_id, root_key, relative_path, added_at) "
+    "SELECT collection_id, ?, ?, added_at FROM image_collection_items "
+    "WHERE root_key = ? AND relative_path = ?"
+)
+DELETE_MEMBERSHIPS_SQL = (
+    "DELETE FROM image_collection_items WHERE root_key = ? AND relative_path = ?"
+)
 COPY_BYTE_SQL = (
     f"SELECT b.id FROM {FILES_TABLE} f JOIN {BYTES_TABLE} b "  # nosec B608
     "ON b.file_id=f.id WHERE f.locator=? AND b.sha256=?"
@@ -73,6 +84,8 @@ FILE_LINK_SQL = (
     DELETE_LINK_SQL,
     FAVORITE_SQL,
     DELETE_FAVORITE_SQL,
+    COPY_MEMBERSHIPS_SQL,
+    DELETE_MEMBERSHIPS_SQL,
     COPY_BYTE_SQL,
     COPY_TAG_COUNT_SQL,
     PENDING_COUNT_SQL,
@@ -135,12 +148,15 @@ class DuplicateCopyOperations:
         by), ``targets`` (``{removed spelling: kept spelling}``, the spelling
         a moved record takes; else ``kept_path``), ``favorites`` (``(root key,
         removed key, kept root key, kept key)`` per listed root holding the
-        file), ``removed_locator``,
-        ``kept_locator``, ``sha256`` and the kept copy's ``kept_size`` and
-        ``kept_mtime_ns``. Returns ``{moved, merged, favorites, tags,
-        review}``.
+        file; they move its favorite and its collection memberships),
+        ``removed_locator``, ``kept_locator``, ``sha256`` and the kept copy's
+        ``kept_size`` and ``kept_mtime_ns``. Returns ``{moved, merged,
+        favorites, tags, review, collections}`` (``collections``: the
+        memberships inserted on the kept copy).
         """
-        counts = dict.fromkeys(("moved", "merged", "favorites", "tags", "review"), 0)
+        counts = dict.fromkeys(
+            ("moved", "merged", "favorites", "tags", "review", "collections"), 0
+        )
         kept_keys = [normalize_image_path(path) for path in move["kept"]]
         with self._pair_transaction() as conn:
             for spelled in dict.fromkeys(move["removed"]):
@@ -173,6 +189,7 @@ class DuplicateCopyOperations:
                     counts["moved"] += 1
             for favorite in move.get("favorites", ()):
                 counts["favorites"] += _move_favorite(conn, *favorite)
+                counts["collections"] += _move_memberships(conn, *favorite)
             counts["tags"] = self._move_copy_tags(conn, move)
             for (item_id,) in conn.execute(
                 _PENDING_OF_LOCATOR, (move["removed_locator"],)
@@ -213,3 +230,12 @@ def _move_favorite(conn, root_key, removed_key, kept_root, kept_key):
     conn.execute(INSERT_FAVORITE_SQL, (kept_root, kept_key, row[0]))
     conn.execute(DELETE_FAVORITE_SQL, (root_key, removed_key))
     return 1
+
+
+def _move_memberships(conn, root_key, removed_key, kept_root, kept_key):
+    """Copy a removed copy's memberships to the kept copy; the rows inserted."""
+    inserted = conn.execute(
+        COPY_MEMBERSHIPS_SQL, (kept_root, kept_key, root_key, removed_key)
+    ).rowcount
+    conn.execute(DELETE_MEMBERSHIPS_SQL, (root_key, removed_key))
+    return inserted

database/image_collections.py

@@ -229,3 +229,231 @@ def migrate_image_collections(conn: sqlite3.Connection) -> None:
         conn.execute("ROLLBACK TO image_collections_schema")
         conn.execute("RELEASE image_collections_schema")
         raise
+
+
+class CollectionMissing(LookupError):
+    """No collection has the id (a 404 ``collection-not-found``)."""
+
+
+class CollectionNameTaken(ValueError):
+    """Another collection already has the name's key (a 409 ``name-taken``)."""
+
+    def __init__(self, collection_id: int):
+        super().__init__("Another collection has this name")
+        self.collection_id = collection_id
+
+
+class CollectionFilesMissing(LookupError):
+    """A new name with nothing to insert (a 404 ``files-not-found``)."""
+
+
+LIST_SQL = f"SELECT id, name FROM {COLLECTIONS_TABLE} ORDER BY name_key, id"
+GET_SQL = f"SELECT id, name FROM {COLLECTIONS_TABLE} WHERE id = ?"
+BY_KEY_SQL = f"SELECT id, name FROM {COLLECTIONS_TABLE} WHERE name_key = ?"
+MEMBERS_SQL = (
+    f"SELECT relative_path FROM {ITEMS_TABLE} "
+    "WHERE collection_id = ? AND root_key = ?"
+)
+ROOT_MEMBERS_SQL = (
+    f"SELECT collection_id, relative_path FROM {ITEMS_TABLE} WHERE root_key = ?"
+)
+COUNTS_SQL = (
+    f"SELECT c.id, (SELECT COUNT(*) FROM {ITEMS_TABLE} i "
+    f"WHERE i.collection_id = c.id) FROM {COLLECTIONS_TABLE} c"
+)
+FILE_SQL = (
+    f"SELECT collection_id FROM {ITEMS_TABLE} "
+    "WHERE root_key = ? AND relative_path = ? ORDER BY collection_id"
+)
+# Every read, for the EXPLAIN QUERY PLAN check: none scans the items table and
+# only the two whole-list reads (LIST_SQL, COUNTS_SQL) scan the collections.
+READ_SQL = (
+    LIST_SQL,
+    GET_SQL,
+    BY_KEY_SQL,
+    MEMBERS_SQL,
+    ROOT_MEMBERS_SQL,
+    COUNTS_SQL,
+    FILE_SQL,
+)
+_INSERT_ITEM_SQL = f"INSERT OR IGNORE INTO {ITEMS_TABLE} VALUES (?, ?, ?, ?)"
+_DELETE_ITEM_SQL = (
+    f"DELETE FROM {ITEMS_TABLE} "
+    "WHERE collection_id = ? AND root_key = ? AND relative_path = ?"
+)
+
+
+def _collection_id(value: object) -> int:
+    if type(value) is not int or not 1 <= value <= 999_999_999:
+        raise CollectionInvalid("Collection id must be a positive integer")
+    return value
+
+
+def _items(items: object) -> list:
+    """Distinct ``(root key, relative key)`` pairs, in order."""
+    if not isinstance(items, (list, tuple)):
+        raise CollectionInvalid("Items must be a list")
+    found = []
+    for item in items:
+        if (
+            not isinstance(item, (list, tuple))
+            or len(item) != 2
+            or not all(
+                isinstance(part, str) and 1 <= len(part) <= MAX_KEY_LENGTH
+                for part in item
+            )
+        ):
+            raise CollectionInvalid("Each item is a root key and a relative key")
+        found.append((item[0], item[1]))
+    return list(dict.fromkeys(found))
+
+
+def _row(row) -> dict:
+    return {"id": row[0], "name": row[1]}
+
+
+class ImageCollectionOperations:
+    """PromptDatabase mixin: collections and their members.
+
+    Each write runs in its own ``BEGIN IMMEDIATE`` transaction
+    (``_pair_transaction``), so two tabs creating one name are serialized.
+    """
+
+    def list_collections(self) -> list[dict]:
+        """Every collection ``{id, name}`` in name-key order, then id."""
+        rows = self.model.get_connection().execute(LIST_SQL)
+        return [_row(row) for row in rows]
+
+    def get_collection(self, collection_id: int) -> Optional[dict]:
+        """``{id, name}`` of one collection, or None."""
+        row = (
+            self.model.get_connection()
+            .execute(GET_SQL, (_collection_id(collection_id),))
+            .fetchone()
+        )
+        return _row(row) if row else None
+
+    def collection_exists(self, collection_id: int) -> bool:
+        return self.get_collection(collection_id) is not None
+
+    def collection_members(self, collection_id: int, root_key: str) -> set[str]:
+        """The relative keys stored for one collection under one source."""
+        rows = self.model.get_connection().execute(
+            MEMBERS_SQL, (_collection_id(collection_id), root_key)
+        )
+        return {path for (path,) in rows}
+
+    def root_collection_members(self, root_key: str) -> dict[int, set[str]]:
+        """``{collection id: relative keys}`` stored under one source."""
+        found: dict[int, set[str]] = {}
+        for collection_id, path in self.model.get_connection().execute(
+            ROOT_MEMBERS_SQL, (root_key,)
+        ):
+            found.setdefault(collection_id, set()).add(path)
+        return found
+
+    def collection_counts(self) -> dict[int, int]:
+        """``{collection id: stored rows}`` for every collection."""
+        rows = self.model.get_connection().execute(COUNTS_SQL)
+        return {collection_id: count for collection_id, count in rows}
+
+    def file_collections(self, root_key: str, relative_path: str) -> list[int]:
+        """Ids of the collections holding one file."""
+        rows = self.model.get_connection().execute(FILE_SQL, (root_key, relative_path))
+        return [collection_id for (collection_id,) in rows]
+
+    def add_to_collection(
+        self, *, items: object, collection_id: object = None, name: object = None
+    ) -> dict:
+        """Add files to an existing id or a name; create the name when needed.
+
+        A name whose key exists uses that collection; otherwise a collection
+        is created only when at least one item is inserted. Returns
+        ``{collection, created, added, already}``.
+        """
+        from .models import utc_now_iso  # models imports this module's schema
+
+        if (collection_id is None) == (name is None):
+            raise CollectionInvalid("Give either a collection id or a name")
+        pairs = _items(items)
+        if name is not None:
+            spelled = normalize_collection_name(name)
+        else:
+            collection_id = _collection_id(collection_id)
+        now = utc_now_iso()
+        with self._pair_transaction() as conn:
+            created = False
+            if name is not None:
+                row = conn.execute(
+                    BY_KEY_SQL, (collection_name_key(spelled),)
+                ).fetchone()
+                if row is None:
+                    if not pairs:
+                        raise CollectionFilesMissing("Nothing to add")
+                    cursor = conn.execute(
+                        f"INSERT INTO {COLLECTIONS_TABLE} "  # nosec B608
+                        "(name, name_key, created_at) VALUES (?, ?, ?)",
+                        (spelled, collection_name_key(spelled), now),
+                    )
+                    row, created = (cursor.lastrowid, spelled), True
+            else:
+                row = conn.execute(GET_SQL, (collection_id,)).fetchone()
+                if row is None:
+                    raise CollectionMissing("No such collection")
+            collection = _row(row)
+            added = sum(
+                conn.execute(
+                    _INSERT_ITEM_SQL, (collection["id"], root, path, now)
+                ).rowcount
+                for root, path in pairs
+            )
+        return {
+            "collection": collection,
+            "created": created,
+            "added": added,
+            "already": len(pairs) - added,
+        }
+
+    def remove_from_collection(self, collection_id: object, items: object) -> dict:
+        """Remove files from one collection; ``{collection, removed}``."""
+        collection_id = _collection_id(collection_id)
+        pairs = _items(items)
+        with self._pair_transaction() as conn:
+            row = conn.execute(GET_SQL, (collection_id,)).fetchone()
+            if row is None:
+                raise CollectionMissing("No such collection")
+            removed = sum(
+                conn.execute(_DELETE_ITEM_SQL, (collection_id, root, path)).rowcount
+                for root, path in pairs
+            )
+        return {"collection": _row(row), "removed": removed}
+
+    def rename_collection(self, collection_id: object, name: object) -> dict:
+        """The renamed ``{id, name}``; a case variant of its own name is fine."""
+        collection_id = _collection_id(collection_id)
+        spelled = normalize_collection_name(name)
+        key = collection_name_key(spelled)
+        with self._pair_transaction() as conn:
+            if conn.execute(GET_SQL, (collection_id,)).fetchone() is None:
+                raise CollectionMissing("No such collection")
+            taken = conn.execute(BY_KEY_SQL, (key,)).fetchone()
+            if taken is not None and taken[0] != collection_id:
+                raise CollectionNameTaken(taken[0])
+            conn.execute(
+                f"UPDATE {COLLECTIONS_TABLE} SET name = ?, name_key = ? "  # nosec B608
+                "WHERE id = ?",
+                (spelled, key, collection_id),
+            )
+        return {"id": collection_id, "name": spelled}
+
+    def delete_collection(self, collection_id: object) -> bool:
+        """Delete one collection and its members (never a file). True."""
+        collection_id = _collection_id(collection_id)
+        with self._pair_transaction() as conn:
+            cursor = conn.execute(
+                f"DELETE FROM {COLLECTIONS_TABLE} WHERE id = ?",  # nosec B608
+                (collection_id,),
+            )
+            if cursor.rowcount == 0:
+                raise CollectionMissing("No such collection")
+        return True

database/operations.py

@@ -13,6 +13,7 @@ from .models import PromptModel, normalize_image_path, utc_now_iso
 from .pairs import PairDependencyError, PairOperations  # noqa: F401 (re-exported)
 from .library import LibraryOperations, titled
 from .image_favorites import ImageFavoriteOperations
+from .image_collections import ImageCollectionOperations
 from .annotations import AnnotationOperations
 from .review import ReviewOperations
 from .review_decisions import ReviewDecisionOperations
@@ -359,6 +360,7 @@ class PromptDatabase(
     ExecutionOutputOperations,
     HistoryReadOperations,
     ImageFavoriteOperations,
+    ImageCollectionOperations,
     AnnotationOperations,
     ReviewOperations,
     ReviewDecisionOperations,

py/api/duplicate_files.py

@@ -588,7 +588,9 @@ class DuplicateFilesMixin:
         job.update(total=len(targets), status_line="Removing duplicate files")
         spellings = await self._run_in_executor(self._source_spellings)
         removed, failure, keepers = 0, None, {}
-        totals = dict.fromkeys(("moved", "merged", "favorites", "tags", "review"), 0)
+        totals = dict.fromkeys(
+            ("moved", "merged", "favorites", "tags", "review", "collections"), 0
+        )
         for done, target in enumerate(targets, start=1):
             if failure is not None:
                 self._record_target(job, "kept", target, NOT_ATTEMPTED)
@@ -631,6 +633,11 @@ class DuplicateFilesMixin:
             job.log_line(
                 f"Moved {_plural(totals['favorites'], 'favorite')} to the kept copies"
             )
+        if totals["collections"]:
+            job.log_line(
+                f"Moved {_plural(totals['collections'], 'collection membership')} "
+                "to the kept copies"
+            )
         if totals["tags"]:
             job.log_line(
                 f"Added {_plural(totals['tags'], 'image tag')} to the kept copies"

py/api/duplicate_spellings.py

@@ -95,7 +95,8 @@ def copy_move(target, kept_stat, spellings):
 
     A record found under a folder spelling moves to the kept copy under
     the same spelling when that folder holds it too, else to its real
-    path. Its favorite moves under every listed root holding the file.
+    path. Its favorite and its collection memberships move under every
+    listed root holding the file.
     """
     removed = spellings.of(target["abs"])
     kept = spellings.of(target["keeper_abs"])

tests/test_api_duplicate_files.py

@@ -687,6 +687,39 @@ class TestCleanupJob(DuplicateFilesTestCase):
             "Moved 2 database links to the kept copies", "\n".join(cleanup.log)
         )
 
+    async def test_duplicate_removal_moves_collection_memberships(self):
+        self.trio()
+        db = self.api.db
+        root_key = duplicate_files.favorite_root_key(self.output_dir)
+        grids = db.add_to_collection(name="LoRA grids", items=[(root_key, "b.png")])
+        grids = grids["collection"]["id"]
+        other = db.add_to_collection(
+            name="Other", items=[(root_key, "sub/c.png"), (root_key, "d.png")]
+        )["collection"]["id"]
+        db.add_to_collection(collection_id=grids, items=[(root_key, "sub/c.png")])
+        _plan, cleanup = await self.remove_all_but("a.png")
+        # The union: a.png joins both; c.png's second grids row was a repeat.
+        self.assertEqual(db.collection_members(grids, root_key), {"a.png"})
+        self.assertEqual(db.collection_members(other, root_key), {"a.png", "d.png"})
+        self.assertEqual(sorted(db.file_collections(root_key, "a.png")), [grids, other])
+        self.assertEqual(db.file_collections(root_key, "b.png"), [])
+        self.assertEqual(db.file_collections(root_key, "sub/c.png"), [])
+        self.assertIn(
+            "Moved 2 collection memberships to the kept copies", "\n".join(cleanup.log)
+        )
+
+    def test_one_collection_membership_is_singular(self):
+        job = type(
+            "Job", (), {"log": [], "log_line": lambda s, line: s.log.append(line)}
+        )()
+        totals = dict.fromkeys(
+            ("moved", "merged", "favorites", "tags", "review", "collections"), 0
+        )
+        duplicate_files.DuplicateFilesMixin._log_moved_records(
+            job, {**totals, "collections": 1}
+        )
+        self.assertEqual(job.log, ["Moved 1 collection membership to the kept copies"])
+
     async def test_a_failed_removal_leaves_its_links_on_the_kept_copy(self):
         self.trio()
         prompt = self.save("Stays on c")
@@ -843,8 +876,11 @@ class TestLinkQueries(DuplicateFilesTestCase):
                 with self.subTest(sql=sql):
                     self.assertNotIn("SCAN generated_images", plan)
                     self.assertNotIn("SCAN image_favorites", plan)
+                    self.assertNotIn("SCAN image_collection_items", plan)
         finally:
             conn.close()
+        statements = duplicate_files.query_plan_sql()
+        self.assertTrue(any("image_collection_items" in sql for sql in statements))
 
 
 class TestReviewHelpers(DuplicateFilesTestCase):

tests/test_image_collections.py

@@ -0,0 +1,223 @@
+"""Collections (#331): the database operations."""
+
+import sqlite3
+import tempfile
+import unittest
+from pathlib import Path
+
+try:
+    from tests.open_handles import assert_closed
+except ImportError:  # discovered with tests/ as the top-level directory
+    from open_handles import assert_closed
+
+from database.models import PromptModel
+from database.operations import PromptDatabase
+
+try:  # guarded so a missing module fails assertions, not test loading
+    from database.image_collections import (
+        CollectionFilesMissing,
+        CollectionInvalid,
+        CollectionMissing,
+        CollectionNameTaken,
+        READ_SQL,
+    )
+except ImportError:
+    CollectionFilesMissing = CollectionMissing = CollectionNameTaken = None
+    CollectionInvalid = ValueError
+    READ_SQL = None
+
+ROOT = "/real/output"
+OTHER = "/real/other"
+
+
+class ImageCollectionOperationsTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.addCleanup(assert_closed, self, self.temp.name)
+        self.addCleanup(PromptModel.reset_schema_cache)
+        self.db = PromptDatabase(str(Path(self.temp.name) / "prompts.db"))
+        self.addCleanup(self.db.close_all)
+
+    def require(self):
+        self.assertTrue(
+            hasattr(self.db, "add_to_collection"),
+            "ImageCollectionOperations is not implemented",
+        )
+
+    def add(self, items, **target):
+        return self.db.add_to_collection(items=items, **target)
+
+    def test_add_creates_a_collection_by_name(self):
+        self.require()
+        result = self.add(
+            [(ROOT, "a.png"), (ROOT, "b.png"), (OTHER, "a.png")], name="LoRa testing"
+        )
+        collection = result["collection"]
+        self.assertEqual(collection["name"], "LoRa testing")
+        self.assertEqual(
+            result,
+            {"collection": collection, "created": True, "added": 3, "already": 0},
+        )
+        self.assertEqual(self.db.list_collections(), [collection])
+        self.assertTrue(self.db.collection_exists(collection["id"]))
+        self.assertEqual(self.db.get_collection(collection["id"]), collection)
+        self.assertEqual(
+            self.db.collection_members(collection["id"], ROOT), {"a.png", "b.png"}
+        )
+        self.assertEqual(
+            self.db.root_collection_members(OTHER), {collection["id"]: {"a.png"}}
+        )
+        self.assertEqual(self.db.collection_counts(), {collection["id"]: 3})
+        self.assertEqual(self.db.file_collections(ROOT, "a.png"), [collection["id"]])
+        self.assertEqual(self.db.file_collections(ROOT, "zzz.png"), [])
+
+    def test_add_by_name_ignores_case_and_unicode_form(self):
+        self.require()
+        first = self.add([(ROOT, "a.png")], name="Café grids")["collection"]
+        # NFD spelling (macOS) in another case joins the same collection.
+        second = self.add([(ROOT, "b.png")], name="CAFÉ  GRIDS")
+        self.assertEqual(second["collection"], first)
+        self.assertFalse(second["created"])
+        self.assertEqual(len(self.db.list_collections()), 1)
+        self.assertEqual(
+            self.db.collection_members(first["id"], ROOT), {"a.png", "b.png"}
+        )
+
+    def test_add_counts_already_present(self):
+        self.require()
+        made = self.add([(ROOT, "a.png"), (ROOT, "b.png")], name="Set")
+        result = self.add(
+            [(ROOT, "b.png"), (ROOT, "c.png"), (ROOT, "c.png")],
+            collection_id=made["collection"]["id"],
+        )
+        self.assertEqual(
+            (result["created"], result["added"], result["already"]), (False, 1, 1)
+        )
+        self.assertEqual(self.db.collection_counts(), {made["collection"]["id"]: 3})
+
+    def test_add_with_a_new_name_and_nothing_inserted_creates_nothing(self):
+        self.require()
+        with self.assertRaises(CollectionFilesMissing):
+            self.add([], name="Empty")
+        self.assertEqual(self.db.list_collections(), [])
+        # An existing name with nothing to add is not an error.
+        made = self.add([(ROOT, "a.png")], name="Kept")["collection"]
+        self.assertEqual(
+            self.add([], name="kept"),
+            {"collection": made, "created": False, "added": 0, "already": 0},
+        )
+
+    def test_add_to_a_deleted_collection_raises_missing(self):
+        self.require()
+        made = self.add([(ROOT, "a.png")], name="Gone")["collection"]
+        self.db.delete_collection(made["id"])
+        with self.assertRaises(CollectionMissing):
+            self.add([(ROOT, "b.png")], collection_id=made["id"])
+        with self.assertRaises(CollectionMissing):
+            self.db.remove_from_collection(made["id"], [(ROOT, "a.png")])
+        with self.assertRaises(CollectionMissing):
+            self.db.rename_collection(made["id"], "Back")
+        self.assertFalse(self.db.collection_exists(made["id"]))
+        self.assertIsNone(self.db.get_collection(made["id"]))
+
+    def test_remove_returns_the_count(self):
+        self.require()
+        made = self.add([(ROOT, "a.png"), (ROOT, "b.png")], name="Set")["collection"]
+        result = self.db.remove_from_collection(
+            made["id"], [(ROOT, "a.png"), (ROOT, "a.png"), (ROOT, "never.png")]
+        )
+        self.assertEqual(result, {"collection": made, "removed": 1})
+        self.assertEqual(self.db.collection_members(made["id"], ROOT), {"b.png"})
+        # Removing the last member keeps the (now empty) collection.
+        self.db.remove_from_collection(made["id"], [(ROOT, "b.png")])
+        self.assertEqual(self.db.list_collections(), [made])
+        self.assertEqual(self.db.collection_counts(), {made["id"]: 0})
+
+    def test_rename_refuses_a_taken_name(self):
+        self.require()
+        first = self.add([(ROOT, "a.png")], name="First")["collection"]
+        second = self.add([(ROOT, "a.png")], name="Second")["collection"]
+        with self.assertRaises(CollectionNameTaken) as caught:
+            self.db.rename_collection(second["id"], " first ")
+        self.assertEqual(caught.exception.collection_id, first["id"])
+        self.assertEqual(self.db.get_collection(second["id"]), second)
+
+    def test_rename_accepts_a_case_variant_of_its_own_name(self):
+        self.require()
+        made = self.add([(ROOT, "a.png")], name="lora grids")["collection"]
+        renamed = self.db.rename_collection(made["id"], "LoRA   grids ")
+        self.assertEqual(renamed, {"id": made["id"], "name": "LoRA grids"})
+        self.assertEqual(self.db.list_collections(), [renamed])
+        other = self.db.rename_collection(made["id"], "Portraits")
+        self.assertEqual(other["name"], "Portraits")
+
+    def test_delete_unknown_raises_missing(self):
+        self.require()
+        made = self.add([(ROOT, "a.png")], name="Doomed")["collection"]
+        self.assertTrue(self.db.delete_collection(made["id"]))
+        self.assertEqual(self.db.root_collection_members(ROOT), {})
+        with self.assertRaises(CollectionMissing):
+            self.db.delete_collection(made["id"])
+        with self.assertRaises(CollectionMissing):
+            self.db.delete_collection(999)
+
+    def test_list_is_in_name_key_order(self):
+        self.require()
+        for name in ("beta", "Alpha", "été", "Zeta", "alpha two"):
+            self.add([(ROOT, "a.png")], name=name)
+        self.assertEqual(
+            [row["name"] for row in self.db.list_collections()],
+            ["Alpha", "alpha two", "beta", "Zeta", "été"],
+        )
+
+    def test_names_are_trimmed_and_collapsed(self):
+        self.require()
+        made = self.add([(ROOT, "a.png")], name="  LoRa \t testing\n ")["collection"]
+        self.assertEqual(made["name"], "LoRa testing")
+        for bad in ("", "   ", "x" * 101, "bell\x07", 5, None):
+            with self.subTest(name=bad):
+                with self.assertRaises(CollectionInvalid):
+                    self.add([(ROOT, "a.png")], name=bad)
+        self.assertEqual(len(self.db.list_collections()), 1)
+
+    def test_targets_and_items_are_validated(self):
+        self.require()
+        made = self.add([(ROOT, "a.png")], name="Set")["collection"]
+        for target in (
+            {},
+            {"name": "x", "collection_id": made["id"]},
+            {"collection_id": True},
+            {"collection_id": 0},
+            {"collection_id": "1"},
+        ):
+            with self.subTest(target=target):
+                with self.assertRaises(CollectionInvalid):
+                    self.add([(ROOT, "a.png")], **target)
+        for items in ([("root",)], [(ROOT, 5)], "a.png", [(ROOT, "x" * 4097)]):
+            with self.subTest(items=items):
+                with self.assertRaises(CollectionInvalid):
+                    self.add(items, collection_id=made["id"])
+
+    def test_reads_use_an_index(self):
+        self.require()
+        self.add([(ROOT, "a.png")], name="Set")
+        conn = sqlite3.connect(self.db.model.db_path)
+        try:
+            for sql in READ_SQL:
+                plan = " ".join(
+                    row[3]
+                    for row in conn.execute(
+                        "EXPLAIN QUERY PLAN " + sql, [None] * sql.count("?")
+                    )
+                )
+                with self.subTest(sql=sql):
+                    self.assertNotIn("SCAN image_collection_items", plan)
+                    if "?" in sql:  # a lookup, not the whole list
+                        self.assertNotIn("SCAN image_collections", plan)
+        finally:
+            conn.close()
+
+
+if __name__ == "__main__":
+    unittest.main()