nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(library): delete a prompt in a saved pair permanently, with its own pairs (#482, server)

Browse Source 
The Delete review's preview now marks which archive ids may be deleted
permanently (in a saved pair, and nobody else's negative), with a review
digest of the prompt and its pairs and the counts the review shows. The
write accepts an optional delete_paired_ids with their digests. Each is
deleted with its own saved pairs (image-pair links, then pairs, then the
prompt as #443 deletes it) only while its digest is unchanged. Any drift
archives it and counts it changed. Paired deletes are backed up and
disk-checked like deletes.
This commit is contained in:
Vito Sansevero
2026-10-29 12:24:57 -07:00
parent 3e2339fe96
commit e47f05e6be
4 changed files with 495 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
database/library_archive.py
+155 -7
View File
@@ -13,6 +13,9 @@ and delete lists, and deletes only reviewed deletes the route backed up
(``candidates``) whose fingerprint is unchanged since that backup.
"""
import hashlib
import json
import re
import sqlite3
from typing import Iterable
@@ -39,8 +42,15 @@ _PAIRED = (
"EXISTS(SELECT 1 FROM prompt_pairs pp "
"WHERE pp.positive_prompt_id=p.id OR pp.negative_prompt_id=p.id)"
)
# #482: used as the negative of another positive's pair (archived or not):
# never deleted, so that positive's variant stays (archive only).
_NEGATIVE_OF_OTHER = (
"EXISTS(SELECT 1 FROM prompt_pairs op "
"WHERE op.negative_prompt_id=p.id AND op.positive_prompt_id!=p.id)"
)
# Classes in the order a prompt is tested; the first that holds wins.
CLASSES = ("not_found", "protected", "already_archived", "archive", "delete")
_DIGEST = re.compile(r"^[0-9a-f]{64}$")
def _batches(ids: list) -> Iterable[list]:
@@ -78,24 +88,31 @@ class LibraryArchiveOperations:
def _classify_library_delete(
self, conn: sqlite3.Connection, prompt_ids: list[int]
) -> dict:
found = {}
found, deletable = {}, set()
for chunk in _batches(prompt_ids):
marks = ",".join("?" for _ in chunk)
for row in conn.execute(
f"SELECT p.id,{_PROTECTED},{_ARCHIVED},{_PAIRED} " # nosec B608
f"SELECT p.id,{_PROTECTED},{_ARCHIVED},{_PAIRED}," # nosec B608
f"{_NEGATIVE_OF_OTHER} "
f"FROM prompts p WHERE p.id IN ({marks}) AND {_ELIGIBLE}",
chunk,
):
protected, archived, paired = row[1], row[2], row[3]
protected, archived, paired, negative_of_other = row[1:5]
if protected:
found[row[0]] = "protected"
elif archived:
found[row[0]] = "already_archived"
elif paired:
found[row[0]] = "archive"
if not negative_of_other:
deletable.add(row[0])
else:
found[row[0]] = "archive" if paired else "delete"
found[row[0]] = "delete"
classes = {name: [] for name in CLASSES}
for pid in prompt_ids:
classes[found.get(pid, "not_found")].append(pid)
# #482: the archive ids Delete permanently may delete with their pairs.
classes["deletable"] = [pid for pid in classes["archive"] if pid in deletable]
return classes
@staticmethod
@@ -137,6 +154,82 @@ class LibraryArchiveOperations:
]
return prints
@staticmethod
def _distinct_images(conn: sqlite3.Connection, prompt_ids: list[int]) -> int:
"""Image rows of ``prompt_ids`` (each image once; #482 deletes them)."""
return LibraryArchiveOperations._image_links(conn, prompt_ids)
@staticmethod
def _pair_prints(conn: sqlite3.Connection, prompt_ids: list[int]) -> dict:
"""#482: each prompt's own saved pairs, as the review saw them.
Per pair (by id): its negative, archived flag, metadata, usage, ordered
tag ids and its image-pair links (count and newest id).
"""
prints = {pid: [] for pid in prompt_ids}
for chunk in _batches(prompt_ids):
marks = ",".join("?" for _ in chunk)
for row in conn.execute(
"SELECT pp.positive_prompt_id,pp.id,pp.negative_prompt_id," # nosec
"pp.archived,pp.title,pp.category,pp.rating,pp.notes,pp.run_count,"
"pp.last_used_at,pp.updated_at,"
"(SELECT group_concat(tag_id) FROM (SELECT tag_id FROM "
"prompt_pair_tags WHERE pair_id=pp.id ORDER BY tag_id)),"
"(SELECT COUNT(*) FROM image_prompt_pairs WHERE pair_id=pp.id),"
"(SELECT MAX(id) FROM image_prompt_pairs WHERE pair_id=pp.id) "
f"FROM prompt_pairs pp WHERE pp.positive_prompt_id IN ({marks}) "
"ORDER BY pp.id",
chunk,
):
prints[row[0]].append(
[
v if isinstance(v, (int, str, type(None))) else repr(v)
for v in tuple(row)[1:]
]
)
return prints
@classmethod
def _review_digests(cls, conn: sqlite3.Connection, prompt_ids: list[int]) -> dict:
"""#482: one digest per deletable prompt of its row and its pairs."""
prompts = cls._fingerprints(conn, prompt_ids)
pairs = cls._pair_prints(conn, prompt_ids)
return {
pid: hashlib.sha256(
json.dumps(
[prompts.get(pid), pairs[pid]], separators=(",", ":")
).encode("utf-8")
).hexdigest()
for pid in prompt_ids
}
@staticmethod
def _paired_counts(
conn: sqlite3.Connection, prompt_ids: list[int], selected: list[int]
) -> dict:
"""#482: what Delete permanently would remove with ``prompt_ids``."""
pairs = archived = variants = 0
negatives = set()
for chunk in _batches(prompt_ids):
marks = ",".join("?" for _ in chunk)
for archived_flag, negative, positive in conn.execute(
"SELECT archived,negative_prompt_id,positive_prompt_id " # nosec
f"FROM prompt_pairs WHERE positive_prompt_id IN ({marks})",
chunk,
):
pairs += 1
archived += 1 if archived_flag else 0
if negative is not None:
variants += 1
if negative != positive:
negatives.add(negative)
return {
"pairs": pairs,
"archived_pairs": archived,
"variants": variants,
"kept_negatives": len(negatives - set(selected)),
}
@staticmethod
def _checked_ids(prompt_ids: list[int], expected_epoch: str) -> list[int]:
_validate_epoch(expected_epoch)
@@ -156,11 +249,16 @@ class LibraryArchiveOperations:
with self.library_snapshot() as (conn, _epoch, _ceiling):
epoch = self._library_epoch(conn, expected_epoch)
classes = self._classify_library_delete(conn, prompt_ids)
deletable = classes["deletable"]
return {
"library_epoch": epoch,
**classes,
"image_links": self._image_links(conn, classes["delete"]),
"fingerprints": self._fingerprints(conn, classes["delete"]),
# #482: Delete permanently, with the deletable prompts' pairs.
"review_digests": self._review_digests(conn, deletable),
"paired_counts": self._paired_counts(conn, deletable, prompt_ids),
"paired_image_links": self._distinct_images(conn, deletable),
}
@staticmethod
@@ -191,6 +289,7 @@ class LibraryArchiveOperations:
*,
expected_epoch: str,
candidates: dict | None,
delete_paired: dict | None = None,
) -> dict:
"""Archive and delete what the user reviewed, all or nothing.
@@ -201,6 +300,11 @@ class LibraryArchiveOperations:
still classifies ``delete`` and its fingerprint is unchanged; one that
is now paired, and a reviewed archive now unpaired, are archived (the
restorable direction); anything else is kept and counted.
``delete_paired`` (#482) maps reviewed archive ids the user chose to
Delete permanently to their review digest. Each is deleted with its
own saved pairs only while it is still deletable and its digest is
unchanged; otherwise it is archived and counted ``changed``.
"""
_validate_epoch(expected_epoch)
for ids in (archive_ids, delete_ids):
@@ -209,6 +313,16 @@ class LibraryArchiveOperations:
):
raise ValueError("prompt ids must be lists of positive integers")
reviewed_delete = set(delete_ids)
delete_paired = {} if delete_paired is None else delete_paired
if not isinstance(delete_paired, dict) or any(
type(pid) is not int
or not isinstance(digest, str)
or not _DIGEST.match(digest)
for pid, digest in delete_paired.items()
):
raise ValueError("delete_paired must map prompt ids to review digests")
if not set(delete_paired) <= set(archive_ids):
raise ValueError("A permanent delete must be a reviewed archive id")
if not archive_ids and not delete_ids:
raise ValueError("Nothing to delete or archive")
if reviewed_delete & set(archive_ids):
@@ -218,16 +332,31 @@ class LibraryArchiveOperations:
epoch = self._library_epoch(conn, expected_epoch)
classes = self._classify_library_delete(conn, prompt_ids)
now_prints = self._fingerprints(conn, classes["delete"])
deletes, archives, changed = [], list(classes["archive"]), 0
deletes, archives, changed = [], [], 0
paired_now = set(classes["deletable"])
chosen = [pid for pid in classes["archive"] if pid in delete_paired]
digests = self._review_digests(
conn, [pid for pid in chosen if pid in paired_now]
)
paired_deletes = []
for pid in classes["archive"]:
if pid not in delete_paired:
archives.append(pid)
elif digests.get(pid) == delete_paired[pid]:
paired_deletes.append(pid)
else:
archives.append(pid) # changed since the review: restorable
changed += 1
for pid in classes["delete"]:
if pid not in reviewed_delete:
archives.append(pid) # reviewed as archive, now unpaired
changed += 1 if pid in delete_paired else 0
elif candidates is None or candidates.get(pid) == now_prints[pid]:
deletes.append(pid)
else:
changed += 1
changed += len(classes["protected"])
image_links = self._image_links(conn, deletes)
image_links = self._image_links(conn, [*deletes, *paired_deletes])
now = utc_now_iso()
for pid in archives:
self._repair_hash(conn, pid)
@@ -240,7 +369,25 @@ class LibraryArchiveOperations:
"UPDATE prompts SET updated_at=? WHERE id=?",
[(now, pid) for pid in archives],
)
deleted = delete_prompt_rows(conn, deletes)
# #482: the image-pair links (RESTRICT) go first, then the pairs
# (pair tags CASCADE, review decisions SET NULL), then the prompts.
pairs_removed = 0
for chunk in _batches(paired_deletes):
marks = ",".join("?" for _ in chunk)
pairs_of = (
"SELECT id FROM prompt_pairs " # nosec B608
f"WHERE positive_prompt_id IN ({marks})"
)
conn.execute(
f"DELETE FROM image_prompt_pairs WHERE pair_id IN ({pairs_of})",
chunk,
)
pairs_removed += conn.execute(
"DELETE FROM prompt_pairs " # nosec B608
f"WHERE positive_prompt_id IN ({marks})",
chunk,
).rowcount
deleted = delete_prompt_rows(conn, [*deletes, *paired_deletes])
return {
"library_epoch": epoch,
"deleted": deleted,
@@ -249,6 +396,7 @@ class LibraryArchiveOperations:
"not_found": len(classes["not_found"]),
"changed": changed,
"image_links": image_links,
"pairs_removed": pairs_removed,
}
def restore_library_prompts(
py/api/library_delete.py
+44 -10
View File
@@ -47,6 +47,8 @@ RESTORE_URL = "/prompt_manager/library/restore"
DELETE_TITLE = "Delete prompts (Library)"
_SELECTION_KEYS = frozenset({"prompt_ids", "expected_epoch"})
_DELETE_KEYS = frozenset({"archive_ids", "delete_ids", "expected_epoch", "backup"})
# #482: Delete permanently, with its digests; both or neither (a #443 body).
_PAIRED_KEYS = frozenset({"delete_paired_ids", "paired_digests"})
STORAGE_CHANGED = "Prompt storage was restored; refresh the library"
@@ -64,19 +66,39 @@ def _storage_changed(exc):
return _failure(409, exc.code, STORAGE_CHANGED, library_epoch=exc.epoch)
async def _body(request, keys):
"""The JSON object with exactly ``keys``, or raise ValueError."""
async def _body(request, keys, optional=frozenset()):
"""The JSON object with exactly ``keys`` (plus all or none of ``optional``)."""
try:
data = await request.json()
except ValueError:
raise ValueError("Expected a JSON object") from None
if not isinstance(data, dict):
raise ValueError("Expected a JSON object")
if set(data) != keys:
extra = set(data) - keys
if not keys <= set(data) or (extra and extra != optional):
raise ValueError("Request must contain exactly " + ", ".join(sorted(keys)))
return data
def _paired_delete(data, archive_ids):
"""#482: the reviewed archive ids to delete with their pairs -> digest."""
if "delete_paired_ids" not in data:
return {}
ids = _reviewed_ids(data["delete_paired_ids"], "delete_paired_ids")
digests = data["paired_digests"]
if not isinstance(digests, dict):
raise ValueError("paired_digests must map prompt ids to digests")
if not set(ids) <= set(archive_ids):
raise ValueError("delete_paired_ids must be reviewed archive ids")
chosen = {}
for pid in ids:
digest = digests.get(str(pid))
if not isinstance(digest, str) or len(digest) != 64:
raise ValueError("Every delete_paired_ids entry needs its review digest")
chosen[pid] = digest
return chosen
def _reviewed_ids(value, name):
"""A reviewed id list: possibly empty, else as ``parse_bulk_prompt_ids``."""
if value == []:
@@ -111,7 +133,7 @@ class LibraryDeleteRoutesMixin:
expected_epoch=data["expected_epoch"],
)
disk = None
if review["delete"] and policy == "on":
if (review["delete"] or review["deletable"]) and policy == "on":
public, short, _size = await self._run_in_executor(
_maintenance.disk_plan,
db.model.db_path,
@@ -145,6 +167,13 @@ class LibraryDeleteRoutesMixin:
"already_archived": len(review["already_archived"]),
"not_found": len(review["not_found"]),
"image_links": review["image_links"],
# #482: the archive ids Delete permanently may delete with pairs.
"deletable": review["deletable"],
"review_digests": {
str(pid): digest for pid, digest in review["review_digests"].items()
},
"paired_counts": review["paired_counts"],
"paired_image_links": review["paired_image_links"],
"backup": policy,
"disk": disk,
}
@@ -152,7 +181,7 @@ class LibraryDeleteRoutesMixin:
async def delete_library_prompts(self, request):
try:
data = await _body(request, _DELETE_KEYS)
data = await _body(request, _DELETE_KEYS, _PAIRED_KEYS)
archive_ids = _reviewed_ids(data["archive_ids"], "archive_ids")
delete_ids = _reviewed_ids(data["delete_ids"], "delete_ids")
if not archive_ids and not delete_ids:
@@ -163,6 +192,7 @@ class LibraryDeleteRoutesMixin:
raise ValueError("A prompt cannot be both archived and deleted")
if data["backup"] not in ("on", "off"):
raise ValueError("backup must be on or off")
paired = _paired_delete(data, archive_ids)
except ValueError as exc:
return _invalid(str(exc))
busy = self._jobs.blocker(PROMPT_DELETE_JOB_KIND, admission="records")
@@ -173,7 +203,7 @@ class LibraryDeleteRoutesMixin:
PROMPT_DELETE_JOB_KIND, DELETE_TITLE, admission="records"
) as job:
response = await self._held_library_delete(
self.db, data, archive_ids, delete_ids
self.db, data, archive_ids, delete_ids, paired
)
if response.status >= 400:
job.finish("failed")
@@ -181,18 +211,21 @@ class LibraryDeleteRoutesMixin:
except JobBusy as busy:
return busy_response(busy)
async def _held_library_delete(self, db, data, archive_ids, delete_ids):
async def _held_library_delete(self, db, data, archive_ids, delete_ids, paired):
epoch = data["expected_epoch"]
promised = data["backup"] == "on" or self._safety_backup_policy() == "on"
backup = None
try:
candidates = None
if delete_ids:
if delete_ids or paired:
review = await self._run_in_executor(
db.classify_library_delete, delete_ids, expected_epoch=epoch
db.classify_library_delete,
[*delete_ids, *paired],
expected_epoch=epoch,
)
candidates = review["fingerprints"]
if candidates and promised:
# #482: a permanent delete of a prompt in a saved pair is backed up too.
if (candidates or review["deletable"]) and promised:
_disk, short, _size = await self._run_in_executor(
_maintenance.disk_plan,
db.model.db_path,
@@ -214,6 +247,7 @@ class LibraryDeleteRoutesMixin:
delete_ids,
expected_epoch=epoch,
candidates=candidates if candidates is not None else {},
delete_paired=paired,
)
except LibraryCommandError as exc:
return _storage_changed(exc)
tests/test_bulk_library_delete.py
+202
View File
@@ -62,6 +62,8 @@ def table_dump(db):
class LibraryDeleteCase(unittest.TestCase):
maxDiff = None
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.root = Path(self.temp.name)
@@ -137,9 +139,23 @@ class TestClassifyAndApply(LibraryDeleteCase):
"not_found": [negative_only, 99999],
"image_links": 1,
"fingerprints": result["fingerprints"],
# #482: the archive ids Delete permanently could remove (with pairs).
"deletable": [positive, archived_pair_positive],
"review_digests": result["review_digests"],
"paired_counts": {
"pairs": 2,
"archived_pairs": 1,
"variants": 2,
# "only negative" is itself selected; only "blur" is kept.
"kept_negatives": 1,
},
"paired_image_links": 0,
},
)
self.assertEqual(list(result["fingerprints"]), [plain])
self.assertEqual(
list(result["review_digests"]), [positive, archived_pair_positive]
)
def test_apply_archives_paired_deletes_unpaired_and_skips_protected(self):
plain = save(self.db, "Plain", tags=["red"])
@@ -159,6 +175,7 @@ class TestClassifyAndApply(LibraryDeleteCase):
"not_found": 0,
"changed": 0,
"image_links": 1,
"pairs_removed": 0,
},
)
self.assertEqual(prompt_ids(self.db), sorted([protected, positive, negative]))
@@ -523,3 +540,188 @@ class TestArchivedReaders(LibraryDeleteCase):
if __name__ == "__main__":
unittest.main()
class TestPairedDelete(LibraryDeleteCase):
"""#482 (D154): Delete permanently also removes a prompt's own saved pairs."""
def pair_rows(self, positive):
return rows(
self.db,
"SELECT id FROM prompt_pairs WHERE positive_prompt_id=? ORDER BY id",
(positive,),
)
def pair_link(self, pair, image):
return self.db.link_image_to_pair(pair, image, {"resolver_version": 1})
def delete_paired(self, ids, review, archive=None, epoch=None):
"""Write ``ids`` as Delete permanently, with the digests ``review`` gave."""
archive = review["archive"] if archive is None else archive
return self.db.archive_or_delete_library_prompts(
archive,
review["delete"],
expected_epoch=epoch or self.epoch,
candidates=review["fingerprints"],
delete_paired={pid: review["review_digests"][pid] for pid in ids},
)
def test_classify_marks_deletable_and_archive_only_with_pair_counts(self):
cat, cat_negative, _pair = self.paired("Cat", "blur")
self.db.save_prompt_pair("Cat", "grain")
archived_pair = self.db.save_prompt_pair("Cat", "noise")
self.db.archive_prompt_pair(archived_pair, True)
shared = save(self.db, "Shared")
self.image(shared, "shared.png")
self.db.save_prompt_pair("Dog", "Shared") # used as another positive's negative
review = self.classify([cat, shared])
self.assertEqual(review["archive"], [cat, shared])
self.assertEqual(review["deletable"], [cat])
self.assertEqual(list(review["review_digests"]), [cat])
self.assertRegex(review["review_digests"][cat], r"^[0-9a-f]{64}$")
self.assertEqual(
review["paired_counts"],
{"pairs": 3, "archived_pairs": 1, "variants": 3, "kept_negatives": 3},
)
self.assertEqual(review["paired_image_links"], 0)
def test_an_empty_negative_pair_is_deleted_with_its_prompt(self):
# Live prompt 7896: an auto-created pair with an explicitly empty negative.
pair = self.db.save_prompt_pair("été", "")
prompt = self.db.get_prompt_pair(pair)["positive_prompt_id"]
review = self.classify([prompt])
self.assertEqual(review["deletable"], [prompt])
self.assertEqual(
review["paired_counts"],
{"pairs": 1, "archived_pairs": 0, "variants": 0, "kept_negatives": 0},
)
result = self.delete_paired([prompt], review)
self.assertEqual((result["deleted"], result["archived"]), (1, 0))
self.assertEqual(result["pairs_removed"], 1)
self.assertNotIn(prompt, prompt_ids(self.db))
self.assertEqual(self.pair_rows(prompt), [])
self.assertEqual(rows(self.db, "PRAGMA foreign_key_check"), [])
def test_rows_removed_are_the_legacy_pair_deletes_then_the_prompt_delete(self):
twin = PromptDatabase(str(self.root / "twin.db"))
self.addCleanup(twin.close_all)
twin_epoch = twin.get_library_options()["library_epoch"]
made = {}
for db in (self.db, twin):
plain = db.save_prompt_pair("Owl", "", tags=["night"], title="Owl")
with_negative = db.save_prompt_pair("Owl", "dark", rating=4)
hidden = db.save_prompt_pair("Owl", "fog")
db.archive_prompt_pair(hidden, True)
owl = db.get_prompt_pair(plain)["positive_prompt_id"]
path = self.root / "owl.png"
path.write_bytes(b"png")
image = db.link_image_to_prompt(owl, str(path), {"workflow": {"nodes": []}})
db.link_image_to_pair(with_negative, image, {"resolver_version": 1})
save(db, "Bystander", tags=["night"])
made[id(db)] = (owl, [plain, with_negative, hidden])
owl, pairs = made[id(self.db)]
review = self.classify([owl])
self.delete_paired([owl], review)
twin_owl, twin_pairs = made[id(twin)]
with twin.model.get_connection() as conn:
conn.execute(
"DELETE FROM image_prompt_pairs WHERE pair_id IN (SELECT id FROM "
"prompt_pairs WHERE positive_prompt_id=?)",
(twin_owl,),
)
conn.commit()
for pair in twin_pairs:
self.assertTrue(twin.delete_prompt_pair(pair))
twin.archive_or_delete_library_prompts(
[], [twin_owl], expected_epoch=twin_epoch, candidates=None
)
self.assertEqual(table_dump(self.db), table_dump(twin))
# The negatives stay, and the image file stays on disk.
self.assertTrue((self.root / "owl.png").exists())
texts = [row[0] for row in rows(self.db, "SELECT text FROM prompts")]
self.assertTrue({"dark", "fog", "Bystander"} <= set(texts))
def test_a_negative_used_by_another_positive_is_archived_never_deleted(self):
shared = save(self.db, "Shared")
self.image(shared, "s.png")
self.db.save_prompt_pair("Dog", "Shared")
review = self.classify([shared])
self.assertEqual(review["deletable"], [])
result = self.db.archive_or_delete_library_prompts(
[shared],
[],
expected_epoch=self.epoch,
candidates={},
delete_paired={shared: "0" * 64},
)
self.assertEqual((result["deleted"], result["archived"]), (0, 1))
self.assertEqual(result["changed"], 1)
self.assertEqual(archived_ids(self.db), [shared])
def test_a_positive_and_its_own_negative_selected_together(self):
cat, blur, _pair = self.paired("Cat", "blur")
self.image(blur, "blur.png") # the negative is a Library prompt too
review = self.classify([cat, blur])
self.assertEqual(review["deletable"], [cat])
result = self.delete_paired([cat], review)
self.assertEqual((result["deleted"], result["archived"]), (1, 1))
self.assertEqual(archived_ids(self.db), [blur])
self.assertNotIn(cat, prompt_ids(self.db))
def test_a_self_pair_is_deleted(self):
pair = self.db.save_prompt_pair("Same", "Same")
same = self.db.get_prompt_pair(pair)["positive_prompt_id"]
review = self.classify([same])
self.assertEqual(review["deletable"], [same])
self.assertEqual(review["paired_counts"]["kept_negatives"], 0)
self.assertEqual(self.delete_paired([same], review)["deleted"], 1)
self.assertNotIn(same, prompt_ids(self.db))
def test_drift_after_the_review_archives_and_counts_changed(self):
cases = {
"pair added": lambda p: self.db.save_prompt_pair("Fox", "new"),
"pair edited": lambda p: self.db.update_prompt_pair_metadata(
self.pair_rows(p)[0][0], title="Renamed"
),
"pairs gone": lambda p: self.db.delete_prompt_pair(self.pair_rows(p)[0][0]),
"now another's negative": lambda p: self.db.save_prompt_pair("Hen", "Fox"),
}
for name, change in cases.items():
with self.subTest(name):
fox, _neg, _pair = self.paired("Fox", "mist " + name)
review = self.classify([fox])
self.assertEqual(review["deletable"], [fox])
change(fox)
result = self.delete_paired([fox], review)
self.assertEqual(
(result["deleted"], result["archived"], result["changed"]),
(0, 1, 1),
)
self.assertIn(fox, archived_ids(self.db))
self.restore([fox])
with self.db.model.get_connection() as conn:
conn.execute(
"DELETE FROM prompt_pairs WHERE positive_prompt_id IN "
"(SELECT id FROM prompts WHERE text IN ('Hen','Fox'))"
)
conn.commit()
def test_delete_paired_must_name_reviewed_archive_ids(self):
fox, _neg, _pair = self.paired("Fox")
review = self.classify([fox])
with self.assertRaises(ValueError):
self.db.archive_or_delete_library_prompts(
[],
[],
expected_epoch=self.epoch,
candidates={},
delete_paired={fox: review["review_digests"][fox]},
)
with self.assertRaises(ValueError):
self.db.archive_or_delete_library_prompts(
[fox],
[],
expected_epoch=self.epoch,
candidates={},
delete_paired={fox: "not a digest"},
)
tests/test_library_delete_routes.py
+94 -1
View File
@@ -130,8 +130,13 @@ class TestLibraryDeleteRoutes(LibraryDeleteRouteCase):
async def test_archive_only_takes_no_backup(self):
self.policy(True)
positive = self.paired("Cat")
shared = self.save("Shared")
self.api.db.save_prompt_pair("Dog", "Shared") # archive-only (#482)
self.api.db.link_image_to_prompt(shared, "/tmp/shared-443.png")
# Nothing could be deleted: no disk plan. (#482: a deletable prompt has one.)
self.assertIsNone((await self.review([shared]))["disk"])
review = await self.review([positive])
self.assertIsNone(review["disk"])
self.assertIsNotNone(review["disk"])
status, body = await self.confirm(review)
self.assertEqual((status, body["archived"], body["backup"]), (200, 1, None))
self.assertEqual(self.backups(), [])
@@ -309,6 +314,94 @@ class TestLibraryDeleteRoutes(LibraryDeleteRouteCase):
self.assertEqual(response.status, 400)
class TestPairedDeleteRoutes(LibraryDeleteRouteCase):
"""#482 (D154): Delete permanently with a prompt's own saved pairs."""
async def confirm_paired(self, review, ids, backup=None, digests=None):
return await self.post(
DELETE,
{
"archive_ids": review["archive"],
"delete_ids": review["delete"],
"expected_epoch": review["library_epoch"],
"backup": backup or review["backup"],
"delete_paired_ids": ids,
"paired_digests": digests
or {str(pid): review["review_digests"][str(pid)] for pid in ids},
},
)
async def test_preview_reports_deletable_and_a_paired_only_delete_backs_up(self):
self.policy(True)
cat = self.paired("Cat")
shared = self.save("Shared")
self.api.db.save_prompt_pair("Dog", "Shared")
self.api.db.link_image_to_prompt(shared, "/tmp/shared-482.png")
review = await self.review([cat, shared])
self.assertEqual(review["archive"], [cat, shared])
self.assertEqual(review["deletable"], [cat])
self.assertEqual(list(review["review_digests"]), [str(cat)])
self.assertEqual(
review["paired_counts"],
{"pairs": 1, "archived_pairs": 0, "variants": 1, "kept_negatives": 1},
)
self.assertEqual(review["paired_image_links"], 0)
# A selection with nothing in the delete class still checks the disk.
self.assertTrue(review["disk"]["sufficient"])
status, body = await self.confirm_paired(review, [cat])
self.assertEqual(status, 200, body)
self.assertEqual(
(body["deleted"], body["archived"], body["pairs_removed"]), (1, 1, 1)
)
self.assertEqual(self.backups(), [body["backup"]])
self.assertNotIn(cat, self.ids())
self.assertEqual(self.archived(), [shared])
async def test_a_body_without_the_new_keys_is_the_443_delete(self):
cat = self.paired("Cat")
review = await self.review([cat])
status, body = await self.confirm(review)
self.assertEqual(status, 200, body)
self.assertEqual((body["deleted"], body["archived"]), (0, 1))
self.assertIn(cat, self.ids())
async def test_paired_bodies_are_validated(self):
cat = self.paired("Cat")
plain = self.save("Plain")
review = await self.review([cat, plain])
digest = review["review_digests"][str(cat)]
base = {
"archive_ids": review["archive"],
"delete_ids": review["delete"],
"expected_epoch": review["library_epoch"],
"backup": "off",
}
cases = [
{"delete_paired_ids": [cat]}, # without its digests
{"delete_paired_ids": [plain], "paired_digests": {str(plain): digest}},
{"delete_paired_ids": [cat], "paired_digests": {}},
{"delete_paired_ids": [cat], "paired_digests": {str(cat): "x"}},
{"delete_paired_ids": "all", "paired_digests": {}},
]
for extra in cases:
with self.subTest(extra=extra):
status, body = await self.post(DELETE, {**base, **extra})
self.assertEqual((status, body["code"]), (400, "invalid-request"))
self.assertIn(cat, self.ids())
async def test_a_short_disk_refuses_a_paired_only_delete(self):
self.policy(True)
cat = self.paired("Cat")
from py.api import maintenance
with patch.object(maintenance, "_free_bytes", return_value=0):
review = await self.review([cat])
self.assertFalse(review["disk"]["sufficient"])
status, body = await self.confirm_paired(review, [cat])
self.assertEqual((status, body["code"]), (507, "insufficient-space"))
self.assertIn(cat, self.ids())
class TestIngestionAfterArchive(LibraryDeleteRouteCase):
def scan(self, name, positive, negative="blur"):
meta = {"prompt": json.dumps(graph(positive, negative))}