diff --git a/database/duplicate_copies.py b/database/duplicate_copies.py index 2baea72b..ee4c07b4 100644 --- a/database/duplicate_copies.py +++ b/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 diff --git a/database/image_collections.py b/database/image_collections.py index 674b3937..43193b0b 100644 --- a/database/image_collections.py +++ b/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 diff --git a/database/operations.py b/database/operations.py index 0f45097f..def1adeb 100644 --- a/database/operations.py +++ b/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, diff --git a/py/api/duplicate_files.py b/py/api/duplicate_files.py index 6edf265d..9cf101f5 100644 --- a/py/api/duplicate_files.py +++ b/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" diff --git a/py/api/duplicate_spellings.py b/py/api/duplicate_spellings.py index 31e0f0b9..3653e5de 100644 --- a/py/api/duplicate_spellings.py +++ b/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"]) diff --git a/tests/test_api_duplicate_files.py b/tests/test_api_duplicate_files.py index 91bd0ea3..a708e229 100644 --- a/tests/test_api_duplicate_files.py +++ b/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): diff --git a/tests/test_image_collections.py b/tests/test_image_collections.py new file mode 100644 index 00000000..9136c862 --- /dev/null +++ b/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()