245 lines
10 KiB
Python
245 lines
10 KiB
Python
"""Mutable metadata galleries never own or retarget frozen output references."""
|
|
|
|
from PIL import Image
|
|
from PIL.PngImagePlugin import PngInfo
|
|
|
|
from database.pairs import PairDependencyError
|
|
from tests import test_execution_history_database as history_tests
|
|
from tests.test_api_admin import AdminAPITestCase
|
|
from tests.test_execution_output_database import bundle
|
|
from utils.execution_outputs import snapshot_output_roots
|
|
|
|
|
|
class OutputGalleryTests(AdminAPITestCase):
|
|
start = history_tests.HistoryDatabaseTests.start
|
|
finish = history_tests.HistoryDatabaseTests.finish
|
|
|
|
def capture(self, filename="first.png", item=0, root=None):
|
|
db = self.api.db
|
|
start = self.start(db, item)
|
|
db.write_execution_start(start)
|
|
key = snapshot_output_roots(
|
|
"posix", str(root or self.output_dir), None
|
|
).output_key
|
|
db.write_execution_outputs_for_finish(
|
|
self.finish(start), bundle(filename, root=key)
|
|
)
|
|
return start.run_id, db.get_execution_outputs(start.run_id)
|
|
|
|
def assert_evidence(self, run, expected):
|
|
self.assertEqual(self.api.db.get_execution_outputs(run), expected)
|
|
reference = expected["references"][0]
|
|
self.assertEqual(reference["context_membership"], "unknown")
|
|
self.assertNotIn("image_id", reference)
|
|
self.assertNotIn("pair_id", reference)
|
|
self.assertNotIn("prompt_id", reference)
|
|
|
|
def gallery(self):
|
|
conn = self.api.db.model.get_connection()
|
|
return {
|
|
table: [
|
|
tuple(row) for row in conn.execute(f"SELECT * FROM {table} ORDER BY id")
|
|
]
|
|
for table in (
|
|
"prompts",
|
|
"prompt_pairs",
|
|
"generated_images",
|
|
"image_prompt_pairs",
|
|
)
|
|
}
|
|
|
|
def test_metadata_disabled_or_malformed_still_captures_without_gallery_rows(self):
|
|
for item, metadata in enumerate(({}, {"prompt": "{malformed"})):
|
|
with self.subTest(metadata=metadata):
|
|
path = self.output_dir / f"metadata-{item}.png"
|
|
info = PngInfo()
|
|
for key, value in metadata.items():
|
|
info.add_text(key, value)
|
|
Image.new("RGB", (2, 2)).save(path, pnginfo=info)
|
|
extracted = self.api._extract_comfyui_metadata(str(path))
|
|
self.assertEqual(
|
|
self.api._ingest_scanned_file(path, extracted),
|
|
None if not metadata else "unresolved",
|
|
)
|
|
before = self.gallery()
|
|
run, evidence = self.capture(path.name, item)
|
|
self.assertEqual(evidence["state"], "complete")
|
|
self.assertEqual(self.gallery(), before)
|
|
self.assertEqual(self.api.db.get_recent_prompts()["total"], 0)
|
|
self.assert_evidence(run, evidence)
|
|
|
|
def test_scans_before_after_delayed_repeated_and_overwrite_keep_exact_counters(
|
|
self,
|
|
):
|
|
metadata = {
|
|
"parameters": "cat\nNegative prompt: blur\nSteps: 20, Sampler: Euler"
|
|
}
|
|
db = self.api.db
|
|
captures = []
|
|
for item, before_capture in enumerate((True, False)):
|
|
path = self.output_dir / f"scan-{item}.png"
|
|
path.write_bytes(b"first contents")
|
|
if before_capture:
|
|
first = self.api._ingest_scan_batch_sync([(path, metadata)])
|
|
before = self.gallery()
|
|
run, evidence = self.capture(path.name, item)
|
|
captures.append((run, evidence))
|
|
self.assertEqual(self.gallery(), before)
|
|
if not before_capture:
|
|
first = self.api._ingest_scan_batch_sync([(path, metadata)])
|
|
self.assertEqual(
|
|
first,
|
|
{
|
|
"processed": 1,
|
|
"found": 1,
|
|
"added": 1,
|
|
"linked": 0,
|
|
"enriched": 0,
|
|
"unchanged": 0,
|
|
"ambiguous": 0,
|
|
"unresolved": 0,
|
|
"conflict": 0,
|
|
"failed": 0,
|
|
"pairs_added": int(before_capture),
|
|
"pair_links_added": 1,
|
|
},
|
|
)
|
|
for _ in range(2):
|
|
repeated = self.api._ingest_scan_batch_sync([(path, metadata)])
|
|
self.assertEqual(
|
|
repeated,
|
|
{
|
|
"processed": 1,
|
|
"found": 1,
|
|
"added": 0,
|
|
"linked": 0,
|
|
"enriched": 0,
|
|
"unchanged": 1,
|
|
"ambiguous": 0,
|
|
"unresolved": 0,
|
|
"conflict": 0,
|
|
"failed": 0,
|
|
"pairs_added": 0,
|
|
"pair_links_added": 0,
|
|
},
|
|
)
|
|
path.write_bytes(b"replacement bytes at the same relative path")
|
|
conflict = self.api._ingest_scan_batch_sync(
|
|
[
|
|
(
|
|
path,
|
|
{"parameters": metadata["parameters"].replace("blur", "noise")},
|
|
)
|
|
]
|
|
)
|
|
self.assertEqual(conflict["conflict"], 1)
|
|
self.assertEqual(conflict["added"], 0)
|
|
self.assertEqual(db.list_prompt_pairs()["total"], 1)
|
|
self.assert_evidence(run, evidence)
|
|
pair = db.list_prompt_pairs()["pairs"][0]
|
|
self.assertEqual((pair["run_count"], pair["image_count"]), (0, 2))
|
|
self.assertEqual(
|
|
[p["run_count"] for p in db.get_recent_prompts()["prompts"]], [0, 0]
|
|
)
|
|
for run, evidence in captures:
|
|
self.assert_evidence(run, evidence)
|
|
|
|
def test_same_relative_path_across_roots_never_retargets(self):
|
|
second_root = self.output_dir / "second-root"
|
|
second_root.mkdir()
|
|
metadata = {
|
|
"parameters": "cat\nNegative prompt: blur\nSteps: 20, Sampler: Euler"
|
|
}
|
|
records = []
|
|
for item, root in enumerate((self.output_dir, second_root)):
|
|
path = root / "same.png"
|
|
path.write_bytes(b"image")
|
|
result = self.api._ingest_scanned_file(path, metadata, detailed=True)
|
|
self.assertEqual(result["status"], "added")
|
|
records.append(self.capture("same.png", item, root))
|
|
self.assertNotEqual(
|
|
records[0][1]["references"][0]["root_key"],
|
|
records[1][1]["references"][0]["root_key"],
|
|
)
|
|
self.assertEqual(self.api.db.list_prompt_pairs()["pairs"][0]["image_count"], 2)
|
|
for root in (self.output_dir, second_root):
|
|
(root / "same.png").unlink()
|
|
second_root.rmdir()
|
|
self.assertEqual(self.api.db.cleanup_missing_images(), 2)
|
|
for run, expected in records:
|
|
self.assert_evidence(run, expected)
|
|
|
|
def test_archive_delete_prune_cleanup_and_numeric_id_reuse_keep_evidence(self):
|
|
db = self.api.db
|
|
path = self.output_dir / "first.png"
|
|
path.write_bytes(b"image")
|
|
result = self.api._ingest_scanned_file(
|
|
path,
|
|
{"parameters": "cat\nNegative prompt: blur\nSteps: 20, Sampler: Euler"},
|
|
detailed=True,
|
|
)
|
|
self.assertEqual(
|
|
(result["status"], result["pairs_added"], result["links_added"]),
|
|
("added", 1, 1),
|
|
)
|
|
image_id, pair_id = result["image_id"], result["pair_id"]
|
|
pair = db.get_prompt_pair(pair_id)
|
|
positive_id = pair["positive_prompt_id"]
|
|
run, evidence = self.capture()
|
|
with self.assertRaises(PairDependencyError):
|
|
db.delete_prompt(positive_id)
|
|
with self.assertRaises(PairDependencyError):
|
|
db.delete_prompt_pair(pair_id)
|
|
self.assertTrue(db.archive_prompt_pair(pair_id))
|
|
self.assertEqual(db.list_prompt_pairs()["total"], 0)
|
|
self.assertEqual(db.list_prompt_pairs(include_archived=True)["total"], 1)
|
|
self.assertEqual(db.get_prompt_pair(pair_id)["run_count"], 0)
|
|
self.assert_evidence(run, evidence)
|
|
self.assertTrue(db.delete_image(image_id))
|
|
self.assertFalse(db.delete_image(image_id))
|
|
self.assertEqual(db.get_image_prompt_pairs(image_id), [])
|
|
self.assertTrue(db.delete_prompt_pair(pair_id))
|
|
self.assertTrue(db.delete_prompt(positive_id))
|
|
self.assertEqual(db.prune_orphaned_prompts(), 1)
|
|
self.assertEqual(db.prune_orphaned_prompts(), 0)
|
|
self.assertEqual(db.cleanup_missing_images(), 0)
|
|
self.assert_evidence(run, evidence)
|
|
|
|
# Explicit IDs reproduce restored/imported numeric-ID reuse without assuming
|
|
# SQLite's allocator reuses AUTOINCREMENT identities.
|
|
conn = db.model.get_connection()
|
|
with conn:
|
|
conn.execute(
|
|
"INSERT INTO prompts(id,text,run_count) "
|
|
"VALUES(?, 'replacement prompt', 17)",
|
|
(positive_id,),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO prompt_pairs(id,positive_prompt_id,run_count) "
|
|
"VALUES(?,?,23)",
|
|
(pair_id, positive_id),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO generated_images "
|
|
"(id,prompt_id,image_path,filename,file_path) "
|
|
"VALUES(?,?,?,'first.png',?)",
|
|
(image_id, positive_id, str(path), str(path)),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO image_prompt_pairs(image_id,pair_id,provenance) "
|
|
"VALUES(?,?,'{}')",
|
|
(image_id, pair_id),
|
|
)
|
|
self.assert_evidence(run, evidence)
|
|
self.assertTrue(db.archive_prompt_pair(pair_id))
|
|
path.unlink()
|
|
self.assertEqual(db.cleanup_missing_images(), 1)
|
|
self.assertEqual(db.get_image_prompt_pairs(image_id), [])
|
|
self.assertTrue(db.delete_prompt_pair(pair_id))
|
|
# Standalone prompt deletion retains its original gallery cascade.
|
|
recreated = db.link_image_to_prompt(positive_id, str(path))
|
|
self.assertGreater(recreated, 0)
|
|
self.assertTrue(db.delete_prompt(positive_id))
|
|
self.assertIsNone(db.get_image_by_id(recreated))
|
|
self.assert_evidence(run, evidence)
|