diff --git a/database/operations.py b/database/operations.py index 9424f1e..f43e2ab 100644 --- a/database/operations.py +++ b/database/operations.py @@ -924,6 +924,8 @@ class PromptDatabase(PairOperations): old_name = old_name.strip() new_name = new_name.strip() + if old_name == new_name: + return {"success": True, "affected_count": 0, "skipped_count": 0} with self.model.get_connection() as conn: cursor = conn.execute("SELECT id FROM tags WHERE name = ?", (old_name,)) @@ -945,12 +947,14 @@ class PromptDatabase(PairOperations): if existing_new: # Target tag exists — merge: move links, handle conflicts, delete old new_tag_id = existing_new["id"] - conn.execute( - "UPDATE OR IGNORE prompt_tags SET tag_id = ? WHERE tag_id = ?", - (new_tag_id, old_tag_id), - ) - # Delete remaining links (conflicts = prompt already had the target tag) - conn.execute("DELETE FROM prompt_tags WHERE tag_id = ?", (old_tag_id,)) + for junction in ("prompt_tags", "prompt_pair_tags"): + conn.execute( + f"UPDATE OR IGNORE {junction} SET tag_id = ? WHERE tag_id = ?", + (new_tag_id, old_tag_id), + ) + conn.execute( + f"DELETE FROM {junction} WHERE tag_id = ?", (old_tag_id,) + ) conn.execute("DELETE FROM tags WHERE id = ?", (old_tag_id,)) else: # Simple rename @@ -969,7 +973,7 @@ class PromptDatabase(PairOperations): """ Remove a tag from all prompts that use it. - With junction tables, deleting from the tags table cascades to prompt_tags. + Deleting from tags cascades to standalone and pair tag relationships. Args: tag_name: Tag name to remove @@ -1036,20 +1040,27 @@ class PromptDatabase(PairOperations): if not src_row: continue src_id = src_row["id"] + if src_id == target_id: + continue cursor = conn.execute( "SELECT COUNT(*) as c FROM prompt_tags WHERE tag_id = ?", (src_id,) ) src_count = cursor.fetchone()["c"] - if src_count > 0: - # Move links from source to target (ignore conflicts) - conn.execute( - "UPDATE OR IGNORE prompt_tags SET tag_id = ? WHERE tag_id = ?", - (target_id, src_id), - ) - # Delete remaining (conflicts = prompt already had target tag) - conn.execute("DELETE FROM prompt_tags WHERE tag_id = ?", (src_id,)) + pair_count = conn.execute( + "SELECT COUNT(*) FROM prompt_pair_tags WHERE tag_id = ?", (src_id,) + ).fetchone()[0] + if src_count or pair_count: + for junction in ("prompt_tags", "prompt_pair_tags"): + conn.execute( + f"UPDATE OR IGNORE {junction} " + "SET tag_id = ? WHERE tag_id = ?", + (target_id, src_id), + ) + conn.execute( + f"DELETE FROM {junction} WHERE tag_id = ?", (src_id,) + ) tags_merged += 1 affected += src_count diff --git a/tests/test_pair_dependency_concurrency.py b/tests/test_pair_dependency_concurrency.py new file mode 100644 index 0000000..e715614 --- /dev/null +++ b/tests/test_pair_dependency_concurrency.py @@ -0,0 +1,55 @@ +"""Dependency checks exclude competing pair writers until mutation finishes.""" + +import os +import sqlite3 +import tempfile +import unittest.mock + +from database.operations import PromptDatabase +from utils.hashing import generate_prompt_hash + + +class TestPairDependencyConcurrency(unittest.TestCase): + def test_legacy_mutators_lock_out_pair_creation_before_dependency_checks(self): + with tempfile.TemporaryDirectory() as directory: + path = os.path.join(directory, "concurrent.db") + db = PromptDatabase(path) + try: + original = db._ensure_prompts_unpaired + checks = [] + + def check_with_competing_writer(conn, ids): + original(conn, ids) + competitor = sqlite3.connect(path, timeout=0) + try: + with self.assertRaisesRegex(sqlite3.OperationalError, "locked"): + competitor.execute("BEGIN IMMEDIATE") + finally: + competitor.close() + checks.append(list(ids)) + + actions = ( + lambda pid: db.update_prompt_text(pid, "changed"), + db.delete_prompt, + lambda pid: db.bulk_delete_prompts([pid]), + lambda _pid: db.delete_prompts_by_category("temporary"), + ) + with unittest.mock.patch.object( + db, "_ensure_prompts_unpaired", check_with_competing_writer + ): + for index, action in enumerate(actions): + with self.subTest(action=index): + text = f"unpaired {index}" + pid = db.save_prompt( + text, + category="temporary", + prompt_hash=generate_prompt_hash(text), + ) + self.assertTrue(action(pid)) + self.assertEqual(len(checks), len(actions)) + finally: + db.close_all() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_pair_tag_maintenance.py b/tests/test_pair_tag_maintenance.py new file mode 100644 index 0000000..9930d7a --- /dev/null +++ b/tests/test_pair_tag_maintenance.py @@ -0,0 +1,101 @@ +"""Global tag maintenance preserves pair and standalone tag relationships.""" + +import os +import sqlite3 +import tempfile +import unittest + +from database.operations import PromptDatabase + + +class TestPairTagMaintenance(unittest.TestCase): + def setUp(self): + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.db = PromptDatabase(os.path.join(directory.name, "pair-tags.db")) + self.addCleanup(self.db.close_all) + + def pair(self, positive: str, tags: list) -> int: + return self.db.save_prompt_pair(positive, "blurry", tags=tags) + + def tags(self, pair_id: int) -> list: + return self.db.get_prompt_pair(pair_id)["tags"] + + def test_rename_to_existing_tag_preserves_pair_links_and_deduplicates(self): + first = self.pair("portrait", ["old", "new", "keep"]) + second = self.pair("landscape", ["old"]) + third = self.pair("studio", ["unrelated"]) + prompt = self.db.save_prompt("standalone", tags=["old", "new"]) + result = self.db.rename_tag_all_prompts("old", "new") + self.assertEqual(result["affected_count"], 1) + self.assertEqual(self.tags(first), ["keep", "new"]) + self.assertEqual(self.tags(second), ["new"]) + self.assertEqual(self.tags(third), ["unrelated"]) + self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["new"]) + self.assertEqual(self.db.list_prompt_pairs(tags=["new"])["total"], 2) + self.assertNotIn("old", self.db.get_all_tags()) + + def test_rename_pair_only_tag_to_new_name_preserves_links(self): + pair = self.pair("portrait", ["old"]) + self.db.rename_tag_all_prompts("old", "new") + self.assertEqual(self.tags(pair), ["new"]) + + def test_merge_pair_only_source_preserves_its_links(self): + pair = self.pair("portrait", ["source"]) + other = self.pair("landscape", ["target"]) + self.db.merge_tags(["source"], "target") + self.assertEqual(self.tags(pair), ["target"]) + self.assertEqual(self.tags(other), ["target"]) + self.assertEqual(self.db.list_prompt_pairs(tags=["target"])["total"], 2) + + def test_merge_multiple_sources_preserves_shared_and_unrelated_tags(self): + pair = self.pair("portrait", ["first", "second", "target", "keep"]) + other = self.pair("landscape", ["second"]) + prompt = self.db.save_prompt("standalone", tags=["first", "second"]) + self.db.merge_tags(["first", "second"], "target") + self.assertEqual(self.tags(pair), ["keep", "target"]) + self.assertEqual(self.tags(other), ["target"]) + self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["target"]) + self.assertEqual(self.db.list_prompt_pairs(tags=["target"])["total"], 2) + + def test_self_rename_does_not_remove_tag_or_relationships(self): + pair = self.pair("portrait", ["same"]) + prompt = self.db.save_prompt("standalone", tags=["same"]) + self.db.rename_tag_all_prompts("same", "same") + self.assertEqual(self.tags(pair), ["same"]) + self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["same"]) + + def test_merge_skips_target_even_when_it_is_in_source_list(self): + pair = self.pair("portrait", ["target", "source"]) + prompt = self.db.save_prompt("standalone", tags=["target", "source"]) + self.db.merge_tags(["target", "source", "target"], "target") + self.assertEqual(self.tags(pair), ["target"]) + self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["target"]) + self.assertIn("target", self.db.get_all_tags()) + + def test_explicit_global_delete_removes_pair_tags_and_keeps_other_tags(self): + pair = self.pair("portrait", ["remove", "keep"]) + prompt = self.db.save_prompt("standalone", tags=["remove", "keep"]) + self.db.delete_tag_all_prompts("remove") + self.assertEqual(self.tags(pair), ["keep"]) + self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["keep"]) + self.assertIsNotNone(self.db.get_prompt_pair(pair)) + + def test_pair_link_failure_rolls_back_standalone_merge_and_source_delete(self): + pair = self.pair("portrait", ["source"]) + prompt = self.db.save_prompt("standalone", tags=["source"]) + conn = self.db.model.get_connection() + conn.execute( + "CREATE TRIGGER reject_pair_tag_change BEFORE UPDATE " + "ON prompt_pair_tags BEGIN SELECT RAISE(ABORT,'pair tag failure'); END" + ) + conn.commit() + with self.assertRaises(sqlite3.IntegrityError): + self.db.merge_tags(["source"], "target") + self.assertEqual(self.tags(pair), ["source"]) + self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["source"]) + self.assertNotIn("target", self.db.get_all_tags()) + + +if __name__ == "__main__": + unittest.main()